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