diff options
| -rw-r--r-- | ChangeLog | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-0 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-10 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-12 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-2 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-4 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-6 | 19 | ||||
| -rw-r--r-- | ChangeLog.pre-2-8 | 19 | ||||
| -rw-r--r-- | Makefile.am | 2 | ||||
| -rw-r--r-- | gasyncqueue.c | 276 | ||||
| -rw-r--r-- | glib.h | 125 | ||||
| -rw-r--r-- | glib/Makefile.am | 2 | ||||
| -rw-r--r-- | glib/gasyncqueue.c | 276 | ||||
| -rw-r--r-- | glib/glib.h | 125 | ||||
| -rw-r--r-- | glib/gthreadpool.c | 520 | ||||
| -rw-r--r-- | gthreadpool.c | 520 | ||||
| -rw-r--r-- | tests/.cvsignore | 1 | ||||
| -rw-r--r-- | tests/Makefile.am | 3 | ||||
| -rw-r--r-- | tests/threadpool-test.c | 57 |
19 files changed, 2049 insertions, 10 deletions
@@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-0 b/ChangeLog.pre-2-0 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-0 +++ b/ChangeLog.pre-2-0 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-10 b/ChangeLog.pre-2-10 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-10 +++ b/ChangeLog.pre-2-10 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-12 b/ChangeLog.pre-2-12 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-12 +++ b/ChangeLog.pre-2-12 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-2 b/ChangeLog.pre-2-2 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-2 +++ b/ChangeLog.pre-2-2 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-4 b/ChangeLog.pre-2-4 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-4 +++ b/ChangeLog.pre-2-4 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-6 b/ChangeLog.pre-2-6 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-6 +++ b/ChangeLog.pre-2-6 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/ChangeLog.pre-2-8 b/ChangeLog.pre-2-8 index 938c76bca..ab5293a48 100644 --- a/ChangeLog.pre-2-8 +++ b/ChangeLog.pre-2-8 @@ -1,3 +1,20 @@ +2000-04-28 Sebastian Wilhelmi <wilhelmi@ira.uka.de> + + * gasyncqueue.c: New File implementing an asynchronous queue to be + used for asynchronous inter-thread communication. + + * gthreadpool.c: New File implementing a thread pool to be used + for distributing work among several threads. + + * glib.h: Added the type and function declarations for these two + types. + + * tests/threadpool-test.c: New File implementing a test for the + thread pool. This also checks the asynchronous queue underlying + the thread pool. + + * tests/Makefile.am: Changed accordingly. + 2000-04-26 Sebastian Wilhelmi <wilhelmi@ira.uka.de> * configure.in: Look for both pthread_create and pthread_join in @@ -8,7 +25,7 @@ * glib.h, gcache.c, gtree.c: Changed the 'value' parameter of g_cache_remove from gpointer to gconstpointer. Dito for the 'key' parameter of g_tree_lookup and g_tree_remove and the 'data' - parameter of g_tree_serach. This function now takes a function of + parameter of g_tree_search. This function now takes a function of type GCompareFunc instead of GSearchFunc. This fixes Bug #8267. Thanks to Juan Toledo <toledo@users.sourceforge.net> for pointing that out. diff --git a/Makefile.am b/Makefile.am index 3f970a3c7..95221d5ad 100644 --- a/Makefile.am +++ b/Makefile.am @@ -35,6 +35,7 @@ lib_LTLIBRARIES = libglib.la libglib_la_SOURCES = \ garray.c \ + gasyncqueue.c \ gcache.c \ gcompletion.c \ gdataset.c \ @@ -58,6 +59,7 @@ libglib_la_SOURCES = \ gstrfuncs.c \ gstring.c \ gthread.c \ + gthreadpool.c \ gtimer.c \ gtree.c \ gutils.c diff --git a/gasyncqueue.c b/gasyncqueue.c new file mode 100644 index 000000000..470d2a78f --- /dev/null +++ b/gasyncqueue.c @@ -0,0 +1,276 @@ +/* GLIB - Library of useful routines for C programming + * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald + * + * GAsyncQueue: asyncronous queue implementation, based on Gqueue. + * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/* + * MT safe + */ + +#include "glib.h" + +struct _GAsyncQueue +{ + GMutex *mutex; + GCond *cond; + GQueue *queue; + guint waiting_threads; + guint ref_count; +}; + +GAsyncQueue* +g_async_queue_new () +{ + GAsyncQueue* retval = g_new (GAsyncQueue, 1); + retval->mutex = g_mutex_new (); + retval->cond = g_cond_new (); + retval->queue = g_queue_new (); + retval->waiting_threads = 0; + retval->ref_count = 1; + return retval; +} + +void +g_async_queue_ref (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); + queue->ref_count++; + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_ref_unlocked (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + queue->ref_count++; +} + +void +g_async_queue_unref_and_unlock (GAsyncQueue *queue) +{ + gboolean stop; + + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + queue->ref_count--; + stop = (queue->ref_count == 0); + g_mutex_unlock (queue->mutex); + + if (stop) + { + g_return_if_fail (queue->waiting_threads == 0); + g_mutex_free (queue->mutex); + g_cond_free (queue->cond); + g_queue_free (queue->queue); + g_free (queue); + } +} + +void +g_async_queue_unref (GAsyncQueue *queue) +{ + gboolean stop; + + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); + g_async_queue_unref_and_unlock (queue); +} + +void +g_async_queue_lock (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); +} + +void +g_async_queue_unlock (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_push (GAsyncQueue* queue, gpointer data) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + g_return_if_fail (data); + + g_mutex_lock (queue->mutex); + g_async_queue_push_unlocked (queue, data); + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + g_return_if_fail (data); + + g_queue_push_head (queue->queue, data); + g_cond_signal (queue->cond); +} + +static gpointer +g_async_queue_pop_intern_unlocked (GAsyncQueue* queue, gboolean try, + GTimeVal *end_time) +{ + gpointer retval; + + if (!g_queue_peek_tail (queue->queue)) + { + if (try) + return NULL; + if (!end_time) + { + queue->waiting_threads++; + while (!g_queue_peek_tail (queue->queue)) + g_cond_wait(queue->cond, queue->mutex); + queue->waiting_threads--; + } + else + { + queue->waiting_threads++; + while (!g_queue_peek_tail (queue->queue)) + if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time)) + break; + queue->waiting_threads--; + if (!g_queue_peek_tail (queue->queue)) + return NULL; + } + } + + retval = g_queue_pop_tail (queue->queue); + + g_assert (retval); + + return retval; +} + +gpointer +g_async_queue_pop (GAsyncQueue* queue) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_pop_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL); +} + +gpointer +g_async_queue_try_pop (GAsyncQueue* queue) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_try_pop_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL); +} + +gpointer +g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time); +} + +gint +g_async_queue_length_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, 0); + g_return_val_if_fail (queue->ref_count > 0, 0); + + return queue->queue->length - queue->waiting_threads; +} + +gint +g_async_queue_length(GAsyncQueue* queue) +{ + glong retval; + + g_return_val_if_fail (queue, 0); + g_return_val_if_fail (queue->ref_count > 0, 0); + + g_mutex_lock (queue->mutex); + retval = queue->queue->length - queue->waiting_threads; + g_mutex_unlock (queue->mutex); + + return retval; +} + @@ -1075,7 +1075,6 @@ void g_queue_push_tail_link (GQueue *queue, GList* g_queue_pop_head_link (GQueue *queue); GList* g_queue_pop_tail_link (GQueue *queue); - /* Hash tables */ GHashTable* g_hash_table_new (GHashFunc hash_func, @@ -2957,6 +2956,8 @@ typedef struct _GMutex GMutex; typedef struct _GCond GCond; typedef struct _GPrivate GPrivate; typedef struct _GStaticPrivate GStaticPrivate; +typedef struct _GAsyncQueue GAsyncQueue; +typedef struct _GThreadPool GThreadPool; typedef struct _GThreadFunctions GThreadFunctions; struct _GThreadFunctions @@ -3169,6 +3170,128 @@ extern void glib_dummy_decl (void); # define G_TRYLOCK(name) (TRUE) #endif /* !G_THREADS_ENABLED */ +/* Asyncronous Queues, can be used to communicate between threads + */ + +/* Get a new GAsyncQueue with the ref_count 1 */ +GAsyncQueue* g_async_queue_new (void); + +/* Lock and unlock an GAsyncQueue, all functions lock the queue for + * themselves, but in certain cirumstances you want to hold the lock longer, + * thus you lock the queue, call the *_unlocked functions and unlock it again + */ +void g_async_queue_lock (GAsyncQueue *queue); +void g_async_queue_unlock (GAsyncQueue *queue); + +/* Ref and unref the GAsyncQueue. g_async_queue_unref_unlocked makes + * no sense, as after the unreffing the Queue might be gone and can't + * be unlocked. So you have a function to call, if you don't hold the + * lock (g_async_queue_unref) and one to call, when you already hold + * the lock (g_async_queue_unref_and_unlock). After that however, you + * don't hold the lock anymore and the Queue might in fact be + * destroyed, if you unrefed to zero */ +void g_async_queue_ref (GAsyncQueue *queue); +void g_async_queue_ref_unlocked (GAsyncQueue *queue); +void g_async_queue_unref (GAsyncQueue *queue); +void g_async_queue_unref_and_unlock (GAsyncQueue *queue); + +/* Push data into the async queue. Must not be NULL */ +void g_async_queue_push (GAsyncQueue *queue, + gpointer data); +void g_async_queue_push_unlocked (GAsyncQueue *queue, + gpointer data); + +/* Pop data from the async queue, when no data is there, the thread is blocked + * 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); + +/* Wait for data until at maximum until end_time is reached, NULL is returned + * 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. */ +gint g_async_queue_length (GAsyncQueue *queue); +gint g_async_queue_length_unlocked (GAsyncQueue *queue); + +/* Thread Pools + */ + +/* The real GThreadPool is bigger, so you may only create a thread + * pool with the constructor function */ +struct _GThreadPool +{ + GFunc thread_func; + gulong stack_size; + gboolean bound; + GThreadPriority priority; + gboolean exclusive; + gpointer user_data; +}; + +/* Get a thread pool with the function thread_func, at most max_threads may + * run at a time (max_threads == -1 means no limit), stack_size, bound, + * priority like in g_thread_create, exclusive == TRUE means, that the threads + * shouldn't be shared and that they will be prestarted (otherwise they are + * started, as needed) user_data is the 2nd argument to the thread_func */ +GThreadPool* g_thread_pool_new (GFunc thread_func, + gint max_threads, + gulong stack_size, + gboolean bound, + GThreadPriority priority, + gboolean exclusive, + gpointer user_data); + +/* Push new data into the thread pool. This task is assigned to a thread later + * (when the maximal number of threads is reached for that pool) or now + * (otherwise). If necessary a new thread will be started. The function + * returns immediatly */ +void g_thread_pool_push (GThreadPool *pool, + gpointer data); + +/* Set the number of threads, which can run concurrently for that pool, -1 + * means no limit. 0 means has the effect, that the pool won't process + * requests until the limit is set higher again */ +void g_thread_pool_set_max_threads (GThreadPool *pool, + gint max_threads); +gint g_thread_pool_get_max_threads (GThreadPool *pool); + +/* Get the number of threads assigned to that pool. This number doesn't + * necessarily represent the number of working threads in that pool */ +guint g_thread_pool_get_num_threads (GThreadPool *pool); + +/* Get the number of unprocessed items in the pool */ +guint g_thread_pool_unprocessed (GThreadPool *pool); + +/* Free the pool, immediate means, that all unprocessed items in the queue + * wont be processed, wait means, that the function doesn't return immediatly, + * but after all threads in the pool are ready processing items. immediate + * does however not mean, that threads are killed. */ +void g_thread_pool_free (GThreadPool *pool, + gboolean immediate, + gboolean wait); + +/* Set the maximal number of unused threads before threads will be stopped by + * GLib, -1 means no limit */ +void g_thread_pool_set_max_unused_threads (gint max_threads); +gint g_thread_pool_get_max_unused_threads (void); +guint g_thread_pool_get_num_unused_threads (void); + +/* Stop all currently unused threads, but leave the limit untouched */ +void g_thread_pool_stop_unused_threads (void); + #ifdef __cplusplus } #endif /* __cplusplus */ diff --git a/glib/Makefile.am b/glib/Makefile.am index 3f970a3c7..95221d5ad 100644 --- a/glib/Makefile.am +++ b/glib/Makefile.am @@ -35,6 +35,7 @@ lib_LTLIBRARIES = libglib.la libglib_la_SOURCES = \ garray.c \ + gasyncqueue.c \ gcache.c \ gcompletion.c \ gdataset.c \ @@ -58,6 +59,7 @@ libglib_la_SOURCES = \ gstrfuncs.c \ gstring.c \ gthread.c \ + gthreadpool.c \ gtimer.c \ gtree.c \ gutils.c diff --git a/glib/gasyncqueue.c b/glib/gasyncqueue.c new file mode 100644 index 000000000..470d2a78f --- /dev/null +++ b/glib/gasyncqueue.c @@ -0,0 +1,276 @@ +/* GLIB - Library of useful routines for C programming + * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald + * + * GAsyncQueue: asyncronous queue implementation, based on Gqueue. + * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/* + * MT safe + */ + +#include "glib.h" + +struct _GAsyncQueue +{ + GMutex *mutex; + GCond *cond; + GQueue *queue; + guint waiting_threads; + guint ref_count; +}; + +GAsyncQueue* +g_async_queue_new () +{ + GAsyncQueue* retval = g_new (GAsyncQueue, 1); + retval->mutex = g_mutex_new (); + retval->cond = g_cond_new (); + retval->queue = g_queue_new (); + retval->waiting_threads = 0; + retval->ref_count = 1; + return retval; +} + +void +g_async_queue_ref (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); + queue->ref_count++; + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_ref_unlocked (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + queue->ref_count++; +} + +void +g_async_queue_unref_and_unlock (GAsyncQueue *queue) +{ + gboolean stop; + + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + queue->ref_count--; + stop = (queue->ref_count == 0); + g_mutex_unlock (queue->mutex); + + if (stop) + { + g_return_if_fail (queue->waiting_threads == 0); + g_mutex_free (queue->mutex); + g_cond_free (queue->cond); + g_queue_free (queue->queue); + g_free (queue); + } +} + +void +g_async_queue_unref (GAsyncQueue *queue) +{ + gboolean stop; + + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); + g_async_queue_unref_and_unlock (queue); +} + +void +g_async_queue_lock (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_lock (queue->mutex); +} + +void +g_async_queue_unlock (GAsyncQueue *queue) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_push (GAsyncQueue* queue, gpointer data) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + g_return_if_fail (data); + + g_mutex_lock (queue->mutex); + g_async_queue_push_unlocked (queue, data); + g_mutex_unlock (queue->mutex); +} + +void +g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data) +{ + g_return_if_fail (queue); + g_return_if_fail (queue->ref_count > 0); + g_return_if_fail (data); + + g_queue_push_head (queue->queue, data); + g_cond_signal (queue->cond); +} + +static gpointer +g_async_queue_pop_intern_unlocked (GAsyncQueue* queue, gboolean try, + GTimeVal *end_time) +{ + gpointer retval; + + if (!g_queue_peek_tail (queue->queue)) + { + if (try) + return NULL; + if (!end_time) + { + queue->waiting_threads++; + while (!g_queue_peek_tail (queue->queue)) + g_cond_wait(queue->cond, queue->mutex); + queue->waiting_threads--; + } + else + { + queue->waiting_threads++; + while (!g_queue_peek_tail (queue->queue)) + if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time)) + break; + queue->waiting_threads--; + if (!g_queue_peek_tail (queue->queue)) + return NULL; + } + } + + retval = g_queue_pop_tail (queue->queue); + + g_assert (retval); + + return retval; +} + +gpointer +g_async_queue_pop (GAsyncQueue* queue) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_pop_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL); +} + +gpointer +g_async_queue_try_pop (GAsyncQueue* queue) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_try_pop_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL); +} + +gpointer +g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time) +{ + gpointer retval; + + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + g_mutex_lock (queue->mutex); + retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time); + g_mutex_unlock (queue->mutex); + + return retval; +} + +gpointer +g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time) +{ + g_return_val_if_fail (queue, NULL); + g_return_val_if_fail (queue->ref_count > 0, NULL); + + return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time); +} + +gint +g_async_queue_length_unlocked (GAsyncQueue* queue) +{ + g_return_val_if_fail (queue, 0); + g_return_val_if_fail (queue->ref_count > 0, 0); + + return queue->queue->length - queue->waiting_threads; +} + +gint +g_async_queue_length(GAsyncQueue* queue) +{ + glong retval; + + g_return_val_if_fail (queue, 0); + g_return_val_if_fail (queue->ref_count > 0, 0); + + g_mutex_lock (queue->mutex); + retval = queue->queue->length - queue->waiting_threads; + g_mutex_unlock (queue->mutex); + + return retval; +} + diff --git a/glib/glib.h b/glib/glib.h index 94bb61a1e..799a07a94 100644 --- a/glib/glib.h +++ b/glib/glib.h @@ -1075,7 +1075,6 @@ void g_queue_push_tail_link (GQueue *queue, GList* g_queue_pop_head_link (GQueue *queue); GList* g_queue_pop_tail_link (GQueue *queue); - /* Hash tables */ GHashTable* g_hash_table_new (GHashFunc hash_func, @@ -2957,6 +2956,8 @@ typedef struct _GMutex GMutex; typedef struct _GCond GCond; typedef struct _GPrivate GPrivate; typedef struct _GStaticPrivate GStaticPrivate; +typedef struct _GAsyncQueue GAsyncQueue; +typedef struct _GThreadPool GThreadPool; typedef struct _GThreadFunctions GThreadFunctions; struct _GThreadFunctions @@ -3169,6 +3170,128 @@ extern void glib_dummy_decl (void); # define G_TRYLOCK(name) (TRUE) #endif /* !G_THREADS_ENABLED */ +/* Asyncronous Queues, can be used to communicate between threads + */ + +/* Get a new GAsyncQueue with the ref_count 1 */ +GAsyncQueue* g_async_queue_new (void); + +/* Lock and unlock an GAsyncQueue, all functions lock the queue for + * themselves, but in certain cirumstances you want to hold the lock longer, + * thus you lock the queue, call the *_unlocked functions and unlock it again + */ +void g_async_queue_lock (GAsyncQueue *queue); +void g_async_queue_unlock (GAsyncQueue *queue); + +/* Ref and unref the GAsyncQueue. g_async_queue_unref_unlocked makes + * no sense, as after the unreffing the Queue might be gone and can't + * be unlocked. So you have a function to call, if you don't hold the + * lock (g_async_queue_unref) and one to call, when you already hold + * the lock (g_async_queue_unref_and_unlock). After that however, you + * don't hold the lock anymore and the Queue might in fact be + * destroyed, if you unrefed to zero */ +void g_async_queue_ref (GAsyncQueue *queue); +void g_async_queue_ref_unlocked (GAsyncQueue *queue); +void g_async_queue_unref (GAsyncQueue *queue); +void g_async_queue_unref_and_unlock (GAsyncQueue *queue); + +/* Push data into the async queue. Must not be NULL */ +void g_async_queue_push (GAsyncQueue *queue, + gpointer data); +void g_async_queue_push_unlocked (GAsyncQueue *queue, + gpointer data); + +/* Pop data from the async queue, when no data is there, the thread is blocked + * 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); + +/* Wait for data until at maximum until end_time is reached, NULL is returned + * 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. */ +gint g_async_queue_length (GAsyncQueue *queue); +gint g_async_queue_length_unlocked (GAsyncQueue *queue); + +/* Thread Pools + */ + +/* The real GThreadPool is bigger, so you may only create a thread + * pool with the constructor function */ +struct _GThreadPool +{ + GFunc thread_func; + gulong stack_size; + gboolean bound; + GThreadPriority priority; + gboolean exclusive; + gpointer user_data; +}; + +/* Get a thread pool with the function thread_func, at most max_threads may + * run at a time (max_threads == -1 means no limit), stack_size, bound, + * priority like in g_thread_create, exclusive == TRUE means, that the threads + * shouldn't be shared and that they will be prestarted (otherwise they are + * started, as needed) user_data is the 2nd argument to the thread_func */ +GThreadPool* g_thread_pool_new (GFunc thread_func, + gint max_threads, + gulong stack_size, + gboolean bound, + GThreadPriority priority, + gboolean exclusive, + gpointer user_data); + +/* Push new data into the thread pool. This task is assigned to a thread later + * (when the maximal number of threads is reached for that pool) or now + * (otherwise). If necessary a new thread will be started. The function + * returns immediatly */ +void g_thread_pool_push (GThreadPool *pool, + gpointer data); + +/* Set the number of threads, which can run concurrently for that pool, -1 + * means no limit. 0 means has the effect, that the pool won't process + * requests until the limit is set higher again */ +void g_thread_pool_set_max_threads (GThreadPool *pool, + gint max_threads); +gint g_thread_pool_get_max_threads (GThreadPool *pool); + +/* Get the number of threads assigned to that pool. This number doesn't + * necessarily represent the number of working threads in that pool */ +guint g_thread_pool_get_num_threads (GThreadPool *pool); + +/* Get the number of unprocessed items in the pool */ +guint g_thread_pool_unprocessed (GThreadPool *pool); + +/* Free the pool, immediate means, that all unprocessed items in the queue + * wont be processed, wait means, that the function doesn't return immediatly, + * but after all threads in the pool are ready processing items. immediate + * does however not mean, that threads are killed. */ +void g_thread_pool_free (GThreadPool *pool, + gboolean immediate, + gboolean wait); + +/* Set the maximal number of unused threads before threads will be stopped by + * GLib, -1 means no limit */ +void g_thread_pool_set_max_unused_threads (gint max_threads); +gint g_thread_pool_get_max_unused_threads (void); +guint g_thread_pool_get_num_unused_threads (void); + +/* Stop all currently unused threads, but leave the limit untouched */ +void g_thread_pool_stop_unused_threads (void); + #ifdef __cplusplus } #endif /* __cplusplus */ diff --git a/glib/gthreadpool.c b/glib/gthreadpool.c new file mode 100644 index 000000000..0ca36444c --- /dev/null +++ b/glib/gthreadpool.c @@ -0,0 +1,520 @@ +/* GLIB - Library of useful routines for C programming + * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald + * + * GAsyncQueue: thread pool implementation. + * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/* + * MT safe + */ + +#include "glib.h" + +typedef struct _GRealThreadPool GRealThreadPool; + +struct _GRealThreadPool +{ + GThreadPool pool; + GAsyncQueue* queue; + gint max_threads; + gint num_threads; + gboolean running; + gboolean immediate; + gboolean waiting; +}; + +/* The following is just an address to mark the stop order for a + * thread, it could be any address (as long, as it isn;t a valid + * GThreadPool address) */ +static const gpointer stop_this_thread_marker = &g_thread_pool_new; + +/* Here all unused threads are waiting, depending on their priority */ +static GAsyncQueue *unused_thread_queue[G_THREAD_PRIORITY_URGENT + 1]; +static gint unused_threads = 0; +static gint max_unused_threads = 0; +G_LOCK_DEFINE_STATIC (unused_threads); + +static GMutex *inform_mutex = NULL; +static GCond *inform_cond = NULL; + +static void g_thread_pool_free_internal (GRealThreadPool* pool); +static void g_thread_pool_thread_proxy (gpointer data); +static void g_thread_pool_start_thread (GRealThreadPool* pool); +static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool); + +#define g_thread_should_run(pool, len) \ + ((pool)->running || (!(pool)->immediate && (len) > 0)) + +static void +g_thread_pool_thread_proxy (gpointer data) +{ + GRealThreadPool *pool = data; + GThread* self = g_thread_self (); + + g_async_queue_lock (pool->queue); + while (TRUE) + { + gpointer task; + gboolean goto_global_pool = !pool->pool.exclusive; + gint len = g_async_queue_length_unlocked (pool->queue); + + if (g_thread_should_run (pool, len)) + { + task = g_async_queue_pop_unlocked (pool->queue); + + if (pool->num_threads > pool->max_threads && pool->max_threads != -1) + /* We are in fact a superfluous threads, so we go to the + * global pool and just hand the data further to the next one + * waiting in the queue */ + { + g_async_queue_push_unlocked (pool->queue, task); + goto_global_pool = TRUE; + } + else if (pool->running || !pool->immediate) + { + g_async_queue_unlock (pool->queue); + pool->pool.thread_func (task, pool->pool.user_data); + g_async_queue_lock (pool->queue); + } + + len = g_async_queue_length_unlocked (pool->queue); + } + + if (!g_thread_should_run (pool, len)) + g_cond_broadcast (inform_cond); + + if (!pool->running && (pool->immediate || len <= 0)) + goto_global_pool = TRUE; + else if (len >= 0) + /* At this pool there is no thread waiting */ + goto_global_pool = FALSE; + + if (goto_global_pool) + { + GThreadPriority priority = pool->pool.priority; + pool->num_threads--; + + if (!pool->running && !pool->waiting) + { + if (pool->num_threads == 0) + { + g_async_queue_unlock (pool->queue); + g_thread_pool_free_internal (pool); + } + else if (len == - pool->num_threads) + g_thread_pool_wakeup_and_stop_all (pool); + } + else + g_async_queue_unlock (pool->queue); + + g_async_queue_lock (unused_thread_queue[priority]); + + G_LOCK (unused_threads); + if (unused_threads >= max_unused_threads) + { + G_UNLOCK (unused_threads); + g_async_queue_unlock (unused_thread_queue[priority]); + /* Stop this thread */ + return; + } + unused_threads++; + G_UNLOCK (unused_threads); + + pool = + g_async_queue_pop_unlocked (unused_thread_queue[priority]); + + G_LOCK (unused_threads); + unused_threads--; + G_UNLOCK (unused_threads); + + g_async_queue_unlock (unused_thread_queue[priority]); + + if (pool == stop_this_thread_marker) + /* Stop this thread */ + return; + + g_async_queue_lock (pool->queue); + + if (pool->pool.priority != self->priority) + g_thread_set_priority (self, pool->pool.priority); + + /* 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. */ + } + } +} + +static void +g_thread_pool_start_thread (GRealThreadPool* pool) +{ + gboolean success = FALSE; + GThreadPriority priority = pool->pool.priority; + GAsyncQueue *queue = unused_thread_queue[priority]; + + if (pool->num_threads >= pool->max_threads && pool->max_threads != -1) + /* Enough threads are already running */ + return; + + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < 0) + { + /* First we try a thread with the right priority */ + g_async_queue_push_unlocked (queue, pool); + success = TRUE; + } + + g_async_queue_unlock (queue); + + if (!success) + { + /* Now we search for threads with other priorities too, but + * only, when there is more than one unused thread with that + * priority. */ + GThreadPriority priority; + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1; priority++) + { + queue = unused_thread_queue[priority]; + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < -1) + { + g_async_queue_push_unlocked (queue, pool); + success = TRUE; + } + + g_async_queue_unlock (queue); + } + } + + if (!success) + /* No thread was found, we have to start one new */ + g_thread_create (g_thread_pool_thread_proxy, pool, pool->pool.stack_size, + FALSE, pool->pool.bound, priority); + + /* See comment in g_thread_pool_thread_proxy as to why this is done + * here and not there */ + pool->num_threads++; +} + +GThreadPool* +g_thread_pool_new (GFunc thread_func, + gint max_threads, + gulong stack_size, + gboolean bound, + GThreadPriority priority, + gboolean exclusive, + gpointer user_data) +{ + GRealThreadPool *retval; + + g_return_val_if_fail (thread_func, NULL); + g_return_val_if_fail (!exclusive || max_threads != -1, NULL); + g_return_val_if_fail (max_threads >= -1, NULL); + g_return_val_if_fail (g_thread_supported (), NULL); + + retval = g_new (GRealThreadPool, 1); + + retval->pool.thread_func = thread_func; + retval->pool.stack_size = stack_size; + retval->pool.bound = bound; + retval->pool.priority = priority; + retval->pool.exclusive = exclusive; + retval->pool.user_data = user_data; + + retval->queue = g_async_queue_new (); + retval->max_threads = max_threads; + retval->num_threads = 0; + retval->running = TRUE; + + if (!inform_mutex) + { + inform_mutex = g_mutex_new (); + inform_cond = g_cond_new (); + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1; priority++) + unused_thread_queue[priority] = g_async_queue_new (); + } + + if (retval->pool.exclusive) + { + g_async_queue_lock (retval->queue); + + while (retval->num_threads < retval->max_threads) + g_thread_pool_start_thread (retval); + + g_async_queue_unlock (retval->queue); + } + + return (GThreadPool*) retval; +} + +void +g_thread_pool_push (GThreadPool *pool, + gpointer data) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + + g_return_if_fail (real); + + g_async_queue_lock (real->queue); + + if (!real->running) + { + g_async_queue_unlock (real->queue); + g_return_if_fail (real->running); + } + + if (!pool->exclusive && g_async_queue_length_unlocked (real->queue) >= 0) + { + /* No thread is waiting in the queue */ + g_thread_pool_start_thread (real); + } + g_async_queue_push_unlocked (real->queue, data); + g_async_queue_unlock (real->queue); +} + +void +g_thread_pool_set_max_threads (GThreadPool *pool, + gint max_threads) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint to_start; + + g_return_if_fail (real); + g_return_if_fail (real->running); + g_return_if_fail (!real->pool.exclusive || max_threads != -1); + g_return_if_fail (max_threads >= -1); + + g_async_queue_lock (real->queue); + + real->max_threads = max_threads; + + if (pool->exclusive) + to_start = real->max_threads - real->num_threads; + else + to_start = g_async_queue_length_unlocked (real->queue); + + for ( ; to_start > 0; to_start--) + g_thread_pool_start_thread (real); + + g_async_queue_unlock (real->queue); +} + +gint +g_thread_pool_get_max_threads (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint retval; + + 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; +} + +guint +g_thread_pool_get_num_threads (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + guint retval; + + 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; +} + +guint +g_thread_pool_unprocessed (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint unprocessed; + + g_return_val_if_fail (real, 0); + g_return_val_if_fail (real->running, 0); + + unprocessed = g_async_queue_length (real->queue); + + return MAX (unprocessed, 0); +} + +void +g_thread_pool_free (GThreadPool *pool, + gboolean immediate, + gboolean wait) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + + g_return_if_fail (real); + g_return_if_fail (real->running); + /* It there's no thread allowed here, there is not much sense in + * not stopping this pool immediatly, 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); + + real->running = FALSE; + real->immediate = immediate; + real->waiting = wait; + + if (wait) + { + g_mutex_lock (inform_mutex); + while (g_async_queue_length_unlocked (real->queue) != -real->num_threads) + { + g_async_queue_unlock (real->queue); + g_cond_wait (inform_cond, inform_mutex); + g_async_queue_lock (real->queue); + } + g_mutex_unlock (inform_mutex); + } + + if (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 */ + { + g_async_queue_unlock (real->queue); + g_thread_pool_free_internal (real); + return; + } + + g_thread_pool_wakeup_and_stop_all (real); + } + + real->waiting = FALSE; /* The last thread should cleanup the pool */ + g_async_queue_unlock (real->queue); +} + +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->num_threads == 0); + + g_async_queue_unref (pool->queue); + + g_free (pool); +} + +static void +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->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_async_queue_push_unlocked (pool->queue, GUINT_TO_POINTER (1)); +} + +void +g_thread_pool_set_max_unused_threads (gint max_threads) +{ + g_return_if_fail (max_threads >= -1); + + G_LOCK (unused_threads); + + max_unused_threads = max_threads; + + if (max_unused_threads < unused_threads && max_unused_threads != -1) + { + guint close_down_num = unused_threads - max_unused_threads; + GThreadPriority priority; + + while (close_down_num > 0) + { + guint old_close_down_num = close_down_num; + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1 && close_down_num > 0; + priority++) + { + GAsyncQueue *queue = unused_thread_queue[priority]; + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < 0) + { + g_async_queue_push_unlocked (queue, + stop_this_thread_marker); + close_down_num--; + } + + g_async_queue_unlock (queue); + } + + /* Just to make sure, there are no counting problems */ + g_assert (old_close_down_num != close_down_num); + } + } + + G_UNLOCK (unused_threads); +} + +gint +g_thread_pool_get_max_unused_threads (void) +{ + gint retval; + + G_LOCK (unused_threads); + retval = max_unused_threads; + G_UNLOCK (unused_threads); + + return retval; +} + +guint g_thread_pool_get_num_unused_threads (void) +{ + guint retval; + + G_LOCK (unused_threads); + retval = unused_threads; + G_UNLOCK (unused_threads); + + return retval; +} + +void g_thread_pool_stop_unused_threads (void) +{ + guint oldval = g_thread_pool_get_max_unused_threads (); + g_thread_pool_set_max_unused_threads (0); + g_thread_pool_set_max_unused_threads (oldval); +} diff --git a/gthreadpool.c b/gthreadpool.c new file mode 100644 index 000000000..0ca36444c --- /dev/null +++ b/gthreadpool.c @@ -0,0 +1,520 @@ +/* GLIB - Library of useful routines for C programming + * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald + * + * GAsyncQueue: thread pool implementation. + * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/* + * MT safe + */ + +#include "glib.h" + +typedef struct _GRealThreadPool GRealThreadPool; + +struct _GRealThreadPool +{ + GThreadPool pool; + GAsyncQueue* queue; + gint max_threads; + gint num_threads; + gboolean running; + gboolean immediate; + gboolean waiting; +}; + +/* The following is just an address to mark the stop order for a + * thread, it could be any address (as long, as it isn;t a valid + * GThreadPool address) */ +static const gpointer stop_this_thread_marker = &g_thread_pool_new; + +/* Here all unused threads are waiting, depending on their priority */ +static GAsyncQueue *unused_thread_queue[G_THREAD_PRIORITY_URGENT + 1]; +static gint unused_threads = 0; +static gint max_unused_threads = 0; +G_LOCK_DEFINE_STATIC (unused_threads); + +static GMutex *inform_mutex = NULL; +static GCond *inform_cond = NULL; + +static void g_thread_pool_free_internal (GRealThreadPool* pool); +static void g_thread_pool_thread_proxy (gpointer data); +static void g_thread_pool_start_thread (GRealThreadPool* pool); +static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool); + +#define g_thread_should_run(pool, len) \ + ((pool)->running || (!(pool)->immediate && (len) > 0)) + +static void +g_thread_pool_thread_proxy (gpointer data) +{ + GRealThreadPool *pool = data; + GThread* self = g_thread_self (); + + g_async_queue_lock (pool->queue); + while (TRUE) + { + gpointer task; + gboolean goto_global_pool = !pool->pool.exclusive; + gint len = g_async_queue_length_unlocked (pool->queue); + + if (g_thread_should_run (pool, len)) + { + task = g_async_queue_pop_unlocked (pool->queue); + + if (pool->num_threads > pool->max_threads && pool->max_threads != -1) + /* We are in fact a superfluous threads, so we go to the + * global pool and just hand the data further to the next one + * waiting in the queue */ + { + g_async_queue_push_unlocked (pool->queue, task); + goto_global_pool = TRUE; + } + else if (pool->running || !pool->immediate) + { + g_async_queue_unlock (pool->queue); + pool->pool.thread_func (task, pool->pool.user_data); + g_async_queue_lock (pool->queue); + } + + len = g_async_queue_length_unlocked (pool->queue); + } + + if (!g_thread_should_run (pool, len)) + g_cond_broadcast (inform_cond); + + if (!pool->running && (pool->immediate || len <= 0)) + goto_global_pool = TRUE; + else if (len >= 0) + /* At this pool there is no thread waiting */ + goto_global_pool = FALSE; + + if (goto_global_pool) + { + GThreadPriority priority = pool->pool.priority; + pool->num_threads--; + + if (!pool->running && !pool->waiting) + { + if (pool->num_threads == 0) + { + g_async_queue_unlock (pool->queue); + g_thread_pool_free_internal (pool); + } + else if (len == - pool->num_threads) + g_thread_pool_wakeup_and_stop_all (pool); + } + else + g_async_queue_unlock (pool->queue); + + g_async_queue_lock (unused_thread_queue[priority]); + + G_LOCK (unused_threads); + if (unused_threads >= max_unused_threads) + { + G_UNLOCK (unused_threads); + g_async_queue_unlock (unused_thread_queue[priority]); + /* Stop this thread */ + return; + } + unused_threads++; + G_UNLOCK (unused_threads); + + pool = + g_async_queue_pop_unlocked (unused_thread_queue[priority]); + + G_LOCK (unused_threads); + unused_threads--; + G_UNLOCK (unused_threads); + + g_async_queue_unlock (unused_thread_queue[priority]); + + if (pool == stop_this_thread_marker) + /* Stop this thread */ + return; + + g_async_queue_lock (pool->queue); + + if (pool->pool.priority != self->priority) + g_thread_set_priority (self, pool->pool.priority); + + /* 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. */ + } + } +} + +static void +g_thread_pool_start_thread (GRealThreadPool* pool) +{ + gboolean success = FALSE; + GThreadPriority priority = pool->pool.priority; + GAsyncQueue *queue = unused_thread_queue[priority]; + + if (pool->num_threads >= pool->max_threads && pool->max_threads != -1) + /* Enough threads are already running */ + return; + + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < 0) + { + /* First we try a thread with the right priority */ + g_async_queue_push_unlocked (queue, pool); + success = TRUE; + } + + g_async_queue_unlock (queue); + + if (!success) + { + /* Now we search for threads with other priorities too, but + * only, when there is more than one unused thread with that + * priority. */ + GThreadPriority priority; + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1; priority++) + { + queue = unused_thread_queue[priority]; + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < -1) + { + g_async_queue_push_unlocked (queue, pool); + success = TRUE; + } + + g_async_queue_unlock (queue); + } + } + + if (!success) + /* No thread was found, we have to start one new */ + g_thread_create (g_thread_pool_thread_proxy, pool, pool->pool.stack_size, + FALSE, pool->pool.bound, priority); + + /* See comment in g_thread_pool_thread_proxy as to why this is done + * here and not there */ + pool->num_threads++; +} + +GThreadPool* +g_thread_pool_new (GFunc thread_func, + gint max_threads, + gulong stack_size, + gboolean bound, + GThreadPriority priority, + gboolean exclusive, + gpointer user_data) +{ + GRealThreadPool *retval; + + g_return_val_if_fail (thread_func, NULL); + g_return_val_if_fail (!exclusive || max_threads != -1, NULL); + g_return_val_if_fail (max_threads >= -1, NULL); + g_return_val_if_fail (g_thread_supported (), NULL); + + retval = g_new (GRealThreadPool, 1); + + retval->pool.thread_func = thread_func; + retval->pool.stack_size = stack_size; + retval->pool.bound = bound; + retval->pool.priority = priority; + retval->pool.exclusive = exclusive; + retval->pool.user_data = user_data; + + retval->queue = g_async_queue_new (); + retval->max_threads = max_threads; + retval->num_threads = 0; + retval->running = TRUE; + + if (!inform_mutex) + { + inform_mutex = g_mutex_new (); + inform_cond = g_cond_new (); + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1; priority++) + unused_thread_queue[priority] = g_async_queue_new (); + } + + if (retval->pool.exclusive) + { + g_async_queue_lock (retval->queue); + + while (retval->num_threads < retval->max_threads) + g_thread_pool_start_thread (retval); + + g_async_queue_unlock (retval->queue); + } + + return (GThreadPool*) retval; +} + +void +g_thread_pool_push (GThreadPool *pool, + gpointer data) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + + g_return_if_fail (real); + + g_async_queue_lock (real->queue); + + if (!real->running) + { + g_async_queue_unlock (real->queue); + g_return_if_fail (real->running); + } + + if (!pool->exclusive && g_async_queue_length_unlocked (real->queue) >= 0) + { + /* No thread is waiting in the queue */ + g_thread_pool_start_thread (real); + } + g_async_queue_push_unlocked (real->queue, data); + g_async_queue_unlock (real->queue); +} + +void +g_thread_pool_set_max_threads (GThreadPool *pool, + gint max_threads) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint to_start; + + g_return_if_fail (real); + g_return_if_fail (real->running); + g_return_if_fail (!real->pool.exclusive || max_threads != -1); + g_return_if_fail (max_threads >= -1); + + g_async_queue_lock (real->queue); + + real->max_threads = max_threads; + + if (pool->exclusive) + to_start = real->max_threads - real->num_threads; + else + to_start = g_async_queue_length_unlocked (real->queue); + + for ( ; to_start > 0; to_start--) + g_thread_pool_start_thread (real); + + g_async_queue_unlock (real->queue); +} + +gint +g_thread_pool_get_max_threads (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint retval; + + 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; +} + +guint +g_thread_pool_get_num_threads (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + guint retval; + + 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; +} + +guint +g_thread_pool_unprocessed (GThreadPool *pool) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + gint unprocessed; + + g_return_val_if_fail (real, 0); + g_return_val_if_fail (real->running, 0); + + unprocessed = g_async_queue_length (real->queue); + + return MAX (unprocessed, 0); +} + +void +g_thread_pool_free (GThreadPool *pool, + gboolean immediate, + gboolean wait) +{ + GRealThreadPool *real = (GRealThreadPool*) pool; + + g_return_if_fail (real); + g_return_if_fail (real->running); + /* It there's no thread allowed here, there is not much sense in + * not stopping this pool immediatly, 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); + + real->running = FALSE; + real->immediate = immediate; + real->waiting = wait; + + if (wait) + { + g_mutex_lock (inform_mutex); + while (g_async_queue_length_unlocked (real->queue) != -real->num_threads) + { + g_async_queue_unlock (real->queue); + g_cond_wait (inform_cond, inform_mutex); + g_async_queue_lock (real->queue); + } + g_mutex_unlock (inform_mutex); + } + + if (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 */ + { + g_async_queue_unlock (real->queue); + g_thread_pool_free_internal (real); + return; + } + + g_thread_pool_wakeup_and_stop_all (real); + } + + real->waiting = FALSE; /* The last thread should cleanup the pool */ + g_async_queue_unlock (real->queue); +} + +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->num_threads == 0); + + g_async_queue_unref (pool->queue); + + g_free (pool); +} + +static void +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->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_async_queue_push_unlocked (pool->queue, GUINT_TO_POINTER (1)); +} + +void +g_thread_pool_set_max_unused_threads (gint max_threads) +{ + g_return_if_fail (max_threads >= -1); + + G_LOCK (unused_threads); + + max_unused_threads = max_threads; + + if (max_unused_threads < unused_threads && max_unused_threads != -1) + { + guint close_down_num = unused_threads - max_unused_threads; + GThreadPriority priority; + + while (close_down_num > 0) + { + guint old_close_down_num = close_down_num; + for (priority = G_THREAD_PRIORITY_LOW; + priority < G_THREAD_PRIORITY_URGENT + 1 && close_down_num > 0; + priority++) + { + GAsyncQueue *queue = unused_thread_queue[priority]; + g_async_queue_lock (queue); + + if (g_async_queue_length_unlocked (queue) < 0) + { + g_async_queue_push_unlocked (queue, + stop_this_thread_marker); + close_down_num--; + } + + g_async_queue_unlock (queue); + } + + /* Just to make sure, there are no counting problems */ + g_assert (old_close_down_num != close_down_num); + } + } + + G_UNLOCK (unused_threads); +} + +gint +g_thread_pool_get_max_unused_threads (void) +{ + gint retval; + + G_LOCK (unused_threads); + retval = max_unused_threads; + G_UNLOCK (unused_threads); + + return retval; +} + +guint g_thread_pool_get_num_unused_threads (void) +{ + guint retval; + + G_LOCK (unused_threads); + retval = unused_threads; + G_UNLOCK (unused_threads); + + return retval; +} + +void g_thread_pool_stop_unused_threads (void) +{ + guint oldval = g_thread_pool_get_max_unused_threads (); + g_thread_pool_set_max_unused_threads (0); + g_thread_pool_set_max_unused_threads (oldval); +} diff --git a/tests/.cvsignore b/tests/.cvsignore index 30e3af867..c6169b2e1 100644 --- a/tests/.cvsignore +++ b/tests/.cvsignore @@ -37,5 +37,6 @@ testgdate testgdateparser testglib thread-test +threadpool-test tree-test type-test diff --git a/tests/Makefile.am b/tests/Makefile.am index c1e68cadb..a53473021 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -21,6 +21,7 @@ TESTS = \ strfunc-test \ string-test \ thread-test \ + threadpool-test \ tree-test \ type-test @@ -40,6 +41,8 @@ strfunc_test_LDADD = $(top_builddir)/libglib.la string_test_LDADD = $(top_builddir)/libglib.la thread_test_LDADD = $(top_builddir)/libglib.la \ $(top_builddir)/gthread/libgthread.la @G_THREAD_LIBS@ +threadpool_test_LDADD = $(top_builddir)/libglib.la \ + $(top_builddir)/gthread/libgthread.la @G_THREAD_LIBS@ tree_test_LDADD = $(top_builddir)/libglib.la type_test_LDADD = $(top_builddir)/libglib.la diff --git a/tests/threadpool-test.c b/tests/threadpool-test.c new file mode 100644 index 000000000..9044e4bbd --- /dev/null +++ b/tests/threadpool-test.c @@ -0,0 +1,57 @@ +#include <glib.h> + +#define RUNS 100 + +G_LOCK_DEFINE_STATIC (thread_counter); +gulong abs_thread_counter; +gulong running_thread_counter; + +void +thread_pool_func (gpointer a, gpointer b) +{ + G_LOCK (thread_counter); + abs_thread_counter++; + running_thread_counter++; + G_UNLOCK (thread_counter); + + g_usleep (g_random_int_range (0, 4000)); + + G_LOCK (thread_counter); + running_thread_counter--; + G_UNLOCK (thread_counter); +} + +int +main (int argc, + char *argv[]) +{ + GThreadPool *pool1, *pool2, *pool3; + guint i; + /* Only run the test, if threads are enabled and a default thread + implementation is available */ +#if defined(G_THREADS_ENABLED) && ! defined(G_THREADS_IMPL_NONE) + g_thread_init (NULL); + + pool1 = g_thread_pool_new (thread_pool_func, 3, 0, FALSE, + G_THREAD_PRIORITY_NORMAL, FALSE, NULL); + pool2 = g_thread_pool_new (thread_pool_func, 5, 0, FALSE, + G_THREAD_PRIORITY_LOW, FALSE, NULL); + pool3 = g_thread_pool_new (thread_pool_func, 7, 0, FALSE, + G_THREAD_PRIORITY_LOW, TRUE, NULL); + + for (i = 0; i < RUNS; i++) + { + g_thread_pool_push (pool1, GUINT_TO_POINTER (1)); + g_thread_pool_push (pool2, GUINT_TO_POINTER (1)); + g_thread_pool_push (pool3, GUINT_TO_POINTER (1)); + } + + g_thread_pool_free (pool1, FALSE, TRUE); + g_thread_pool_free (pool2, FALSE, TRUE); + g_thread_pool_free (pool3, FALSE, TRUE); + + g_assert (RUNS * 3 == abs_thread_counter); + g_assert (running_thread_counter == 0); +#endif + return 0; +} |