diff options
| -rw-r--r-- | ChangeLog | 17 | ||||
| -rw-r--r-- | ChangeLog.pre-2-12 | 17 | ||||
| -rw-r--r-- | glib/gasyncqueue.c | 13 | ||||
| -rw-r--r-- | glib/gasyncqueue.h | 27 | ||||
| -rw-r--r-- | glib/gthreadpool.c | 534 | ||||
| -rw-r--r-- | tests/threadpool-test.c | 136 |
6 files changed, 468 insertions, 276 deletions
@@ -1,3 +1,20 @@ +2006-04-07 Martyn Russell <martyn@imendio.com> + + * glib/gasyncqueue.[ch]: Added private API + _g_async_queue_get_mutex so that g_thread_pool_free() can use the + async queue mutex. + + * glib/gthreadpool.c: Make sure + g_thread_pool_stop_unused_threads() actually stops unused threads + and global limits (like max idle time and max unused threads) can + be set without creating a thread pool first. Fixed #335215 (patch + from Chris Wilson). + + * tests/threadpool-test.c: Added two new tests, tests setting + global limits before creating a thread pool. The second test + makes sure unused threads are actually stopped when using the + g_thread_pool_stop_unused_threads(). + 2006-04-05 Matthias Clasen <mclasen@redhat.com> * glib/gnulib/vasnprintf.c (vasnprintf): Make diff --git a/ChangeLog.pre-2-12 b/ChangeLog.pre-2-12 index 1506f6d1d..7b62e8857 100644 --- a/ChangeLog.pre-2-12 +++ b/ChangeLog.pre-2-12 @@ -1,3 +1,20 @@ +2006-04-07 Martyn Russell <martyn@imendio.com> + + * glib/gasyncqueue.[ch]: Added private API + _g_async_queue_get_mutex so that g_thread_pool_free() can use the + async queue mutex. + + * glib/gthreadpool.c: Make sure + g_thread_pool_stop_unused_threads() actually stops unused threads + and global limits (like max idle time and max unused threads) can + be set without creating a thread pool first. Fixed #335215 (patch + from Chris Wilson). + + * tests/threadpool-test.c: Added two new tests, tests setting + global limits before creating a thread pool. The second test + makes sure unused threads are actually stopped when using the + g_thread_pool_stop_unused_threads(). + 2006-04-05 Matthias Clasen <mclasen@redhat.com> * glib/gnulib/vasnprintf.c (vasnprintf): Make diff --git a/glib/gasyncqueue.c b/glib/gasyncqueue.c index 8d6f3f295..646efd261 100644 --- a/glib/gasyncqueue.c +++ b/glib/gasyncqueue.c @@ -625,5 +625,18 @@ g_async_queue_sort_unlocked (GAsyncQueue *queue, &sd); } +/* + * Private API + */ + +GMutex* +_g_async_queue_get_mutex (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL); + + return queue->mutex; +} + #define __G_ASYNCQUEUE_C__ #include "galiasdef.c" diff --git a/glib/gasyncqueue.h b/glib/gasyncqueue.h index ce11e15ad..413093f3c 100644 --- a/glib/gasyncqueue.h +++ b/glib/gasyncqueue.h @@ -31,13 +31,12 @@ G_BEGIN_DECLS -typedef struct _GAsyncQueue GAsyncQueue; +typedef struct _GAsyncQueue GAsyncQueue; -/* Asyncronous Queues, can be used to communicate between threads - */ +/* Asyncronous Queues, can be used to communicate between threads */ /* Get a new GAsyncQueue with the ref_count 1 */ -GAsyncQueue* g_async_queue_new (void); +GAsyncQueue* g_async_queue_new (void); /* Lock and unlock a GAsyncQueue. All functions lock the queue for * themselves, but in certain cirumstances you want to hold the lock longer, @@ -46,19 +45,14 @@ GAsyncQueue* g_async_queue_new (void); void g_async_queue_lock (GAsyncQueue *queue); void g_async_queue_unlock (GAsyncQueue *queue); - - /* Ref and unref the GAsyncQueue. */ GAsyncQueue* g_async_queue_ref (GAsyncQueue *queue); void g_async_queue_unref (GAsyncQueue *queue); - #ifndef G_DISABLE_DEPRECATED /* You don't have to hold the lock for calling *_ref and *_unref anymore. */ void g_async_queue_ref_unlocked (GAsyncQueue *queue); void g_async_queue_unref_and_unlock (GAsyncQueue *queue); - - #endif /* !G_DISABLE_DEPRECATED */ /* Push data into the async queue. Must not be NULL. */ @@ -77,12 +71,11 @@ void g_async_queue_push_sorted_unlocked (GAsyncQueue *queue, gpointer user_data); /* Pop data from the async queue. When no data is there, the thread is blocked - * until data arrives. */ + * until data arrives. + */ gpointer g_async_queue_pop (GAsyncQueue *queue); gpointer g_async_queue_pop_unlocked (GAsyncQueue *queue); - - /* Try to pop data. NULL is returned in case of empty queue. */ gpointer g_async_queue_try_pop (GAsyncQueue *queue); gpointer g_async_queue_try_pop_unlocked (GAsyncQueue *queue); @@ -90,20 +83,20 @@ gpointer g_async_queue_try_pop_unlocked (GAsyncQueue *queue); /* Wait for data until at maximum until end_time is reached. NULL is returned - * in case of empty queue. */ + * in case of empty queue. + */ gpointer g_async_queue_timed_pop (GAsyncQueue *queue, GTimeVal *end_time); gpointer g_async_queue_timed_pop_unlocked (GAsyncQueue *queue, GTimeVal *end_time); - - /* Return the length of the queue. Negative values mean that threads * are waiting, positve values mean that there are entries in the * queue. Actually this function returns the length of the queue minus * the number of waiting threads, g_async_queue_length == 0 could also * mean 'n' entries in the queue and 'n' thread waiting. Such can - * happen due to locking of the queue or due to scheduling. */ + * happen due to locking of the queue or due to scheduling. + */ gint g_async_queue_length (GAsyncQueue *queue); gint g_async_queue_length_unlocked (GAsyncQueue *queue); void g_async_queue_sort (GAsyncQueue *queue, @@ -113,6 +106,8 @@ void g_async_queue_sort_unlocked (GAsyncQueue *queue, GCompareDataFunc func, gpointer user_data); +/* Private API */ +GMutex* _g_async_queue_get_mutex (GAsyncQueue *queue); G_END_DECLS diff --git a/glib/gthreadpool.c b/glib/gthreadpool.c index 685a314b4..b136405e0 100644 --- a/glib/gthreadpool.c +++ b/glib/gthreadpool.c @@ -29,8 +29,8 @@ #include "glib.h" #include "galias.h" -#define DEBUG_MSG(x) -/* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n"); */ +#define DEBUG_MSG(x) +/* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n"); */ typedef struct _GRealThreadPool GRealThreadPool; @@ -38,6 +38,7 @@ struct _GRealThreadPool { GThreadPool pool; GAsyncQueue* queue; + GCond* cond; gint max_threads; gint num_threads; gboolean running; @@ -50,27 +51,25 @@ struct _GRealThreadPool /* The following is just an address to mark the wakeup order for a * thread, it could be any address (as long, as it isn't a valid * GThreadPool address) */ -static const gpointer wakeup_thread_marker = (gpointer) &g_thread_pool_new; +static gconstpointer const wakeup_thread_marker = (gconstpointer) &g_thread_pool_new; static gint wakeup_thread_serial = 0; /* Here all unused threads are waiting */ -static GAsyncQueue *unused_thread_queue; +static GAsyncQueue *unused_thread_queue = NULL; static gint unused_threads = 0; static gint max_unused_threads = 0; +static gint kill_unused_threads = 0; static guint max_idle_time = 0; -static GMutex *inform_mutex = NULL; -static GCond *inform_cond = NULL; - - -static void g_thread_pool_queue_push_unlocked (GRealThreadPool *pool, - gpointer data); -static void g_thread_pool_free_internal (GRealThreadPool *pool); -static gpointer g_thread_pool_thread_proxy (gpointer data); -static void g_thread_pool_start_thread (GRealThreadPool *pool, - GError **error); -static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool *pool); - +static void g_thread_pool_queue_push_unlocked (GRealThreadPool *pool, + gpointer data); +static void g_thread_pool_free_internal (GRealThreadPool *pool); +static gpointer g_thread_pool_thread_proxy (gpointer data); +static void g_thread_pool_start_thread (GRealThreadPool *pool, + GError **error); +static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool *pool); +static GRealThreadPool* g_thread_pool_wait_for_new_pool (void); +static gpointer g_thread_pool_wait_for_new_task (GRealThreadPool *pool); static void g_thread_pool_queue_push_unlocked (GRealThreadPool *pool, @@ -85,73 +84,181 @@ g_thread_pool_queue_push_unlocked (GRealThreadPool *pool, g_async_queue_push_unlocked (pool->queue, data); } -static gpointer -g_thread_pool_thread_proxy (gpointer data) +static GRealThreadPool* +g_thread_pool_wait_for_new_pool (void) { - GRealThreadPool *pool = data; - guint last_wakeup_thread_serial = 0; + GRealThreadPool *pool; + gint local_wakeup_thread_serial; + guint local_max_unused_threads; + gint local_max_idle_time; + gint last_wakeup_thread_serial; + gboolean have_relayed_thread_marker = FALSE; - DEBUG_MSG (("thread %p started for pool %p.", - g_thread_self (), pool)); + local_max_unused_threads = g_atomic_int_get (&max_unused_threads); + local_max_idle_time = g_atomic_int_get (&max_idle_time); + last_wakeup_thread_serial = g_atomic_int_get (&wakeup_thread_serial); - g_async_queue_lock (pool->queue); - while (TRUE) + g_atomic_int_inc (&unused_threads); + + do { - gpointer task = NULL; + if (g_atomic_int_get (&unused_threads) >= local_max_unused_threads) + { + /* If this is a superfluous thread, stop it. */ + pool = NULL; + } + else if (local_max_idle_time > 0) + { + /* If a maximal idle time is given, wait for the given time. */ + GTimeVal end_time; + + g_get_current_time (&end_time); + g_time_val_add (&end_time, local_max_idle_time * 1000); + + DEBUG_MSG (("thread %p waiting in global pool for %f seconds.", + g_thread_self (), local_max_idle_time / 1000.0)); - if (pool->running || (!pool->immediate && - g_async_queue_length_unlocked (pool->queue) > 0)) + pool = g_async_queue_timed_pop (unused_thread_queue, &end_time); + } + else { - /* This thread pool is still active. */ - if (pool->num_threads > pool->max_threads && - pool->max_threads != -1) - { - /* This is a superfluous thread, so it goes to the - * global pool. */ - DEBUG_MSG (("superfluous thread %p in pool %p.", - g_thread_self (), pool)); - } - else if (pool->pool.exclusive) + /* If no maximal idle time is given, wait indefinitely. */ + DEBUG_MSG (("thread %p waiting in global pool.", + g_thread_self ())); + pool = g_async_queue_pop (unused_thread_queue); + } + + if (pool == wakeup_thread_marker) + { + local_wakeup_thread_serial = g_atomic_int_get (&wakeup_thread_serial); + if (last_wakeup_thread_serial == local_wakeup_thread_serial) { - /* Exclusive threads stay attached to the pool. */ - task = g_async_queue_pop_unlocked (pool->queue); - DEBUG_MSG (("thread %p in exclusive pool %p waits for task " - "(%d running, %d unprocessed).", - g_thread_self (), pool, pool->num_threads, - g_async_queue_length_unlocked (pool->queue))); + if (!have_relayed_thread_marker) + { + /* If this wakeup marker has been received for + * the second time, relay it. + */ + DEBUG_MSG (("thread %p relaying wakeup message to " + "waiting thread with lower serial.", + g_thread_self ())); + + g_async_queue_push (unused_thread_queue, wakeup_thread_marker); + have_relayed_thread_marker = TRUE; + + /* If a wakeup marker has been relayed, this thread + * will get out of the way for 100 microseconds to + * avoid receiving this marker again. */ + g_usleep (100); + } } else { - /* A thread will wait for new tasks for at most 1/2 - * second before going to the global pool. */ - GTimeVal end_time; - g_get_current_time (&end_time); - g_time_val_add (&end_time, G_USEC_PER_SEC / 2); /* 1/2 second */ - DEBUG_MSG (("thread %p in pool %p waits 1/2 second for task " - "(%d running, %d unprocessed).", - g_thread_self (), pool, pool->num_threads, - g_async_queue_length_unlocked (pool->queue))); - task = g_async_queue_timed_pop_unlocked (pool->queue, &end_time); + if (g_atomic_int_exchange_and_add (&kill_unused_threads, -1) > 0) + { + pool = NULL; + break; + } + + DEBUG_MSG (("thread %p updating to new limits.", + g_thread_self ())); + + local_max_unused_threads = g_atomic_int_get (&max_unused_threads); + local_max_idle_time = g_atomic_int_get (&max_idle_time); + last_wakeup_thread_serial = local_wakeup_thread_serial; + + have_relayed_thread_marker = FALSE; } } + } + while (pool == wakeup_thread_marker); + + g_atomic_int_add (&unused_threads, -1); + + return pool; +} + +static gpointer +g_thread_pool_wait_for_new_task (GRealThreadPool *pool) +{ + gpointer task = NULL; + + if (pool->running || (!pool->immediate && + g_async_queue_length_unlocked (pool->queue) > 0)) + { + /* This thread pool is still active. */ + if (pool->num_threads > pool->max_threads && pool->max_threads != -1) + { + /* This is a superfluous thread, so it goes to the global pool. */ + DEBUG_MSG (("superfluous thread %p in pool %p.", + g_thread_self (), pool)); + } + else if (pool->pool.exclusive) + { + /* Exclusive threads stay attached to the pool. */ + task = g_async_queue_pop_unlocked (pool->queue); + + DEBUG_MSG (("thread %p in exclusive pool %p waits for task " + "(%d running, %d unprocessed).", + g_thread_self (), pool, pool->num_threads, + g_async_queue_length_unlocked (pool->queue))); + } else { - /* This thread pool should is inactive, it will no longer - * process tasks. */ - DEBUG_MSG (("pool %p not active, thread %p will go to global pool " - "(running: %s, immediate: %s, len: %d).", - pool, g_thread_self (), - pool->running ? "true" : "false", - pool->immediate ? "true" : "false", + /* A thread will wait for new tasks for at most 1/2 + * second before going to the global pool. + */ + GTimeVal end_time; + + g_get_current_time (&end_time); + g_time_val_add (&end_time, G_USEC_PER_SEC / 2); /* 1/2 second */ + + DEBUG_MSG (("thread %p in pool %p waits for up to a 1/2 second for task " + "(%d running, %d unprocessed).", + g_thread_self (), pool, pool->num_threads, g_async_queue_length_unlocked (pool->queue))); + + task = g_async_queue_timed_pop_unlocked (pool->queue, &end_time); } - + } + else + { + /* This thread pool is inactive, it will no longer process tasks. */ + DEBUG_MSG (("pool %p not active, thread %p will go to global pool " + "(running: %s, immediate: %s, len: %d).", + pool, g_thread_self (), + pool->running ? "true" : "false", + pool->immediate ? "true" : "false", + g_async_queue_length_unlocked (pool->queue))); + } + + return task; +} + + +static gpointer +g_thread_pool_thread_proxy (gpointer data) +{ + GRealThreadPool *pool; + + pool = data; + + DEBUG_MSG (("thread %p started for pool %p.", + g_thread_self (), pool)); + + g_async_queue_lock (pool->queue); + + while (TRUE) + { + gpointer task; + + task = g_thread_pool_wait_for_new_task (pool); if (task) { if (pool->running || !pool->immediate) { /* A task was received and the thread pool is active, so - * execute the function. */ + * execute the function. + */ g_async_queue_unlock (pool->queue); DEBUG_MSG (("thread %p in pool %p calling func.", g_thread_self (), pool)); @@ -162,7 +269,8 @@ g_thread_pool_thread_proxy (gpointer data) else { /* No task was received, so this thread goes to the global - * pool. */ + * pool. + */ gboolean free_pool = FALSE; DEBUG_MSG (("thread %p leaving pool %p for global pool.", @@ -178,7 +286,8 @@ g_thread_pool_thread_proxy (gpointer data) /* If the pool is not running and no other * thread is waiting for this thread pool to * finish and this is the last thread of this - * pool, free the pool. */ + * pool, free the pool. + */ free_pool = TRUE; } else @@ -187,7 +296,8 @@ g_thread_pool_thread_proxy (gpointer data) * thread is waiting for this thread pool to * finish and this is not the last thread of * this pool and there are no tasks left in the - * queue, wakeup the remaining threads. */ + * queue, wakeup the remaining threads. + */ if (g_async_queue_length_unlocked (pool->queue) == - pool->num_threads) g_thread_pool_wakeup_and_stop_all (pool); @@ -200,98 +310,20 @@ g_thread_pool_thread_proxy (gpointer data) * waiting for this thread pool to finish and there * are either no tasks left or the pool shall stop * immediatly, inform the waiting thread of a change - * of the thread pool state. */ - g_mutex_lock (inform_mutex); - g_cond_broadcast (inform_cond); - g_mutex_unlock (inform_mutex); + * of the thread pool state. + */ + g_cond_broadcast (pool->cond); } } + g_async_queue_unlock (pool->queue); if (free_pool) g_thread_pool_free_internal (pool); - - g_async_queue_lock (unused_thread_queue); - unused_threads++; - do - { - - if ((unused_threads >= max_unused_threads && - max_unused_threads != -1)) - { - /* If this is a superflous thread, stop it. */ - unused_threads--; - g_async_queue_unlock (unused_thread_queue); - DEBUG_MSG (("stopping thread %p.", g_thread_self ())); - return NULL; - } - - if (max_idle_time > 0) - { - /* If a maximal idle time is given, wait for the - * given time. */ - GTimeVal end_time; - g_get_current_time (&end_time); - g_time_val_add (&end_time, max_idle_time * 1000); - - DEBUG_MSG (("thread %p waiting in global pool for " - "%f seconds.", - g_thread_self (), max_idle_time / 1000.0)); - - pool = g_async_queue_timed_pop_unlocked (unused_thread_queue, - &end_time); - if (!pool) - { - /* If no new task was received in the given - * time, stop this thread. */ - unused_threads--; - g_async_queue_unlock (unused_thread_queue); - DEBUG_MSG (("stopping thread %p after max-idle-time.", - g_thread_self ())); - /* Stop this thread */ - return NULL; - } - } - else - { - /* If no maximal idle time is given, wait - * indefinitly. */ - DEBUG_MSG (("thread %p waiting in global pool.", - g_thread_self ())); - pool = g_async_queue_pop_unlocked (unused_thread_queue); - } - - if (pool == wakeup_thread_marker) - { - if (last_wakeup_thread_serial == wakeup_thread_serial) - { - /* If this wakeup marker has been received for - * the second time, relay it. */ - DEBUG_MSG (("thread %p relaying wakeup message to " - "waiting thread with lower serial.", - g_thread_self ())); - g_async_queue_push_unlocked (unused_thread_queue, - wakeup_thread_marker); - } - else - { - last_wakeup_thread_serial = wakeup_thread_serial; - } - - /* If a wakeup marker has been received, this thread - * will get out of the way for 100 microseconds to - * avoid receiving this marker again. */ - g_async_queue_unlock (unused_thread_queue); - g_usleep (100); - g_async_queue_lock (unused_thread_queue); - } + if ((pool = g_thread_pool_wait_for_new_pool ()) == NULL) + break; - } while (pool == wakeup_thread_marker); - - unused_threads--; - g_async_queue_unlock (unused_thread_queue); - g_async_queue_lock (pool->queue); DEBUG_MSG (("thread %p entering pool %p from global pool.", @@ -299,9 +331,11 @@ g_thread_pool_thread_proxy (gpointer data) /* pool->num_threads++ is not done here, but in * g_thread_pool_start_thread to make the new started thread - * known to the pool, before itself can do it. */ + * known to the pool, before itself can do it. + */ } } + return NULL; } @@ -339,7 +373,8 @@ g_thread_pool_start_thread (GRealThreadPool *pool, } /* See comment in g_thread_pool_thread_proxy as to why this is done - * here and not there */ + * here and not there + */ pool->num_threads++; } @@ -399,6 +434,7 @@ g_thread_pool_new (GFunc func, retval->pool.user_data = user_data; retval->pool.exclusive = exclusive; retval->queue = g_async_queue_new (); + retval->cond = NULL; retval->max_threads = max_threads; retval->num_threads = 0; retval->running = TRUE; @@ -406,14 +442,8 @@ g_thread_pool_new (GFunc func, retval->sort_user_data = NULL; G_LOCK (init); - - if (!inform_mutex) - { - inform_mutex = g_mutex_new (); - inform_cond = g_cond_new (); + if (!unused_thread_queue) unused_thread_queue = g_async_queue_new (); - } - G_UNLOCK (init); if (retval->pool.exclusive) @@ -456,21 +486,18 @@ g_thread_pool_new (GFunc func, * to do. **/ void -g_thread_pool_push (GThreadPool *pool, - gpointer data, - GError **error) +g_thread_pool_push (GThreadPool *pool, + gpointer data, + GError **error) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; + + real = (GRealThreadPool*) pool; g_return_if_fail (real); + g_return_if_fail (real->running); g_async_queue_lock (real->queue); - - if (!real->running) - { - g_async_queue_unlock (real->queue); - g_return_if_fail (real->running); - } if (g_async_queue_length_unlocked (real->queue) >= 0) /* No thread is waiting in the queue */ @@ -504,13 +531,15 @@ g_thread_pool_push (GThreadPool *pool, * created. **/ void -g_thread_pool_set_max_threads (GThreadPool *pool, - gint max_threads, - GError **error) +g_thread_pool_set_max_threads (GThreadPool *pool, + gint max_threads, + GError **error) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; gint to_start; + real = (GRealThreadPool*) pool; + g_return_if_fail (real); g_return_if_fail (real->running); g_return_if_fail (!real->pool.exclusive || max_threads != -1); @@ -528,6 +557,7 @@ g_thread_pool_set_max_threads (GThreadPool *pool, for ( ; to_start > 0; to_start--) { GError *local_error = NULL; + g_thread_pool_start_thread (real, &local_error); if (local_error) { @@ -548,18 +578,18 @@ g_thread_pool_set_max_threads (GThreadPool *pool, * Return value: the maximal number of threads **/ gint -g_thread_pool_get_max_threads (GThreadPool *pool) +g_thread_pool_get_max_threads (GThreadPool *pool) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; gint retval; + real = (GRealThreadPool*) pool; + g_return_val_if_fail (real, 0); g_return_val_if_fail (real->running, 0); g_async_queue_lock (real->queue); - retval = real->max_threads; - g_async_queue_unlock (real->queue); return retval; @@ -574,18 +604,18 @@ g_thread_pool_get_max_threads (GThreadPool *pool) * Return value: the number of threads currently running **/ guint -g_thread_pool_get_num_threads (GThreadPool *pool) +g_thread_pool_get_num_threads (GThreadPool *pool) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; guint retval; + real = (GRealThreadPool*) pool; + g_return_val_if_fail (real, 0); g_return_val_if_fail (real->running, 0); g_async_queue_lock (real->queue); - retval = real->num_threads; - g_async_queue_unlock (real->queue); return retval; @@ -600,11 +630,13 @@ g_thread_pool_get_num_threads (GThreadPool *pool) * Return value: the number of unprocessed tasks **/ guint -g_thread_pool_unprocessed (GThreadPool *pool) +g_thread_pool_unprocessed (GThreadPool *pool) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; gint unprocessed; + real = (GRealThreadPool*) pool; + g_return_val_if_fail (real, 0); g_return_val_if_fail (real->running, 0); @@ -634,17 +666,22 @@ g_thread_pool_unprocessed (GThreadPool *pool) * After calling this function @pool must not be used anymore. **/ void -g_thread_pool_free (GThreadPool *pool, - gboolean immediate, - gboolean wait) +g_thread_pool_free (GThreadPool *pool, + gboolean immediate, + gboolean wait) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; + + real = (GRealThreadPool*) pool; g_return_if_fail (real); g_return_if_fail (real->running); + /* If there's no thread allowed here, there is not much sense in - * not stopping this pool immediately, when it's not empty */ - g_return_if_fail (immediate || real->max_threads != 0 || + * not stopping this pool immediately, when it's not empty + */ + g_return_if_fail (immediate || + real->max_threads != 0 || g_async_queue_length (real->queue) == 0); g_async_queue_lock (real->queue); @@ -655,30 +692,21 @@ g_thread_pool_free (GThreadPool *pool, if (wait) { + real->cond = g_cond_new (); + while (g_async_queue_length_unlocked (real->queue) != -real->num_threads && !(immediate && real->num_threads == 0)) - { - /* This locking is a bit delicate to avoid the broadcast of - * inform_cond to overtake the wait for it. Therefore the - * inform_mutex is locked before the queue is unlocked. To - * avoid deadlocks however, the queue _must_ always be - * locked before locking the inform_mutex, if both are to be - * locked at the same time. */ - g_mutex_lock (inform_mutex); - g_async_queue_unlock (real->queue); - g_cond_wait (inform_cond, inform_mutex); - g_mutex_unlock (inform_mutex); - g_async_queue_lock (real->queue); - } + g_cond_wait (real->cond, _g_async_queue_get_mutex (real->queue)); } - if (immediate || - g_async_queue_length_unlocked (real->queue) == -real->num_threads) + if (immediate || g_async_queue_length_unlocked (real->queue) == -real->num_threads) { /* No thread is currently doing something (and nothing is left - * to process in the queue) */ - if (real->num_threads == 0) /* No threads left, we clean up */ + * to process in the queue) + */ + if (real->num_threads == 0) { + /* No threads left, we clean up */ g_async_queue_unlock (real->queue); g_thread_pool_free_internal (real); return; @@ -687,7 +715,8 @@ g_thread_pool_free (GThreadPool *pool, g_thread_pool_wakeup_and_stop_all (real); } - real->waiting = FALSE; /* The last thread should cleanup the pool */ + /* The last thread should cleanup the pool */ + real->waiting = FALSE; g_async_queue_unlock (real->queue); } @@ -695,11 +724,14 @@ static void g_thread_pool_free_internal (GRealThreadPool* pool) { g_return_if_fail (pool); - g_return_if_fail (!pool->running); + g_return_if_fail (pool->running == FALSE); g_return_if_fail (pool->num_threads == 0); g_async_queue_unref (pool->queue); + if (pool->cond) + g_cond_free (pool->cond); + g_free (pool); } @@ -709,12 +741,11 @@ g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool) guint i; g_return_if_fail (pool); - g_return_if_fail (!pool->running); + g_return_if_fail (pool->running == FALSE); g_return_if_fail (pool->num_threads != 0); - g_return_if_fail (g_async_queue_length_unlocked (pool->queue) == - -pool->num_threads); pool->immediate = TRUE; + for (i = 0; i < pool->num_threads; i++) g_thread_pool_queue_push_unlocked (pool, GUINT_TO_POINTER (1)); } @@ -732,21 +763,28 @@ g_thread_pool_set_max_unused_threads (gint max_threads) { g_return_if_fail (max_threads >= -1); - g_async_queue_lock (unused_thread_queue); - - max_unused_threads = max_threads; + g_atomic_int_set (&max_unused_threads, max_threads); - if (max_unused_threads < unused_threads && max_unused_threads != -1) + if (max_threads != -1) { - guint i; - wakeup_thread_serial++; + max_threads -= g_atomic_int_get (&unused_threads); + if (max_threads < 0) + { + g_atomic_int_set (&kill_unused_threads, -max_threads); + g_atomic_int_inc (&wakeup_thread_serial); - for (i = unused_threads - max_unused_threads; i > 0; i--) - g_async_queue_push_unlocked (unused_thread_queue, - wakeup_thread_marker); + g_async_queue_lock (unused_thread_queue); + + do + { + g_async_queue_push_unlocked (unused_thread_queue, + wakeup_thread_marker); + } + while (++max_threads); + + g_async_queue_unlock (unused_thread_queue); + } } - - g_async_queue_unlock (unused_thread_queue); } /** @@ -759,13 +797,7 @@ g_thread_pool_set_max_unused_threads (gint max_threads) gint g_thread_pool_get_max_unused_threads (void) { - gint retval; - - g_async_queue_lock (unused_thread_queue); - retval = max_unused_threads; - g_async_queue_unlock (unused_thread_queue); - - return retval; + return g_atomic_int_get (&max_unused_threads); } /** @@ -778,13 +810,7 @@ g_thread_pool_get_max_unused_threads (void) guint g_thread_pool_get_num_unused_threads (void) { - guint retval; - - g_async_queue_lock (unused_thread_queue); - retval = unused_threads; - g_async_queue_unlock (unused_thread_queue); - - return retval; + return g_atomic_int_get (&unused_threads); } /** @@ -797,7 +823,10 @@ g_thread_pool_get_num_unused_threads (void) void g_thread_pool_stop_unused_threads (void) { - guint oldval = g_thread_pool_get_max_unused_threads (); + guint oldval; + + oldval = g_thread_pool_get_max_unused_threads (); + g_thread_pool_set_max_unused_threads (0); g_thread_pool_set_max_unused_threads (oldval); } @@ -830,7 +859,9 @@ g_thread_pool_set_sort_function (GThreadPool *pool, GCompareDataFunc func, gpointer user_data) { - GRealThreadPool *real = (GRealThreadPool*) pool; + GRealThreadPool *real; + + real = (GRealThreadPool*) pool; g_return_if_fail (real); g_return_if_fail (real->running); @@ -871,14 +902,23 @@ g_thread_pool_set_max_idle_time (guint interval) { guint i; - g_async_queue_lock (unused_thread_queue); - max_idle_time = interval; - wakeup_thread_serial++; + g_atomic_int_set (&max_idle_time, interval); - for (i = 0; i < unused_threads; i++) - g_async_queue_push_unlocked (unused_thread_queue, wakeup_thread_marker); + i = g_atomic_int_get (&unused_threads); + if (i > 0) + { + g_atomic_int_inc (&wakeup_thread_serial); + g_async_queue_lock (unused_thread_queue); - g_async_queue_unlock (unused_thread_queue); + do + { + g_async_queue_push_unlocked (unused_thread_queue, + wakeup_thread_marker); + } + while (--i); + + g_async_queue_unlock (unused_thread_queue); + } } /** @@ -898,13 +938,7 @@ g_thread_pool_set_max_idle_time (guint interval) guint g_thread_pool_get_max_idle_time (void) { - guint retval; - - g_async_queue_lock (unused_thread_queue); - retval = max_idle_time; - g_async_queue_unlock (unused_thread_queue); - - return retval; + return g_atomic_int_get (&max_idle_time); } #define __G_THREADPOOL_C__ diff --git a/tests/threadpool-test.c b/tests/threadpool-test.c index 970bc3237..be85eff3f 100644 --- a/tests/threadpool-test.c +++ b/tests/threadpool-test.c @@ -8,8 +8,6 @@ #define DEBUG_MSG(x) /* #define DEBUG_MSG(args) g_printerr args ; g_printerr ("\n"); */ -#define RUNS 100 - #define WAIT 5 /* seconds */ #define MAX_THREADS 10 @@ -36,6 +34,116 @@ static GThreadPool *idle_pool = NULL; static GMainLoop *main_loop = NULL; +static void +test_thread_functions (void) +{ + gint max_unused_threads; + guint max_idle_time; + + /* This function attempts to call functions which don't need a + * threadpool to operate to make sure no uninitialised pointers + * accessed and no errors occur. + */ + + max_unused_threads = 3; + + DEBUG_MSG (("[funcs] Setting max unused threads to %d", + max_unused_threads)); + g_thread_pool_set_max_unused_threads (max_unused_threads); + + DEBUG_MSG (("[funcs] Getting max unused threads = %d", + g_thread_pool_get_max_unused_threads ())); + g_assert (g_thread_pool_get_max_unused_threads() == max_unused_threads); + + DEBUG_MSG (("[funcs] Getting num unused threads = %d", + g_thread_pool_get_num_unused_threads ())); + g_assert (g_thread_pool_get_num_unused_threads () == 0); + + DEBUG_MSG (("[funcs] Stopping unused threads")); + g_thread_pool_stop_unused_threads (); + + max_idle_time = 10 * G_USEC_PER_SEC; + + DEBUG_MSG (("[funcs] Setting max idle time to %d", + max_idle_time)); + g_thread_pool_set_max_idle_time (max_idle_time); + + DEBUG_MSG (("[funcs] Getting max idle time = %d", + g_thread_pool_get_max_idle_time ())); + g_assert (g_thread_pool_get_max_idle_time () == max_idle_time); + + DEBUG_MSG (("[funcs] Setting max idle time to 0")); + g_thread_pool_set_max_idle_time (0); + + DEBUG_MSG (("[funcs] Getting max idle time = %d", + g_thread_pool_get_max_idle_time ())); + g_assert (g_thread_pool_get_max_idle_time () == 0); +} + +static void +test_count_threads_foreach (GThread *thread, + guint *count) +{ + ++*count; +} + +static guint +test_count_threads (void) +{ + guint count = 0; + + g_thread_foreach ((GFunc) test_count_threads_foreach, &count); + + /* Exclude main thread */ + return count - 1; +} + +static void +test_thread_stop_unused (void) +{ + GThreadPool *pool; + guint i; + guint limit = 100; + + /* Spawn a few threads. */ + g_thread_pool_set_max_unused_threads (-1); + pool = g_thread_pool_new ((GFunc) g_usleep, NULL, -1, FALSE, NULL); + + for (i = 0; i < limit; i++) + g_thread_pool_push (pool, GUINT_TO_POINTER (1000), NULL); + + DEBUG_MSG (("[unused] ===> pushed %d threads onto the idle pool", + limit)); + + /* Wait for the threads to migrate. */ + g_usleep (G_USEC_PER_SEC); + + DEBUG_MSG (("[unused] current threads %d", + test_count_threads())); + + g_assert (test_count_threads () == limit); + + DEBUG_MSG (("[unused] stopping unused threads")); + + g_thread_pool_stop_unused_threads (); + + DEBUG_MSG (("[unused] waiting ONE second for threads to die")); + + /* Some time for threads to die. */ + g_usleep (G_USEC_PER_SEC); + + DEBUG_MSG (("[unused] stopped idle threads, %d remain, %d threads still exist", + g_thread_pool_get_num_unused_threads (), + test_count_threads ())); + + g_assert (g_thread_pool_get_num_unused_threads () == test_count_threads ()); + g_assert (g_thread_pool_get_num_unused_threads () == 0); + + g_thread_pool_set_max_unused_threads (MAX_THREADS); + + DEBUG_MSG (("[unused] cleaning up thread pool")); + g_thread_pool_free (pool, FALSE, TRUE); +} static void test_thread_pools_entry_func (gpointer data, gpointer user_data) @@ -68,13 +176,15 @@ static void test_thread_pools (void) { GThreadPool *pool1, *pool2, *pool3; + guint runs; guint i; pool1 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 3, FALSE, NULL); pool2 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 5, TRUE, NULL); pool3 = g_thread_pool_new ((GFunc)test_thread_pools_entry_func, NULL, 7, TRUE, NULL); - for (i = 0; i < RUNS; i++) + runs = 300; + for (i = 0; i < runs; i++) { g_thread_pool_push (pool1, GUINT_TO_POINTER (i + 1), NULL); g_thread_pool_push (pool2, GUINT_TO_POINTER (i + 1), NULL); @@ -86,7 +196,7 @@ test_thread_pools (void) g_thread_pool_free (pool2, FALSE, TRUE); g_thread_pool_free (pool3, FALSE, TRUE); - g_assert (RUNS * 3 == abs_thread_counter + leftover_task_counter); + g_assert (runs * 3 == abs_thread_counter + leftover_task_counter); g_assert (running_thread_counter == 0); } @@ -283,15 +393,21 @@ test_check_start_and_stop (gpointer user_data) switch (test_number) { case 1: - test_thread_pools (); + test_thread_functions (); break; case 2: - test_thread_sort (FALSE); + test_thread_stop_unused (); break; case 3: - test_thread_sort (TRUE); + test_thread_pools (); break; case 4: + test_thread_sort (FALSE); + break; + case 5: + test_thread_sort (TRUE); + break; + case 6: test_thread_idle_time (); break; default: @@ -305,7 +421,7 @@ test_check_start_and_stop (gpointer user_data) return TRUE; } - if (test_number == 1) { + if (test_number == 3) { G_LOCK (thread_counter_pools); quit &= running_thread_counter <= 0; DEBUG_MSG (("***** POOL RUNNING THREAD COUNT:%ld", @@ -313,7 +429,7 @@ test_check_start_and_stop (gpointer user_data) G_UNLOCK (thread_counter_pools); } - if (test_number == 2 || test_number == 3) { + if (test_number == 4 || test_number == 5) { G_LOCK (thread_counter_sort); quit &= sort_thread_counter <= 0; DEBUG_MSG (("***** POOL SORT THREAD COUNT:%ld", @@ -321,7 +437,7 @@ test_check_start_and_stop (gpointer user_data) G_UNLOCK (thread_counter_sort); } - if (test_number == 4) { + if (test_number == 6) { guint idle; idle = g_thread_pool_get_num_unused_threads (); |