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