+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
* 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.
+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
* 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.
+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
* 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.
+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
* 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.
+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
* 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.
+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
* 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.
+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
* 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.
+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
* 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.
libglib_la_SOURCES = \
garray.c \
+ gasyncqueue.c \
gcache.c \
gcompletion.c \
gdataset.c \
gstrfuncs.c \
gstring.c \
gthread.c \
+ gthreadpool.c \
gtimer.c \
gtree.c \
gutils.c
--- /dev/null
+/* 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;
+}
+
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,
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
# 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 */
libglib_la_SOURCES = \
garray.c \
+ gasyncqueue.c \
gcache.c \
gcompletion.c \
gdataset.c \
gstrfuncs.c \
gstring.c \
gthread.c \
+ gthreadpool.c \
gtimer.c \
gtree.c \
gutils.c
--- /dev/null
+/* 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;
+}
+
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,
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
# 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 */
--- /dev/null
+/* 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);
+}
--- /dev/null
+/* 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);
+}
testgdateparser
testglib
thread-test
+threadpool-test
tree-test
type-test
strfunc-test \
string-test \
thread-test \
+ threadpool-test \
tree-test \
type-test
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
--- /dev/null
+#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;
+}
projects / platform / upstream / glib.git / commitdiff