glib/gthreadpool.c

   1 /* GLIB - Library of useful routines for C programming
   2  * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
   3  *
   4  * GAsyncQueue: thread pool implementation.
   5  * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 /*
  24  * MT safe
  25  */
  26
  27 #include "config.h"
  28
  29 #include "glib.h"
  30 #include "galias.h"
  31
  32
  33 typedef struct _GRealThreadPool GRealThreadPool;
  34
  35 struct _GRealThreadPool
  36 {
  37   GThreadPool pool;
  38   GAsyncQueue* queue;
  39   gint max_threads;
  40   gint num_threads;
  41   gboolean running;
  42   gboolean immediate;
  43   gboolean waiting;
  44 };
  45
  46 /* The following is just an address to mark the stop order for a
  47  * thread, it could be any address (as long, as it isn't a valid
  48  * GThreadPool address) */
  49 static const gpointer stop_this_thread_marker = (gpointer) &g_thread_pool_new;
  50
  51 /* Here all unused threads are waiting  */
  52 static GAsyncQueue *unused_thread_queue;
  53 static gint unused_threads = 0;
  54 static gint max_unused_threads = 0;
  55 G_LOCK_DEFINE_STATIC (unused_threads);
  56
  57 static GMutex *inform_mutex = NULL;
  58 static GCond *inform_cond = NULL;
  59
  60 static void     g_thread_pool_free_internal (GRealThreadPool* pool);
  61 static gpointer g_thread_pool_thread_proxy (gpointer data);
  62 static void     g_thread_pool_start_thread (GRealThreadPool* pool,
  63                                             GError **error);
  64 static void     g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool);
  65
  66 #define g_thread_should_run(pool, len) \
  67   ((pool)->running || (!(pool)->immediate && (len) > 0))
  68
  69 static gpointer
  70 g_thread_pool_thread_proxy (gpointer data)
  71 {
  72   GRealThreadPool *pool = data;
  73   gboolean watcher = FALSE;
  74
  75   g_async_queue_lock (pool->queue);
  76   while (TRUE)
  77     {
  78       gpointer task;
  79       gboolean goto_global_pool = !pool->pool.exclusive;
  80       gint len = g_async_queue_length_unlocked (pool->queue);
  81
  82       g_print ("thread pool proxy loop\n");
  83       if (g_thread_should_run (pool, len))
  84         {
  85           if (watcher)
  86             {
  87               /* This thread is actually not needed here, but it waits
  88                * for some time anyway. If during that time a new
  89                * request arrives, this saves process
  90                * swicthes. Otherwise the thread will go to the global
  91                * pool afterwards */
  92               GTimeVal end_time;
  93               g_get_current_time (&end_time);
  94               g_time_val_add (&end_time, G_USEC_PER_SEC / 2); /* 1/2 second */
  95               task = g_async_queue_timed_pop_unlocked (pool->queue, &end_time);
  96             }
  97           else
  98             task = g_async_queue_pop_unlocked (pool->queue);
  99
 100           if (task)
 101             {
 102               watcher = FALSE;
 103               if (pool->num_threads > pool->max_threads &&
 104                   pool->max_threads != -1)
 105                 /* We are in fact a superfluous threads, so we go to
 106                  * the global pool and just hand the data further to
 107                  * the next one waiting in the queue */
 108                 {
 109                   g_async_queue_push_unlocked (pool->queue, task);
 110                   goto_global_pool = TRUE;
 111                 }
 112               else if (pool->running || !pool->immediate)
 113                 {
 114                   g_async_queue_unlock (pool->queue);
 115                   pool->pool.func (task, pool->pool.user_data);
 116                   g_async_queue_lock (pool->queue);
 117                 }
 118             }
 119           len = g_async_queue_length_unlocked (pool->queue);
 120         }
 121
 122       g_print ("queue len %d\n", len);
 123       if (!g_thread_should_run (pool, len))
 124         {
 125           g_print ("shouldn't run, go to global pool\n");
 126           g_cond_broadcast (inform_cond);
 127           goto_global_pool = TRUE;
 128         }
 129       else if (len > 0)
 130         {
 131           g_print ("should run, don't go to global pool\n");
 132           /* At this pool there are no threads waiting, but tasks are. */
 133           goto_global_pool = FALSE;
 134         }
 135       else if (len == 0 && !watcher && !pool->pool.exclusive)
 136         {
 137           /* Here neither threads nor tasks are queued and we didn't
 138            * just return from a timed wait. We now wait for a limited
 139            * time at this pool for new tasks to avoid costly context
 140            * switches. */
 141           g_print ("no threads, no tasks, wait for a while\n");
 142           goto_global_pool = FALSE;
 143           watcher = TRUE;
 144         }
 145
 146       if (goto_global_pool)
 147         {
 148           pool->num_threads--;
 149
 150           if (!pool->running && !pool->waiting)
 151             {
 152               if (pool->num_threads == 0)
 153                 {
 154                   g_async_queue_unlock (pool->queue);
 155                   g_thread_pool_free_internal (pool);
 156                 }
 157               else
 158                 {
 159                   if (len == - pool->num_threads)
 160                     g_thread_pool_wakeup_and_stop_all (pool);
 161
 162                   g_async_queue_unlock (pool->queue);
 163                 }
 164             }
 165           else
 166             g_async_queue_unlock (pool->queue);
 167
 168           g_async_queue_lock (unused_thread_queue);
 169
 170           G_LOCK (unused_threads);
 171           if ((unused_threads >= max_unused_threads &&
 172                max_unused_threads != -1))
 173             {
 174               G_UNLOCK (unused_threads);
 175               g_async_queue_unlock (unused_thread_queue);
 176               /* Stop this thread */
 177               return NULL;
 178             }
 179           unused_threads++;
 180           G_UNLOCK (unused_threads);
 181
 182           pool = g_async_queue_pop_unlocked (unused_thread_queue);
 183
 184           G_LOCK (unused_threads);
 185           unused_threads--;
 186           G_UNLOCK (unused_threads);
 187
 188           g_async_queue_unlock (unused_thread_queue);
 189
 190           if (pool == stop_this_thread_marker)
 191             /* Stop this thread */
 192             return NULL;
 193
 194           g_async_queue_lock (pool->queue);
 195
 196           /* pool->num_threads++ is not done here, but in
 197            * g_thread_pool_start_thread to make the new started thread
 198            * known to the pool, before itself can do it. */
 199         }
 200     }
 201   return NULL;
 202 }
 203
 204 static void
 205 g_thread_pool_start_thread (GRealThreadPool  *pool,
 206                             GError          **error)
 207 {
 208   gboolean success = FALSE;
 209
 210   if (pool->num_threads >= pool->max_threads && pool->max_threads != -1)
 211     /* Enough threads are already running */
 212     return;
 213
 214   g_async_queue_lock (unused_thread_queue);
 215
 216   if (g_async_queue_length_unlocked (unused_thread_queue) < 0)
 217     {
 218       g_async_queue_push_unlocked (unused_thread_queue, pool);
 219       success = TRUE;
 220     }
 221
 222   g_async_queue_unlock (unused_thread_queue);
 223
 224   if (!success)
 225     {
 226       GError *local_error = NULL;
 227       /* No thread was found, we have to start a new one */
 228       g_thread_create (g_thread_pool_thread_proxy, pool, FALSE, &local_error);
 229
 230       if (local_error)
 231         {
 232           g_propagate_error (error, local_error);
 233           return;
 234         }
 235     }
 236
 237   /* See comment in g_thread_pool_thread_proxy as to why this is done
 238    * here and not there */
 239   pool->num_threads++;
 240 }
 241
 242 /**
 243  * g_thread_pool_new:
 244  * @func: a function to execute in the threads of the new thread pool
 245  * @user_data: user data that is handed over to @func every time it
 246  *   is called
 247  * @max_threads: the maximal number of threads to execute concurrently in
 248  *   the new thread pool, -1 means no limit
 249  * @exclusive: should this thread pool be exclusive?
 250  * @error: return location for error
 251  *
 252  * This function creates a new thread pool.
 253  *
 254  * Whenever you call g_thread_pool_push(), either a new thread is
 255  * created or an unused one is reused. At most @max_threads threads
 256  * are running concurrently for this thread pool. @max_threads = -1
 257  * allows unlimited threads to be created for this thread pool. The
 258  * newly created or reused thread now executes the function @func with
 259  * the two arguments. The first one is the parameter to
 260  * g_thread_pool_push() and the second one is @user_data.
 261  *
 262  * The parameter @exclusive determines, whether the thread pool owns
 263  * all threads exclusive or whether the threads are shared
 264  * globally. If @exclusive is %TRUE, @max_threads threads are started
 265  * immediately and they will run exclusively for this thread pool until
 266  * it is destroyed by g_thread_pool_free(). If @exclusive is %FALSE,
 267  * threads are created, when needed and shared between all
 268  * non-exclusive thread pools. This implies that @max_threads may not
 269  * be -1 for exclusive thread pools.
 270  *
 271  * @error can be %NULL to ignore errors, or non-%NULL to report
 272  * errors. An error can only occur when @exclusive is set to %TRUE and
 273  * not all @max_threads threads could be created.
 274  *
 275  * Return value: the new #GThreadPool
 276  **/
 277 GThreadPool*
 278 g_thread_pool_new (GFunc            func,
 279                    gpointer         user_data,
 280                    gint             max_threads,
 281                    gboolean         exclusive,
 282                    GError         **error)
 283 {
 284   GRealThreadPool *retval;
 285   G_LOCK_DEFINE_STATIC (init);
 286
 287   g_return_val_if_fail (func, NULL);
 288   g_return_val_if_fail (!exclusive || max_threads != -1, NULL);
 289   g_return_val_if_fail (max_threads >= -1, NULL);
 290   g_return_val_if_fail (g_thread_supported (), NULL);
 291
 292   retval = g_new (GRealThreadPool, 1);
 293
 294   retval->pool.func = func;
 295   retval->pool.user_data = user_data;
 296   retval->pool.exclusive = exclusive;
 297   retval->queue = g_async_queue_new ();
 298   retval->max_threads = max_threads;
 299   retval->num_threads = 0;
 300   retval->running = TRUE;
 301
 302   G_LOCK (init);
 303
 304   if (!inform_mutex)
 305     {
 306       inform_mutex = g_mutex_new ();
 307       inform_cond = g_cond_new ();
 308       unused_thread_queue = g_async_queue_new ();
 309     }
 310
 311   G_UNLOCK (init);
 312
 313   if (retval->pool.exclusive)
 314     {
 315       g_async_queue_lock (retval->queue);
 316
 317       while (retval->num_threads < retval->max_threads)
 318         {
 319           GError *local_error = NULL;
 320           g_thread_pool_start_thread (retval, &local_error);
 321           if (local_error)
 322             {
 323               g_propagate_error (error, local_error);
 324               break;
 325             }
 326         }
 327
 328       g_async_queue_unlock (retval->queue);
 329     }
 330
 331   return (GThreadPool*) retval;
 332 }
 333
 334 /**
 335  * g_thread_pool_push:
 336  * @pool: a #GThreadPool
 337  * @data: a new task for @pool
 338  * @error: return location for error
 339  *
 340  * Inserts @data into the list of tasks to be executed by @pool. When
 341  * the number of currently running threads is lower than the maximal
 342  * allowed number of threads, a new thread is started (or reused) with
 343  * the properties given to g_thread_pool_new (). Otherwise @data stays
 344  * in the queue until a thread in this pool finishes its previous task
 345  * and processes @data.
 346  *
 347  * @error can be %NULL to ignore errors, or non-%NULL to report
 348  * errors. An error can only occur when a new thread couldn't be
 349  * created. In that case @data is simply appended to the queue of work
 350  * to do.
 351  **/
 352 void
 353 g_thread_pool_push (GThreadPool     *pool,
 354                     gpointer         data,
 355                     GError         **error)
 356 {
 357   GRealThreadPool *real = (GRealThreadPool*) pool;
 358
 359   g_return_if_fail (real);
 360
 361   g_async_queue_lock (real->queue);
 362
 363   if (!real->running)
 364     {
 365       g_async_queue_unlock (real->queue);
 366       g_return_if_fail (real->running);
 367     }
 368
 369   if (g_async_queue_length_unlocked (real->queue) >= 0)
 370     /* No thread is waiting in the queue */
 371     g_thread_pool_start_thread (real, error);
 372
 373   g_async_queue_push_unlocked (real->queue, data);
 374   g_async_queue_unlock (real->queue);
 375 }
 376
 377 /**
 378  * g_thread_pool_set_max_threads:
 379  * @pool: a #GThreadPool
 380  * @max_threads: a new maximal number of threads for @pool
 381  * @error: return location for error
 382  *
 383  * Sets the maximal allowed number of threads for @pool. A value of -1
 384  * means, that the maximal number of threads is unlimited.
 385  *
 386  * Setting @max_threads to 0 means stopping all work for @pool. It is
 387  * effectively frozen until @max_threads is set to a non-zero value
 388  * again.
 389  *
 390  * A thread is never terminated while calling @func, as supplied by
 391  * g_thread_pool_new (). Instead the maximal number of threads only
 392  * has effect for the allocation of new threads in g_thread_pool_push().
 393  * A new thread is allocated, whenever the number of currently
 394  * running threads in @pool is smaller than the maximal number.
 395  *
 396  * @error can be %NULL to ignore errors, or non-%NULL to report
 397  * errors. An error can only occur when a new thread couldn't be
 398  * created.
 399  **/
 400 void
 401 g_thread_pool_set_max_threads (GThreadPool     *pool,
 402                                gint             max_threads,
 403                                GError         **error)
 404 {
 405   GRealThreadPool *real = (GRealThreadPool*) pool;
 406   gint to_start;
 407
 408   g_return_if_fail (real);
 409   g_return_if_fail (real->running);
 410   g_return_if_fail (!real->pool.exclusive || max_threads != -1);
 411   g_return_if_fail (max_threads >= -1);
 412
 413   g_async_queue_lock (real->queue);
 414
 415   real->max_threads = max_threads;
 416
 417   if (pool->exclusive)
 418     to_start = real->max_threads - real->num_threads;
 419   else
 420     to_start = g_async_queue_length_unlocked (real->queue);
 421
 422   for ( ; to_start > 0; to_start--)
 423     {
 424       GError *local_error = NULL;
 425       g_thread_pool_start_thread (real, &local_error);
 426       if (local_error)
 427         {
 428           g_propagate_error (error, local_error);
 429           break;
 430         }
 431     }
 432
 433   g_async_queue_unlock (real->queue);
 434 }
 435
 436 /**
 437  * g_thread_pool_get_max_threads:
 438  * @pool: a #GThreadPool
 439  *
 440  * Returns the maximal number of threads for @pool.
 441  *
 442  * Return value: the maximal number of threads
 443  **/
 444 gint
 445 g_thread_pool_get_max_threads (GThreadPool     *pool)
 446 {
 447   GRealThreadPool *real = (GRealThreadPool*) pool;
 448   gint retval;
 449
 450   g_return_val_if_fail (real, 0);
 451   g_return_val_if_fail (real->running, 0);
 452
 453   g_async_queue_lock (real->queue);
 454
 455   retval = real->max_threads;
 456
 457   g_async_queue_unlock (real->queue);
 458
 459   return retval;
 460 }
 461
 462 /**
 463  * g_thread_pool_get_num_threads:
 464  * @pool: a #GThreadPool
 465  *
 466  * Returns the number of threads currently running in @pool.
 467  *
 468  * Return value: the number of threads currently running
 469  **/
 470 guint
 471 g_thread_pool_get_num_threads (GThreadPool     *pool)
 472 {
 473   GRealThreadPool *real = (GRealThreadPool*) pool;
 474   guint retval;
 475
 476   g_return_val_if_fail (real, 0);
 477   g_return_val_if_fail (real->running, 0);
 478
 479   g_async_queue_lock (real->queue);
 480
 481   retval = real->num_threads;
 482
 483   g_async_queue_unlock (real->queue);
 484
 485   return retval;
 486 }
 487
 488 /**
 489  * g_thread_pool_unprocessed:
 490  * @pool: a #GThreadPool
 491  *
 492  * Returns the number of tasks still unprocessed in @pool.
 493  *
 494  * Return value: the number of unprocessed tasks
 495  **/
 496 guint
 497 g_thread_pool_unprocessed (GThreadPool     *pool)
 498 {
 499   GRealThreadPool *real = (GRealThreadPool*) pool;
 500   gint unprocessed;
 501
 502   g_return_val_if_fail (real, 0);
 503   g_return_val_if_fail (real->running, 0);
 504
 505   unprocessed = g_async_queue_length (real->queue);
 506
 507   return MAX (unprocessed, 0);
 508 }
 509
 510 /**
 511  * g_thread_pool_free:
 512  * @pool: a #GThreadPool
 513  * @immediate: should @pool shut down immediately?
 514  * @wait: should the function wait for all tasks to be finished?
 515  *
 516  * Frees all resources allocated for @pool.
 517  *
 518  * If @immediate is %TRUE, no new task is processed for
 519  * @pool. Otherwise @pool is not freed before the last task is
 520  * processed. Note however, that no thread of this pool is
 521  * interrupted, while processing a task. Instead at least all still
 522  * running threads can finish their tasks before the @pool is freed.
 523  *
 524  * If @wait is %TRUE, the functions does not return before all tasks
 525  * to be processed (dependent on @immediate, whether all or only the
 526  * currently running) are ready. Otherwise the function returns immediately.
 527  *
 528  * After calling this function @pool must not be used anymore.
 529  **/
 530 void
 531 g_thread_pool_free (GThreadPool     *pool,
 532                     gboolean         immediate,
 533                     gboolean         wait)
 534 {
 535   GRealThreadPool *real = (GRealThreadPool*) pool;
 536
 537   g_return_if_fail (real);
 538   g_return_if_fail (real->running);
 539   /* It there's no thread allowed here, there is not much sense in
 540    * not stopping this pool immediately, when it's not empty */
 541   g_return_if_fail (immediate || real->max_threads != 0 ||
 542                     g_async_queue_length (real->queue) == 0);
 543
 544   g_async_queue_lock (real->queue);
 545
 546   real->running = FALSE;
 547   real->immediate = immediate;
 548   real->waiting = wait;
 549
 550   if (wait)
 551     {
 552       g_mutex_lock (inform_mutex);
 553       while (g_async_queue_length_unlocked (real->queue) != -real->num_threads &&
 554              !(immediate && real->num_threads == 0))
 555         {
 556           g_async_queue_unlock (real->queue);
 557           g_cond_wait (inform_cond, inform_mutex);
 558           g_async_queue_lock (real->queue);
 559         }
 560       g_mutex_unlock (inform_mutex);
 561     }
 562
 563   if (immediate ||
 564       g_async_queue_length_unlocked (real->queue) == -real->num_threads)
 565     {
 566       /* No thread is currently doing something (and nothing is left
 567        * to process in the queue) */
 568       if (real->num_threads == 0) /* No threads left, we clean up */
 569         {
 570           g_async_queue_unlock (real->queue);
 571           g_thread_pool_free_internal (real);
 572           return;
 573         }
 574
 575       g_thread_pool_wakeup_and_stop_all (real);
 576     }
 577
 578   real->waiting = FALSE; /* The last thread should cleanup the pool */
 579   g_async_queue_unlock (real->queue);
 580 }
 581
 582 static void
 583 g_thread_pool_free_internal (GRealThreadPool* pool)
 584 {
 585   g_return_if_fail (pool);
 586   g_return_if_fail (!pool->running);
 587   g_return_if_fail (pool->num_threads == 0);
 588
 589   g_async_queue_unref (pool->queue);
 590
 591   g_free (pool);
 592 }
 593
 594 static void
 595 g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool)
 596 {
 597   guint i;
 598
 599   g_return_if_fail (pool);
 600   g_return_if_fail (!pool->running);
 601   g_return_if_fail (pool->num_threads != 0);
 602   g_return_if_fail (g_async_queue_length_unlocked (pool->queue) ==
 603                     -pool->num_threads);
 604
 605   pool->immediate = TRUE;
 606   for (i = 0; i < pool->num_threads; i++)
 607     g_async_queue_push_unlocked (pool->queue, GUINT_TO_POINTER (1));
 608 }
 609
 610 /**
 611  * g_thread_pool_set_max_unused_threads:
 612  * @max_threads: maximal number of unused threads
 613  *
 614  * Sets the maximal number of unused threads to @max_threads. If
 615  * @max_threads is -1, no limit is imposed on the number of unused
 616  * threads.
 617  **/
 618 void
 619 g_thread_pool_set_max_unused_threads (gint max_threads)
 620 {
 621   g_return_if_fail (max_threads >= -1);
 622
 623   G_LOCK (unused_threads);
 624
 625   max_unused_threads = max_threads;
 626
 627   if (max_unused_threads < unused_threads && max_unused_threads != -1)
 628     {
 629       guint i;
 630
 631       g_async_queue_lock (unused_thread_queue);
 632       for (i = unused_threads - max_unused_threads; i > 0; i--)
 633         g_async_queue_push_unlocked (unused_thread_queue,
 634                                      stop_this_thread_marker);
 635       g_async_queue_unlock (unused_thread_queue);
 636     }
 637
 638   G_UNLOCK (unused_threads);
 639 }
 640
 641 /**
 642  * g_thread_pool_get_max_unused_threads:
 643  *
 644  * Returns the maximal allowed number of unused threads.
 645  *
 646  * Return value: the maximal number of unused threads
 647  **/
 648 gint
 649 g_thread_pool_get_max_unused_threads (void)
 650 {
 651   gint retval;
 652
 653   G_LOCK (unused_threads);
 654   retval = max_unused_threads;
 655   G_UNLOCK (unused_threads);
 656
 657   return retval;
 658 }
 659
 660 /**
 661  * g_thread_pool_get_num_unused_threads:
 662  *
 663  * Returns the number of currently unused threads.
 664  *
 665  * Return value: the number of currently unused threads
 666  **/
 667 guint g_thread_pool_get_num_unused_threads (void)
 668 {
 669   guint retval;
 670
 671   G_LOCK (unused_threads);
 672   retval = unused_threads;
 673   G_UNLOCK (unused_threads);
 674
 675   return retval;
 676 }
 677
 678 /**
 679  * g_thread_pool_stop_unused_threads:
 680  *
 681  * Stops all currently unused threads. This does not change the
 682  * maximal number of unused threads. This function can be used to
 683  * regularly stop all unused threads e.g. from g_timeout_add().
 684  **/
 685 void g_thread_pool_stop_unused_threads (void)
 686 {
 687   guint oldval = g_thread_pool_get_max_unused_threads ();
 688   g_thread_pool_set_max_unused_threads (0);
 689   g_thread_pool_set_max_unused_threads (oldval);
 690 }
 691
 692 #define __G_THREADPOOL_C__
 693 #include "galiasdef.c"