luc14n0/glib
1
0
Fork 0
mirror of https://gitlab.gnome.org/GNOME/glib.git synced 2025-02-08 11:55:47 +01:00
Code Issues Packages Projects Releases Wiki Activity

New File implementing an asynchronous queue to be used for asynchronous

Browse Source 
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.
This commit is contained in:
Sebastian Wilhelmi 2000-04-28 12:24:53 +00:00 committed by Sebastian Wilhelmi
parent f0e9f522c2
commit 1418fe3a6d
19 changed files with 2049 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
ChangeLog
View File

@ -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.

19
ChangeLog.pre-2-0
View File

@ -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.

19
ChangeLog.pre-2-10
View File

@ -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.

19
ChangeLog.pre-2-12
View File

@ -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.

19
ChangeLog.pre-2-2
View File

@ -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.

19
ChangeLog.pre-2-4
View File

@ -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.

19
ChangeLog.pre-2-6
View File

@ -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.

19
ChangeLog.pre-2-8
View File

@ -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.

2
Makefile.am
View File

@ -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

276
gasyncqueue.c Normal file
View File

@ -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;
}

125
glib.h
View File

@ -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 */

2
glib/Makefile.am
View File

@ -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

276
glib/gasyncqueue.c Normal file
View File

@ -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;
}

125
glib/glib.h
View File

@ -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 */

520
glib/gthreadpool.c Normal file
View File

@ -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);
}

520
gthreadpool.c Normal file
View File

@ -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);
}

1
tests/.cvsignore
View File

@ -37,5 +37,6 @@ testgdate
testgdateparser
testglib
thread-test
threadpool-test
tree-test
type-test

3
tests/Makefile.am
View File

@ -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

57
tests/threadpool-test.c Normal file
View File

@ -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;
}