1 /* GLIB - Library of useful routines for C programming
2 * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald
4 * GAsyncQueue: thread pool implementation.
5 * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
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.
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.
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.
32 typedef struct _GRealThreadPool GRealThreadPool
;
34 struct _GRealThreadPool
45 /* The following is just an address to mark the stop order for a
46 * thread, it could be any address (as long, as it isn't a valid
47 * GThreadPool address) */
48 static const gpointer stop_this_thread_marker
= (gpointer
) &g_thread_pool_new
;
50 /* Here all unused threads are waiting */
51 static GAsyncQueue
*unused_thread_queue
;
52 static gint unused_threads
= 0;
53 static gint max_unused_threads
= 0;
54 G_LOCK_DEFINE_STATIC (unused_threads
);
56 static GMutex
*inform_mutex
= NULL
;
57 static GCond
*inform_cond
= NULL
;
59 static void g_thread_pool_free_internal (GRealThreadPool
* pool
);
60 static gpointer
g_thread_pool_thread_proxy (gpointer data
);
61 static void g_thread_pool_start_thread (GRealThreadPool
* pool
,
63 static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool
* pool
);
65 #define g_thread_should_run(pool, len) \
66 ((pool)->running || (!(pool)->immediate && (len) > 0))
69 g_thread_pool_thread_proxy (gpointer data
)
71 GRealThreadPool
*pool
= data
;
72 gboolean watcher
= FALSE
;
74 g_async_queue_lock (pool
->queue
);
78 gboolean goto_global_pool
= !pool
->pool
.exclusive
;
79 gint len
= g_async_queue_length_unlocked (pool
->queue
);
81 if (g_thread_should_run (pool
, len
))
85 /* This thread is actually not needed here, but it waits
86 * for some time anyway. If during that time a new
87 * request arrives, this saves process
88 * swicthes. Otherwise the thread will go to the global
91 g_get_current_time (&end_time
);
92 g_time_val_add (&end_time
, G_USEC_PER_SEC
/ 2); /* 1/2 second */
93 task
= g_async_queue_timed_pop_unlocked (pool
->queue
, &end_time
);
96 task
= g_async_queue_pop_unlocked (pool
->queue
);
101 if (pool
->num_threads
> pool
->max_threads
&&
102 pool
->max_threads
!= -1)
103 /* We are in fact a superfluous threads, so we go to
104 * the global pool and just hand the data further to
105 * the next one waiting in the queue */
107 g_async_queue_push_unlocked (pool
->queue
, task
);
108 goto_global_pool
= TRUE
;
110 else if (pool
->running
|| !pool
->immediate
)
112 g_async_queue_unlock (pool
->queue
);
113 pool
->pool
.func (task
, pool
->pool
.user_data
);
114 g_async_queue_lock (pool
->queue
);
117 len
= g_async_queue_length_unlocked (pool
->queue
);
120 if (!g_thread_should_run (pool
, len
))
122 g_cond_broadcast (inform_cond
);
123 goto_global_pool
= TRUE
;
127 /* At this pool there are no threads waiting, but tasks are. */
128 goto_global_pool
= FALSE
;
130 else if (len
== 0 && !watcher
&& !pool
->pool
.exclusive
)
132 /* Here neither threads nor tasks are queued and we didn't
133 * just return from a timed wait. We now wait for a limited
134 * time at this pool for new tasks to avoid costly context
136 goto_global_pool
= FALSE
;
140 if (goto_global_pool
)
144 if (!pool
->running
&& !pool
->waiting
)
146 if (pool
->num_threads
== 0)
148 g_async_queue_unlock (pool
->queue
);
149 g_thread_pool_free_internal (pool
);
153 if (len
== - pool
->num_threads
)
154 g_thread_pool_wakeup_and_stop_all (pool
);
156 g_async_queue_unlock (pool
->queue
);
160 g_async_queue_unlock (pool
->queue
);
162 g_async_queue_lock (unused_thread_queue
);
164 G_LOCK (unused_threads
);
165 if ((unused_threads
>= max_unused_threads
&&
166 max_unused_threads
!= -1))
168 G_UNLOCK (unused_threads
);
169 g_async_queue_unlock (unused_thread_queue
);
170 /* Stop this thread */
174 G_UNLOCK (unused_threads
);
176 pool
= g_async_queue_pop_unlocked (unused_thread_queue
);
178 G_LOCK (unused_threads
);
180 G_UNLOCK (unused_threads
);
182 g_async_queue_unlock (unused_thread_queue
);
184 if (pool
== stop_this_thread_marker
)
185 /* Stop this thread */
188 g_async_queue_lock (pool
->queue
);
190 /* pool->num_threads++ is not done here, but in
191 * g_thread_pool_start_thread to make the new started thread
192 * known to the pool, before itself can do it. */
199 g_thread_pool_start_thread (GRealThreadPool
*pool
,
202 gboolean success
= FALSE
;
204 if (pool
->num_threads
>= pool
->max_threads
&& pool
->max_threads
!= -1)
205 /* Enough threads are already running */
208 g_async_queue_lock (unused_thread_queue
);
210 if (g_async_queue_length_unlocked (unused_thread_queue
) < 0)
212 g_async_queue_push_unlocked (unused_thread_queue
, pool
);
216 g_async_queue_unlock (unused_thread_queue
);
220 GError
*local_error
= NULL
;
221 /* No thread was found, we have to start a new one */
222 g_thread_create (g_thread_pool_thread_proxy
, pool
, FALSE
, &local_error
);
226 g_propagate_error (error
, local_error
);
231 /* See comment in g_thread_pool_thread_proxy as to why this is done
232 * here and not there */
238 * @func: a function to execute in the threads of the new thread pool
239 * @user_data: user data that is handed over to @func every time it
241 * @max_threads: the maximal number of threads to execute concurrently in
242 * the new thread pool, -1 means no limit
243 * @exclusive: should this thread pool be exclusive?
244 * @error: return location for error
246 * This function creates a new thread pool.
248 * Whenever you call g_thread_pool_push(), either a new thread is
249 * created or an unused one is reused. At most @max_threads threads
250 * are running concurrently for this thread pool. @max_threads = -1
251 * allows unlimited threads to be created for this thread pool. The
252 * newly created or reused thread now executes the function @func with
253 * the two arguments. The first one is the parameter to
254 * g_thread_pool_push() and the second one is @user_data.
256 * The parameter @exclusive determines, whether the thread pool owns
257 * all threads exclusive or whether the threads are shared
258 * globally. If @exclusive is %TRUE, @max_threads threads are started
259 * immediately and they will run exclusively for this thread pool until
260 * it is destroyed by g_thread_pool_free(). If @exclusive is %FALSE,
261 * threads are created, when needed and shared between all
262 * non-exclusive thread pools. This implies that @max_threads may not
263 * be -1 for exclusive thread pools.
265 * @error can be %NULL to ignore errors, or non-%NULL to report
266 * errors. An error can only occur when @exclusive is set to %TRUE and
267 * not all @max_threads threads could be created.
269 * Return value: the new #GThreadPool
272 g_thread_pool_new (GFunc func
,
278 GRealThreadPool
*retval
;
279 G_LOCK_DEFINE_STATIC (init
);
281 g_return_val_if_fail (func
, NULL
);
282 g_return_val_if_fail (!exclusive
|| max_threads
!= -1, NULL
);
283 g_return_val_if_fail (max_threads
>= -1, NULL
);
284 g_return_val_if_fail (g_thread_supported (), NULL
);
286 retval
= g_new (GRealThreadPool
, 1);
288 retval
->pool
.func
= func
;
289 retval
->pool
.user_data
= user_data
;
290 retval
->pool
.exclusive
= exclusive
;
291 retval
->queue
= g_async_queue_new ();
292 retval
->max_threads
= max_threads
;
293 retval
->num_threads
= 0;
294 retval
->running
= TRUE
;
300 inform_mutex
= g_mutex_new ();
301 inform_cond
= g_cond_new ();
302 unused_thread_queue
= g_async_queue_new ();
307 if (retval
->pool
.exclusive
)
309 g_async_queue_lock (retval
->queue
);
311 while (retval
->num_threads
< retval
->max_threads
)
313 GError
*local_error
= NULL
;
314 g_thread_pool_start_thread (retval
, &local_error
);
317 g_propagate_error (error
, local_error
);
322 g_async_queue_unlock (retval
->queue
);
325 return (GThreadPool
*) retval
;
329 * g_thread_pool_push:
330 * @pool: a #GThreadPool
331 * @data: a new task for @pool
332 * @error: return location for error
334 * Inserts @data into the list of tasks to be executed by @pool. When
335 * the number of currently running threads is lower than the maximal
336 * allowed number of threads, a new thread is started (or reused) with
337 * the properties given to g_thread_pool_new (). Otherwise @data stays
338 * in the queue until a thread in this pool finishes its previous task
339 * and processes @data.
341 * @error can be %NULL to ignore errors, or non-%NULL to report
342 * errors. An error can only occur when a new thread couldn't be
343 * created. In that case @data is simply appended to the queue of work
347 g_thread_pool_push (GThreadPool
*pool
,
351 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
353 g_return_if_fail (real
);
355 g_async_queue_lock (real
->queue
);
359 g_async_queue_unlock (real
->queue
);
360 g_return_if_fail (real
->running
);
363 if (g_async_queue_length_unlocked (real
->queue
) >= 0)
364 /* No thread is waiting in the queue */
365 g_thread_pool_start_thread (real
, error
);
367 g_async_queue_push_unlocked (real
->queue
, data
);
368 g_async_queue_unlock (real
->queue
);
372 * g_thread_pool_set_max_threads:
373 * @pool: a #GThreadPool
374 * @max_threads: a new maximal number of threads for @pool
375 * @error: return location for error
377 * Sets the maximal allowed number of threads for @pool. A value of -1
378 * means, that the maximal number of threads is unlimited.
380 * Setting @max_threads to 0 means stopping all work for @pool. It is
381 * effectively frozen until @max_threads is set to a non-zero value
384 * A thread is never terminated while calling @func, as supplied by
385 * g_thread_pool_new (). Instead the maximal number of threads only
386 * has effect for the allocation of new threads in g_thread_pool_push().
387 * A new thread is allocated, whenever the number of currently
388 * running threads in @pool is smaller than the maximal number.
390 * @error can be %NULL to ignore errors, or non-%NULL to report
391 * errors. An error can only occur when a new thread couldn't be
395 g_thread_pool_set_max_threads (GThreadPool
*pool
,
399 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
402 g_return_if_fail (real
);
403 g_return_if_fail (real
->running
);
404 g_return_if_fail (!real
->pool
.exclusive
|| max_threads
!= -1);
405 g_return_if_fail (max_threads
>= -1);
407 g_async_queue_lock (real
->queue
);
409 real
->max_threads
= max_threads
;
412 to_start
= real
->max_threads
- real
->num_threads
;
414 to_start
= g_async_queue_length_unlocked (real
->queue
);
416 for ( ; to_start
> 0; to_start
--)
418 GError
*local_error
= NULL
;
419 g_thread_pool_start_thread (real
, &local_error
);
422 g_propagate_error (error
, local_error
);
427 g_async_queue_unlock (real
->queue
);
431 * g_thread_pool_get_max_threads:
432 * @pool: a #GThreadPool
434 * Returns the maximal number of threads for @pool.
436 * Return value: the maximal number of threads
439 g_thread_pool_get_max_threads (GThreadPool
*pool
)
441 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
444 g_return_val_if_fail (real
, 0);
445 g_return_val_if_fail (real
->running
, 0);
447 g_async_queue_lock (real
->queue
);
449 retval
= real
->max_threads
;
451 g_async_queue_unlock (real
->queue
);
457 * g_thread_pool_get_num_threads:
458 * @pool: a #GThreadPool
460 * Returns the number of threads currently running in @pool.
462 * Return value: the number of threads currently running
465 g_thread_pool_get_num_threads (GThreadPool
*pool
)
467 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
470 g_return_val_if_fail (real
, 0);
471 g_return_val_if_fail (real
->running
, 0);
473 g_async_queue_lock (real
->queue
);
475 retval
= real
->num_threads
;
477 g_async_queue_unlock (real
->queue
);
483 * g_thread_pool_unprocessed:
484 * @pool: a #GThreadPool
486 * Returns the number of tasks still unprocessed in @pool.
488 * Return value: the number of unprocessed tasks
491 g_thread_pool_unprocessed (GThreadPool
*pool
)
493 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
496 g_return_val_if_fail (real
, 0);
497 g_return_val_if_fail (real
->running
, 0);
499 unprocessed
= g_async_queue_length (real
->queue
);
501 return MAX (unprocessed
, 0);
505 * g_thread_pool_free:
506 * @pool: a #GThreadPool
507 * @immediate: should @pool shut down immediately?
508 * @wait: should the function wait for all tasks to be finished?
510 * Frees all resources allocated for @pool.
512 * If @immediate is %TRUE, no new task is processed for
513 * @pool. Otherwise @pool is not freed before the last task is
514 * processed. Note however, that no thread of this pool is
515 * interrupted, while processing a task. Instead at least all still
516 * running threads can finish their tasks before the @pool is freed.
518 * If @wait is %TRUE, the functions does not return before all tasks
519 * to be processed (dependent on @immediate, whether all or only the
520 * currently running) are ready. Otherwise the function returns immediately.
522 * After calling this function @pool must not be used anymore.
525 g_thread_pool_free (GThreadPool
*pool
,
529 GRealThreadPool
*real
= (GRealThreadPool
*) pool
;
531 g_return_if_fail (real
);
532 g_return_if_fail (real
->running
);
533 /* It there's no thread allowed here, there is not much sense in
534 * not stopping this pool immediately, when it's not empty */
535 g_return_if_fail (immediate
|| real
->max_threads
!= 0 ||
536 g_async_queue_length (real
->queue
) == 0);
538 g_async_queue_lock (real
->queue
);
540 real
->running
= FALSE
;
541 real
->immediate
= immediate
;
542 real
->waiting
= wait
;
546 g_mutex_lock (inform_mutex
);
547 while (g_async_queue_length_unlocked (real
->queue
) != -real
->num_threads
)
549 g_async_queue_unlock (real
->queue
);
550 g_cond_wait (inform_cond
, inform_mutex
);
551 g_async_queue_lock (real
->queue
);
553 g_mutex_unlock (inform_mutex
);
556 if (g_async_queue_length_unlocked (real
->queue
) == -real
->num_threads
)
558 /* No thread is currently doing something (and nothing is left
559 * to process in the queue) */
560 if (real
->num_threads
== 0) /* No threads left, we clean up */
562 g_async_queue_unlock (real
->queue
);
563 g_thread_pool_free_internal (real
);
567 g_thread_pool_wakeup_and_stop_all (real
);
570 real
->waiting
= FALSE
; /* The last thread should cleanup the pool */
571 g_async_queue_unlock (real
->queue
);
575 g_thread_pool_free_internal (GRealThreadPool
* pool
)
577 g_return_if_fail (pool
);
578 g_return_if_fail (!pool
->running
);
579 g_return_if_fail (pool
->num_threads
== 0);
581 g_async_queue_unref (pool
->queue
);
587 g_thread_pool_wakeup_and_stop_all (GRealThreadPool
* pool
)
591 g_return_if_fail (pool
);
592 g_return_if_fail (!pool
->running
);
593 g_return_if_fail (pool
->num_threads
!= 0);
594 g_return_if_fail (g_async_queue_length_unlocked (pool
->queue
) ==
597 pool
->immediate
= TRUE
;
598 for (i
= 0; i
< pool
->num_threads
; i
++)
599 g_async_queue_push_unlocked (pool
->queue
, GUINT_TO_POINTER (1));
603 * g_thread_pool_set_max_unused_threads:
604 * @max_threads: maximal number of unused threads
606 * Sets the maximal number of unused threads to @max_threads. If
607 * @max_threads is -1, no limit is imposed on the number of unused
611 g_thread_pool_set_max_unused_threads (gint max_threads
)
613 g_return_if_fail (max_threads
>= -1);
615 G_LOCK (unused_threads
);
617 max_unused_threads
= max_threads
;
619 if (max_unused_threads
< unused_threads
&& max_unused_threads
!= -1)
623 g_async_queue_lock (unused_thread_queue
);
624 for (i
= unused_threads
- max_unused_threads
; i
> 0; i
--)
625 g_async_queue_push_unlocked (unused_thread_queue
,
626 stop_this_thread_marker
);
627 g_async_queue_unlock (unused_thread_queue
);
630 G_UNLOCK (unused_threads
);
634 * g_thread_pool_get_max_unused_threads:
636 * Returns the maximal allowed number of unused threads.
638 * Return value: the maximal number of unused threads
641 g_thread_pool_get_max_unused_threads (void)
645 G_LOCK (unused_threads
);
646 retval
= max_unused_threads
;
647 G_UNLOCK (unused_threads
);
653 * g_thread_pool_get_num_unused_threads:
655 * Returns the number of currently unused threads.
657 * Return value: the number of currently unused threads
659 guint
g_thread_pool_get_num_unused_threads (void)
663 G_LOCK (unused_threads
);
664 retval
= unused_threads
;
665 G_UNLOCK (unused_threads
);
671 * g_thread_pool_stop_unused_threads:
673 * Stops all currently unused threads. This does not change the
674 * maximal number of unused threads. This function can be used to
675 * regularly stop all unused threads e.g. from g_timeout_add().
677 void g_thread_pool_stop_unused_threads (void)
679 guint oldval
= g_thread_pool_get_max_unused_threads ();
680 g_thread_pool_set_max_unused_threads (0);
681 g_thread_pool_set_max_unused_threads (oldval
);