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 #define DEBUG_MSG(x)
  33 /* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n");  */
  34
  35 typedef struct _GRealThreadPool GRealThreadPool;
  36
  37 struct _GRealThreadPool
  38 {
  39   GThreadPool pool;
  40   GAsyncQueue* queue;
  41   gint max_threads;
  42   gint num_threads;
  43   gboolean running;
  44   gboolean immediate;
  45   gboolean waiting;
  46   GCompareDataFunc sort_func;
  47   gpointer sort_user_data;
  48 };
  49
  50 /* The following is just an address to mark the wakeup order for a
  51  * thread, it could be any address (as long, as it isn't a valid
  52  * GThreadPool address) */
  53 static const gpointer wakeup_thread_marker = (gpointer) &g_thread_pool_new;
  54 static gint wakeup_thread_serial = 0;
  55
  56 /* Here all unused threads are waiting  */
  57 static GAsyncQueue *unused_thread_queue;
  58 static gint unused_threads = 0;
  59 static gint max_unused_threads = 0;
  60 static guint max_idle_time = 0;
  61
  62 static GMutex *inform_mutex = NULL;
  63 static GCond *inform_cond = NULL;
  64
  65
  66 static void     g_thread_pool_queue_push_unlocked (GRealThreadPool  *pool,
  67                                                    gpointer          data);
  68 static void     g_thread_pool_free_internal       (GRealThreadPool  *pool);
  69 static gpointer g_thread_pool_thread_proxy        (gpointer          data);
  70 static void     g_thread_pool_start_thread        (GRealThreadPool  *pool,
  71                                                    GError          **error);
  72 static void     g_thread_pool_wakeup_and_stop_all (GRealThreadPool  *pool);
  73
  74
  75 static void
  76 g_thread_pool_queue_push_unlocked (GRealThreadPool *pool,
  77                                    gpointer         data)
  78 {
  79   if (pool->sort_func)
  80     g_async_queue_push_sorted_unlocked (pool->queue,
  81                                         data,
  82                                         pool->sort_func,
  83                                         pool->sort_user_data);
  84   else
  85     g_async_queue_push_unlocked (pool->queue, data);
  86 }
  87
  88 static gpointer
  89 g_thread_pool_thread_proxy (gpointer data)
  90 {
  91   GRealThreadPool *pool = data;
  92   guint last_wakeup_thread_serial = 0;
  93
  94   DEBUG_MSG (("thread %p started for pool %p.",
  95               g_thread_self (), pool));
  96
  97   g_async_queue_lock (pool->queue);
  98   while (TRUE)
  99     {
 100       gpointer task = NULL;
 101
 102       if (pool->running || (!pool->immediate &&
 103                             g_async_queue_length_unlocked (pool->queue) > 0))
 104         {
 105           /* This thread pool is still active. */
 106           if (pool->num_threads > pool->max_threads &&
 107               pool->max_threads != -1)
 108             {
 109               /* This is a superfluous thread, so it goes to the
 110                * global pool. */
 111               DEBUG_MSG (("superfluous thread %p in pool %p.",
 112                           g_thread_self (), pool));
 113             }
 114           else if (pool->pool.exclusive)
 115             {
 116               /* Exclusive threads stay attached to the pool. */
 117               task = g_async_queue_pop_unlocked (pool->queue);
 118               DEBUG_MSG (("thread %p in exclusive pool %p waits for task "
 119                           "(%d running, %d unprocessed).",
 120                           g_thread_self (), pool, pool->num_threads,
 121                           g_async_queue_length_unlocked (pool->queue)));
 122             }
 123           else
 124             {
 125               /* A thread will wait for new tasks for at most 1/2
 126                * second before going to the global pool. */
 127               GTimeVal end_time;
 128               g_get_current_time (&end_time);
 129               g_time_val_add (&end_time, G_USEC_PER_SEC / 2); /* 1/2 second */
 130               DEBUG_MSG (("thread %p in pool %p waits 1/2 second for task "
 131                           "(%d running, %d unprocessed).",
 132                           g_thread_self (), pool, pool->num_threads,
 133                           g_async_queue_length_unlocked (pool->queue)));
 134               task = g_async_queue_timed_pop_unlocked (pool->queue, &end_time);
 135             }
 136         }
 137       else
 138         {
 139           /* This thread pool should is inactive, it will no longer
 140            * process tasks. */
 141           DEBUG_MSG (("pool %p not active, thread %p will go to global pool "
 142                       "(running: %s, immediate: %s, len: %d).",
 143                       pool, g_thread_self (),
 144                       pool->running ? "true" : "false",
 145                       pool->immediate ? "true" : "false",
 146                       g_async_queue_length_unlocked (pool->queue)));
 147         }
 148
 149       if (task)
 150         {
 151           if (pool->running || !pool->immediate)
 152             {
 153               /* A task was received and the thread pool is active, so
 154                * execute the function. */
 155               g_async_queue_unlock (pool->queue);
 156               DEBUG_MSG (("thread %p in pool %p calling func.",
 157                           g_thread_self (), pool));
 158               pool->pool.func (task, pool->pool.user_data);
 159               g_async_queue_lock (pool->queue);
 160             }
 161         }
 162       else
 163         {
 164           /* No task was received, so this thread goes to the global
 165            * pool. */
 166           gboolean free_pool = FALSE;
 167
 168           DEBUG_MSG (("thread %p leaving pool %p for global pool.",
 169                       g_thread_self (), pool));
 170           pool->num_threads--;
 171
 172           if (!pool->running)
 173             {
 174               if (!pool->waiting)
 175                 {
 176                   if (pool->num_threads == 0)
 177                     {
 178                       /* If the pool is not running and no other
 179                        * thread is waiting for this thread pool to
 180                        * finish and this is the last thread of this
 181                        * pool, free the pool. */
 182                       free_pool = TRUE;
 183                     }
 184                   else
 185                     {
 186                       /* If the pool is not running and no other
 187                        * thread is waiting for this thread pool to
 188                        * finish and this is not the last thread of
 189                        * this pool and there are no tasks left in the
 190                        * queue, wakeup the remaining threads. */
 191                       if (g_async_queue_length_unlocked (pool->queue) ==
 192                           - pool->num_threads)
 193                         g_thread_pool_wakeup_and_stop_all (pool);
 194                     }
 195                 }
 196               else if (pool->immediate ||
 197                        g_async_queue_length_unlocked (pool->queue) <= 0)
 198                 {
 199                   /* If the pool is not running and another thread is
 200                    * waiting for this thread pool to finish and there
 201                    * are either no tasks left or the pool shall stop
 202                    * immediatly, inform the waiting thread of a change
 203                    * of the thread pool state. */
 204                   g_cond_broadcast (inform_cond);
 205                 }
 206             }
 207           g_async_queue_unlock (pool->queue);
 208
 209           if (free_pool)
 210             g_thread_pool_free_internal (pool);
 211
 212           g_async_queue_lock (unused_thread_queue);
 213           unused_threads++;
 214
 215           do
 216             {
 217
 218               if ((unused_threads >= max_unused_threads &&
 219                    max_unused_threads != -1))
 220                 {
 221                   /* If this is a superflous thread, stop it. */
 222                   unused_threads--;
 223                   g_async_queue_unlock (unused_thread_queue);
 224                   DEBUG_MSG (("stopping thread %p.", g_thread_self ()));
 225                   return NULL;
 226                 }
 227
 228               if (max_idle_time > 0)
 229                 {
 230                   /* If a maximal idle time is given, wait for the
 231                    * given time. */
 232                   GTimeVal end_time;
 233                   g_get_current_time (&end_time);
 234                   g_time_val_add (&end_time, max_idle_time * 1000);
 235
 236                   DEBUG_MSG (("thread %p waiting in global pool for "
 237                               "%f seconds.",
 238                               g_thread_self (), max_idle_time / 1000.0));
 239
 240                   pool = g_async_queue_timed_pop_unlocked (unused_thread_queue,
 241                                                            &end_time);
 242                   if (!pool)
 243                     {
 244                       /* If no new task was received in the given
 245                        * time, stop this thread. */
 246                       unused_threads--;
 247                       g_async_queue_unlock (unused_thread_queue);
 248                       DEBUG_MSG (("stopping thread %p after max-idle-time.",
 249                                   g_thread_self ()));
 250                       /* Stop this thread */
 251                       return NULL;
 252                     }
 253                 }
 254               else
 255                 {
 256                   /* If no maximal idle time is given, wait
 257                    * indefinitly. */
 258                   DEBUG_MSG (("thread %p waiting in global pool.",
 259                               g_thread_self ()));
 260                   pool = g_async_queue_pop_unlocked (unused_thread_queue);
 261                 }
 262
 263               if (pool == wakeup_thread_marker)
 264                 {
 265                   if (last_wakeup_thread_serial == wakeup_thread_serial)
 266                     {
 267                       /* If this wakeup marker has been received for
 268                        * the second time, relay it. */
 269                       DEBUG_MSG (("thread %p relaying wakeup message to "
 270                                   "waiting thread with lower serial.",
 271                                   g_thread_self ()));
 272                       g_async_queue_push_unlocked (unused_thread_queue,
 273                                                    wakeup_thread_marker);
 274                     }
 275                   else
 276                     {
 277                       last_wakeup_thread_serial = wakeup_thread_serial;
 278                     }
 279
 280                   /* If a wakeup marker has been received, this thread
 281                    * will get out of the way for 100 microseconds to
 282                    * avoid receiving this marker again. */
 283                   g_async_queue_unlock (unused_thread_queue);
 284                   g_usleep (100);
 285                   g_async_queue_lock (unused_thread_queue);
 286                 }
 287
 288             } while (pool == wakeup_thread_marker);
 289
 290           unused_threads--;
 291           g_async_queue_unlock (unused_thread_queue);
 292
 293           g_async_queue_lock (pool->queue);
 294
 295           DEBUG_MSG (("thread %p entering pool %p from global pool.",
 296                       g_thread_self (), pool));
 297
 298           /* pool->num_threads++ is not done here, but in
 299            * g_thread_pool_start_thread to make the new started thread
 300            * known to the pool, before itself can do it. */
 301         }
 302     }
 303   return NULL;
 304 }
 305
 306 static void
 307 g_thread_pool_start_thread (GRealThreadPool  *pool,
 308                             GError          **error)
 309 {
 310   gboolean success = FALSE;
 311
 312   if (pool->num_threads >= pool->max_threads && pool->max_threads != -1)
 313     /* Enough threads are already running */
 314     return;
 315
 316   g_async_queue_lock (unused_thread_queue);
 317
 318   if (g_async_queue_length_unlocked (unused_thread_queue) < 0)
 319     {
 320       g_async_queue_push_unlocked (unused_thread_queue, pool);
 321       success = TRUE;
 322     }
 323
 324   g_async_queue_unlock (unused_thread_queue);
 325
 326   if (!success)
 327     {
 328       GError *local_error = NULL;
 329       /* No thread was found, we have to start a new one */
 330       g_thread_create (g_thread_pool_thread_proxy, pool, FALSE, &local_error);
 331
 332       if (local_error)
 333         {
 334           g_propagate_error (error, local_error);
 335           return;
 336         }
 337     }
 338
 339   /* See comment in g_thread_pool_thread_proxy as to why this is done
 340    * here and not there */
 341   pool->num_threads++;
 342 }
 343
 344 /**
 345  * g_thread_pool_new:
 346  * @func: a function to execute in the threads of the new thread pool
 347  * @user_data: user data that is handed over to @func every time it
 348  *   is called
 349  * @max_threads: the maximal number of threads to execute concurrently in
 350  *   the new thread pool, -1 means no limit
 351  * @exclusive: should this thread pool be exclusive?
 352  * @error: return location for error
 353  *
 354  * This function creates a new thread pool.
 355  *
 356  * Whenever you call g_thread_pool_push(), either a new thread is
 357  * created or an unused one is reused. At most @max_threads threads
 358  * are running concurrently for this thread pool. @max_threads = -1
 359  * allows unlimited threads to be created for this thread pool. The
 360  * newly created or reused thread now executes the function @func with
 361  * the two arguments. The first one is the parameter to
 362  * g_thread_pool_push() and the second one is @user_data.
 363  *
 364  * The parameter @exclusive determines, whether the thread pool owns
 365  * all threads exclusive or whether the threads are shared
 366  * globally. If @exclusive is %TRUE, @max_threads threads are started
 367  * immediately and they will run exclusively for this thread pool until
 368  * it is destroyed by g_thread_pool_free(). If @exclusive is %FALSE,
 369  * threads are created, when needed and shared between all
 370  * non-exclusive thread pools. This implies that @max_threads may not
 371  * be -1 for exclusive thread pools.
 372  *
 373  * @error can be %NULL to ignore errors, or non-%NULL to report
 374  * errors. An error can only occur when @exclusive is set to %TRUE and
 375  * not all @max_threads threads could be created.
 376  *
 377  * Return value: the new #GThreadPool
 378  **/
 379 GThreadPool*
 380 g_thread_pool_new (GFunc            func,
 381                    gpointer         user_data,
 382                    gint             max_threads,
 383                    gboolean         exclusive,
 384                    GError         **error)
 385 {
 386   GRealThreadPool *retval;
 387   G_LOCK_DEFINE_STATIC (init);
 388
 389   g_return_val_if_fail (func, NULL);
 390   g_return_val_if_fail (!exclusive || max_threads != -1, NULL);
 391   g_return_val_if_fail (max_threads >= -1, NULL);
 392   g_return_val_if_fail (g_thread_supported (), NULL);
 393
 394   retval = g_new (GRealThreadPool, 1);
 395
 396   retval->pool.func = func;
 397   retval->pool.user_data = user_data;
 398   retval->pool.exclusive = exclusive;
 399   retval->queue = g_async_queue_new ();
 400   retval->max_threads = max_threads;
 401   retval->num_threads = 0;
 402   retval->running = TRUE;
 403   retval->sort_func = NULL;
 404   retval->sort_user_data = NULL;
 405
 406   G_LOCK (init);
 407
 408   if (!inform_mutex)
 409     {
 410       inform_mutex = g_mutex_new ();
 411       inform_cond = g_cond_new ();
 412       unused_thread_queue = g_async_queue_new ();
 413     }
 414
 415   G_UNLOCK (init);
 416
 417   if (retval->pool.exclusive)
 418     {
 419       g_async_queue_lock (retval->queue);
 420
 421       while (retval->num_threads < retval->max_threads)
 422         {
 423           GError *local_error = NULL;
 424           g_thread_pool_start_thread (retval, &local_error);
 425           if (local_error)
 426             {
 427               g_propagate_error (error, local_error);
 428               break;
 429             }
 430         }
 431
 432       g_async_queue_unlock (retval->queue);
 433     }
 434
 435   return (GThreadPool*) retval;
 436 }
 437
 438 /**
 439  * g_thread_pool_push:
 440  * @pool: a #GThreadPool
 441  * @data: a new task for @pool
 442  * @error: return location for error
 443  *
 444  * Inserts @data into the list of tasks to be executed by @pool. When
 445  * the number of currently running threads is lower than the maximal
 446  * allowed number of threads, a new thread is started (or reused) with
 447  * the properties given to g_thread_pool_new (). Otherwise @data stays
 448  * in the queue until a thread in this pool finishes its previous task
 449  * and processes @data.
 450  *
 451  * @error can be %NULL to ignore errors, or non-%NULL to report
 452  * errors. An error can only occur when a new thread couldn't be
 453  * created. In that case @data is simply appended to the queue of work
 454  * to do.
 455  **/
 456 void
 457 g_thread_pool_push (GThreadPool     *pool,
 458                     gpointer         data,
 459                     GError         **error)
 460 {
 461   GRealThreadPool *real = (GRealThreadPool*) pool;
 462
 463   g_return_if_fail (real);
 464
 465   g_async_queue_lock (real->queue);
 466
 467   if (!real->running)
 468     {
 469       g_async_queue_unlock (real->queue);
 470       g_return_if_fail (real->running);
 471     }
 472
 473   if (g_async_queue_length_unlocked (real->queue) >= 0)
 474     /* No thread is waiting in the queue */
 475     g_thread_pool_start_thread (real, error);
 476
 477   g_thread_pool_queue_push_unlocked (real, data);
 478   g_async_queue_unlock (real->queue);
 479 }
 480
 481 /**
 482  * g_thread_pool_set_max_threads:
 483  * @pool: a #GThreadPool
 484  * @max_threads: a new maximal number of threads for @pool
 485  * @error: return location for error
 486  *
 487  * Sets the maximal allowed number of threads for @pool. A value of -1
 488  * means, that the maximal number of threads is unlimited.
 489  *
 490  * Setting @max_threads to 0 means stopping all work for @pool. It is
 491  * effectively frozen until @max_threads is set to a non-zero value
 492  * again.
 493  *
 494  * A thread is never terminated while calling @func, as supplied by
 495  * g_thread_pool_new (). Instead the maximal number of threads only
 496  * has effect for the allocation of new threads in g_thread_pool_push().
 497  * A new thread is allocated, whenever the number of currently
 498  * running threads in @pool is smaller than the maximal number.
 499  *
 500  * @error can be %NULL to ignore errors, or non-%NULL to report
 501  * errors. An error can only occur when a new thread couldn't be
 502  * created.
 503  **/
 504 void
 505 g_thread_pool_set_max_threads (GThreadPool     *pool,
 506                                gint             max_threads,
 507                                GError         **error)
 508 {
 509   GRealThreadPool *real = (GRealThreadPool*) pool;
 510   gint to_start;
 511
 512   g_return_if_fail (real);
 513   g_return_if_fail (real->running);
 514   g_return_if_fail (!real->pool.exclusive || max_threads != -1);
 515   g_return_if_fail (max_threads >= -1);
 516
 517   g_async_queue_lock (real->queue);
 518
 519   real->max_threads = max_threads;
 520
 521   if (pool->exclusive)
 522     to_start = real->max_threads - real->num_threads;
 523   else
 524     to_start = g_async_queue_length_unlocked (real->queue);
 525
 526   for ( ; to_start > 0; to_start--)
 527     {
 528       GError *local_error = NULL;
 529       g_thread_pool_start_thread (real, &local_error);
 530       if (local_error)
 531         {
 532           g_propagate_error (error, local_error);
 533           break;
 534         }
 535     }
 536
 537   g_async_queue_unlock (real->queue);
 538 }
 539
 540 /**
 541  * g_thread_pool_get_max_threads:
 542  * @pool: a #GThreadPool
 543  *
 544  * Returns the maximal number of threads for @pool.
 545  *
 546  * Return value: the maximal number of threads
 547  **/
 548 gint
 549 g_thread_pool_get_max_threads (GThreadPool     *pool)
 550 {
 551   GRealThreadPool *real = (GRealThreadPool*) pool;
 552   gint retval;
 553
 554   g_return_val_if_fail (real, 0);
 555   g_return_val_if_fail (real->running, 0);
 556
 557   g_async_queue_lock (real->queue);
 558
 559   retval = real->max_threads;
 560
 561   g_async_queue_unlock (real->queue);
 562
 563   return retval;
 564 }
 565
 566 /**
 567  * g_thread_pool_get_num_threads:
 568  * @pool: a #GThreadPool
 569  *
 570  * Returns the number of threads currently running in @pool.
 571  *
 572  * Return value: the number of threads currently running
 573  **/
 574 guint
 575 g_thread_pool_get_num_threads (GThreadPool     *pool)
 576 {
 577   GRealThreadPool *real = (GRealThreadPool*) pool;
 578   guint retval;
 579
 580   g_return_val_if_fail (real, 0);
 581   g_return_val_if_fail (real->running, 0);
 582
 583   g_async_queue_lock (real->queue);
 584
 585   retval = real->num_threads;
 586
 587   g_async_queue_unlock (real->queue);
 588
 589   return retval;
 590 }
 591
 592 /**
 593  * g_thread_pool_unprocessed:
 594  * @pool: a #GThreadPool
 595  *
 596  * Returns the number of tasks still unprocessed in @pool.
 597  *
 598  * Return value: the number of unprocessed tasks
 599  **/
 600 guint
 601 g_thread_pool_unprocessed (GThreadPool     *pool)
 602 {
 603   GRealThreadPool *real = (GRealThreadPool*) pool;
 604   gint unprocessed;
 605
 606   g_return_val_if_fail (real, 0);
 607   g_return_val_if_fail (real->running, 0);
 608
 609   unprocessed = g_async_queue_length (real->queue);
 610
 611   return MAX (unprocessed, 0);
 612 }
 613
 614 /**
 615  * g_thread_pool_free:
 616  * @pool: a #GThreadPool
 617  * @immediate: should @pool shut down immediately?
 618  * @wait: should the function wait for all tasks to be finished?
 619  *
 620  * Frees all resources allocated for @pool.
 621  *
 622  * If @immediate is %TRUE, no new task is processed for
 623  * @pool. Otherwise @pool is not freed before the last task is
 624  * processed. Note however, that no thread of this pool is
 625  * interrupted, while processing a task. Instead at least all still
 626  * running threads can finish their tasks before the @pool is freed.
 627  *
 628  * If @wait is %TRUE, the functions does not return before all tasks
 629  * to be processed (dependent on @immediate, whether all or only the
 630  * currently running) are ready. Otherwise the function returns immediately.
 631  *
 632  * After calling this function @pool must not be used anymore.
 633  **/
 634 void
 635 g_thread_pool_free (GThreadPool     *pool,
 636                     gboolean         immediate,
 637                     gboolean         wait)
 638 {
 639   GRealThreadPool *real = (GRealThreadPool*) pool;
 640
 641   g_return_if_fail (real);
 642   g_return_if_fail (real->running);
 643   /* It there's no thread allowed here, there is not much sense in
 644    * not stopping this pool immediately, when it's not empty */
 645   g_return_if_fail (immediate || real->max_threads != 0 ||
 646                     g_async_queue_length (real->queue) == 0);
 647
 648   g_async_queue_lock (real->queue);
 649
 650   real->running = FALSE;
 651   real->immediate = immediate;
 652   real->waiting = wait;
 653
 654   if (wait)
 655     {
 656       g_mutex_lock (inform_mutex);
 657       while (g_async_queue_length_unlocked (real->queue) != -real->num_threads &&
 658              !(immediate && real->num_threads == 0))
 659         {
 660           g_async_queue_unlock (real->queue);
 661           g_cond_wait (inform_cond, inform_mutex);
 662           g_async_queue_lock (real->queue);
 663         }
 664       g_mutex_unlock (inform_mutex);
 665     }
 666
 667   if (immediate ||
 668       g_async_queue_length_unlocked (real->queue) == -real->num_threads)
 669     {
 670       /* No thread is currently doing something (and nothing is left
 671        * to process in the queue) */
 672       if (real->num_threads == 0) /* No threads left, we clean up */
 673         {
 674           g_async_queue_unlock (real->queue);
 675           g_thread_pool_free_internal (real);
 676           return;
 677         }
 678
 679       g_thread_pool_wakeup_and_stop_all (real);
 680     }
 681
 682   real->waiting = FALSE; /* The last thread should cleanup the pool */
 683   g_async_queue_unlock (real->queue);
 684 }
 685
 686 static void
 687 g_thread_pool_free_internal (GRealThreadPool* pool)
 688 {
 689   g_return_if_fail (pool);
 690   g_return_if_fail (!pool->running);
 691   g_return_if_fail (pool->num_threads == 0);
 692
 693   g_async_queue_unref (pool->queue);
 694
 695   g_free (pool);
 696 }
 697
 698 static void
 699 g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool)
 700 {
 701   guint i;
 702
 703   g_return_if_fail (pool);
 704   g_return_if_fail (!pool->running);
 705   g_return_if_fail (pool->num_threads != 0);
 706   g_return_if_fail (g_async_queue_length_unlocked (pool->queue) ==
 707                     -pool->num_threads);
 708
 709   pool->immediate = TRUE;
 710   for (i = 0; i < pool->num_threads; i++)
 711     g_thread_pool_queue_push_unlocked (pool, GUINT_TO_POINTER (1));
 712 }
 713
 714 /**
 715  * g_thread_pool_set_max_unused_threads:
 716  * @max_threads: maximal number of unused threads
 717  *
 718  * Sets the maximal number of unused threads to @max_threads. If
 719  * @max_threads is -1, no limit is imposed on the number of unused
 720  * threads.
 721  **/
 722 void
 723 g_thread_pool_set_max_unused_threads (gint max_threads)
 724 {
 725   g_return_if_fail (max_threads >= -1);
 726
 727   g_async_queue_lock (unused_thread_queue);
 728
 729   max_unused_threads = max_threads;
 730
 731   if (max_unused_threads < unused_threads && max_unused_threads != -1)
 732     {
 733       guint i;
 734       wakeup_thread_serial++;
 735
 736       for (i = unused_threads - max_unused_threads; i > 0; i--)
 737         g_async_queue_push_unlocked (unused_thread_queue,
 738                                      wakeup_thread_marker);
 739     }
 740
 741   g_async_queue_unlock (unused_thread_queue);
 742 }
 743
 744 /**
 745  * g_thread_pool_get_max_unused_threads:
 746  *
 747  * Returns the maximal allowed number of unused threads.
 748  *
 749  * Return value: the maximal number of unused threads
 750  **/
 751 gint
 752 g_thread_pool_get_max_unused_threads (void)
 753 {
 754   gint retval;
 755
 756   g_async_queue_lock (unused_thread_queue);
 757   retval = max_unused_threads;
 758   g_async_queue_unlock (unused_thread_queue);
 759
 760   return retval;
 761 }
 762
 763 /**
 764  * g_thread_pool_get_num_unused_threads:
 765  *
 766  * Returns the number of currently unused threads.
 767  *
 768  * Return value: the number of currently unused threads
 769  **/
 770 guint
 771 g_thread_pool_get_num_unused_threads (void)
 772 {
 773   guint retval;
 774
 775   g_async_queue_lock (unused_thread_queue);
 776   retval = unused_threads;
 777   g_async_queue_unlock (unused_thread_queue);
 778
 779   return retval;
 780 }
 781
 782 /**
 783  * g_thread_pool_stop_unused_threads:
 784  *
 785  * Stops all currently unused threads. This does not change the
 786  * maximal number of unused threads. This function can be used to
 787  * regularly stop all unused threads e.g. from g_timeout_add().
 788  **/
 789 void
 790 g_thread_pool_stop_unused_threads (void)
 791 {
 792   guint oldval = g_thread_pool_get_max_unused_threads ();
 793   g_thread_pool_set_max_unused_threads (0);
 794   g_thread_pool_set_max_unused_threads (oldval);
 795 }
 796
 797 /**
 798  * g_thread_pool_set_sort_function:
 799  * @pool: a #GThreadPool
 800  * @func: the #GCompareDataFunc used to sort the list of tasks.
 801  *     This function is passed two tasks. It should return
 802  *     0 if the order in which they are handled does not matter,
 803  *     a negative value if the first task should be processed before
 804  *     the second or a positive value if the second task should be
 805  *     processed first.
 806  * @user_data: user data passed to @func.
 807  *
 808  * Sets the function used to sort the list of tasks. This allows the
 809  * tasks to be processed by a priority determined by @func, and not
 810  * just in the order in which they were added to the pool.
 811  *
 812  * Since: 2.10
 813  **/
 814 void
 815 g_thread_pool_set_sort_function (GThreadPool      *pool,
 816                                  GCompareDataFunc  func,
 817                                  gpointer          user_data)
 818 {
 819   GRealThreadPool *real = (GRealThreadPool*) pool;
 820
 821   g_return_if_fail (real);
 822   g_return_if_fail (real->running);
 823
 824   g_async_queue_lock (real->queue);
 825
 826   real->sort_func = func;
 827   real->sort_user_data = user_data;
 828
 829   if (func)
 830     g_async_queue_sort_unlocked (real->queue,
 831                                  real->sort_func,
 832                                  real->sort_user_data);
 833
 834   g_async_queue_unlock (real->queue);
 835 }
 836
 837 /**
 838  * g_thread_pool_set_max_idle_time:
 839  * @interval: the maximum @interval (1/1000ths of a second) a thread
 840  *     can be idle.
 841  *
 842  * This function will set the maximum @interval that a thread waiting
 843  * in the pool for new tasks can be idle for before being
 844  * stopped. This function is similar to calling
 845  * g_thread_pool_stop_unused_threads() on a regular timeout, except,
 846  * this is done on a per thread basis.
 847  *
 848  * By setting @interval to 0, idle threads will not be stopped.
 849  *
 850  * This function makes use of g_async_queue_timed_pop () using
 851  * @interval.
 852  *
 853  * Since: 2.10
 854  **/
 855 void
 856 g_thread_pool_set_max_idle_time (guint interval)
 857 {
 858   guint i;
 859
 860   g_async_queue_lock (unused_thread_queue);
 861   max_idle_time = interval;
 862   wakeup_thread_serial++;
 863
 864   for (i = 0; i < unused_threads; i++)
 865     g_async_queue_push_unlocked (unused_thread_queue, wakeup_thread_marker);
 866
 867   g_async_queue_unlock (unused_thread_queue);
 868 }
 869
 870 /**
 871  * g_thread_pool_get_max_idle_time:
 872  *
 873  * This function will return the maximum @interval that a thread will
 874  * wait in the thread pool for new tasks before being stopped.
 875  *
 876  * If this function returns 0, threads waiting in the thread pool for
 877  * new work are not stopped.
 878  *
 879  * Return value: the maximum @interval to wait for new tasks in the
 880  *     thread pool before stopping the thread (1/1000ths of a second).
 881  *
 882  * Since: 2.10
 883  **/
 884 guint
 885 g_thread_pool_get_max_idle_time (void)
 886 {
 887   guint retval;
 888
 889   g_async_queue_lock (unused_thread_queue);
 890   retval = max_idle_time;
 891   g_async_queue_unlock (unused_thread_queue);
 892
 893   return retval;
 894 }
 895
 896 #define __G_THREADPOOL_C__
 897 #include "galiasdef.c"