New File implementing an asynchronous queue to be used for asynchronous

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.

4 files changed, 922 insertions, 1 deletions

diff --git a/glib/Makefile.am b/glib/Makefile.am
index 3f970a3c7..95221d5ad 100644
--- a/glib/Makefile.am
+++ b/glib/Makefile.am
@@ -35,6 +35,7 @@ lib_LTLIBRARIES = libglib.la
 
 libglib_la_SOURCES = \
 	garray.c	\
+	gasyncqueue.c	\
 	gcache.c	\
 	gcompletion.c	\
 	gdataset.c	\
@@ -58,6 +59,7 @@ libglib_la_SOURCES = \
 	gstrfuncs.c	\
 	gstring.c	\
 	gthread.c       \
+	gthreadpool.c   \
 	gtimer.c	\
 	gtree.c		\
 	gutils.c
diff --git a/glib/gasyncqueue.c b/glib/gasyncqueue.c
new file mode 100644
index 000000000..470d2a78f
--- /dev/null
+++ b/glib/gasyncqueue.c
@@ -0,0 +1,276 @@
+/* GLIB - Library of useful routines for C programming
+ * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
+ *
+ * GAsyncQueue: asyncronous queue implementation, based on Gqueue.
+ * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/*
+ * MT safe
+ */
+
+#include "glib.h"
+
+struct _GAsyncQueue
+{
+  GMutex *mutex;
+  GCond *cond;
+  GQueue *queue;
+  guint waiting_threads;
+  guint ref_count;
+};
+
+GAsyncQueue*
+g_async_queue_new ()
+{
+  GAsyncQueue* retval = g_new (GAsyncQueue, 1);
+  retval->mutex = g_mutex_new ();
+  retval->cond = g_cond_new ();
+  retval->queue = g_queue_new ();
+  retval->waiting_threads = 0;
+  retval->ref_count = 1;
+  return retval;
+}
+
+void 
+g_async_queue_ref (GAsyncQueue *queue)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+  
+  g_mutex_lock (queue->mutex);
+  queue->ref_count++;
+  g_mutex_unlock (queue->mutex);
+}
+
+void 
+g_async_queue_ref_unlocked (GAsyncQueue *queue)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+  
+  queue->ref_count++;
+}
+
+void 
+g_async_queue_unref_and_unlock (GAsyncQueue *queue)
+{
+  gboolean stop;
+
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+
+  queue->ref_count--;
+  stop = (queue->ref_count == 0);
+  g_mutex_unlock (queue->mutex);
+  
+  if (stop)
+    {
+      g_return_if_fail (queue->waiting_threads == 0);
+      g_mutex_free (queue->mutex);
+      g_cond_free (queue->cond);
+      g_queue_free (queue->queue);
+      g_free (queue);
+    }
+}
+
+void 
+g_async_queue_unref (GAsyncQueue *queue)
+{
+  gboolean stop;
+
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+
+  g_mutex_lock (queue->mutex);
+  g_async_queue_unref_and_unlock (queue);
+}
+
+void
+g_async_queue_lock (GAsyncQueue *queue)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+
+  g_mutex_lock (queue->mutex);
+}
+
+void 
+g_async_queue_unlock (GAsyncQueue *queue)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+
+  g_mutex_unlock (queue->mutex);
+}
+
+void
+g_async_queue_push (GAsyncQueue* queue, gpointer data)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+  g_return_if_fail (data);
+
+  g_mutex_lock (queue->mutex);
+  g_async_queue_push_unlocked (queue, data);
+  g_mutex_unlock (queue->mutex);
+}
+
+void
+g_async_queue_push_unlocked (GAsyncQueue* queue, gpointer data)
+{
+  g_return_if_fail (queue);
+  g_return_if_fail (queue->ref_count > 0);
+  g_return_if_fail (data);
+
+  g_queue_push_head (queue->queue, data);
+  g_cond_signal (queue->cond);
+}
+
+static gpointer
+g_async_queue_pop_intern_unlocked (GAsyncQueue* queue, gboolean try, 
+				   GTimeVal *end_time)
+{
+  gpointer retval;
+
+  if (!g_queue_peek_tail (queue->queue))
+    {
+      if (try)
+	return NULL;
+      if (!end_time)
+        {
+          queue->waiting_threads++;
+	  while (!g_queue_peek_tail (queue->queue))
+            g_cond_wait(queue->cond, queue->mutex);
+          queue->waiting_threads--;
+        }
+      else
+        {
+          queue->waiting_threads++;
+          while (!g_queue_peek_tail (queue->queue))
+            if (!g_cond_timed_wait (queue->cond, queue->mutex, end_time))
+              break;
+          queue->waiting_threads--;
+          if (!g_queue_peek_tail (queue->queue))
+	    return NULL;
+        }
+    }
+
+  retval = g_queue_pop_tail (queue->queue);
+
+  g_assert (retval);
+
+  return retval;
+}
+
+gpointer
+g_async_queue_pop (GAsyncQueue* queue)
+{
+  gpointer retval;
+
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  g_mutex_lock (queue->mutex);
+  retval = g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
+  g_mutex_unlock (queue->mutex);
+
+  return retval;
+}
+
+gpointer
+g_async_queue_pop_unlocked (GAsyncQueue* queue)
+{
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  return g_async_queue_pop_intern_unlocked (queue, FALSE, NULL);
+}
+
+gpointer
+g_async_queue_try_pop (GAsyncQueue* queue)
+{
+  gpointer retval;
+
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  g_mutex_lock (queue->mutex);
+  retval = g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
+  g_mutex_unlock (queue->mutex);
+
+  return retval;
+}
+
+gpointer
+g_async_queue_try_pop_unlocked (GAsyncQueue* queue)
+{
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  return g_async_queue_pop_intern_unlocked (queue, TRUE, NULL);
+}
+
+gpointer
+g_async_queue_timed_pop (GAsyncQueue* queue, GTimeVal *end_time)
+{
+  gpointer retval;
+
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  g_mutex_lock (queue->mutex);
+  retval = g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
+  g_mutex_unlock (queue->mutex);
+
+  return retval;  
+}
+
+gpointer
+g_async_queue_timed_pop_unlocked (GAsyncQueue* queue, GTimeVal *end_time)
+{
+  g_return_val_if_fail (queue, NULL);
+  g_return_val_if_fail (queue->ref_count > 0, NULL);
+
+  return g_async_queue_pop_intern_unlocked (queue, FALSE, end_time);
+}
+
+gint
+g_async_queue_length_unlocked (GAsyncQueue* queue)
+{
+  g_return_val_if_fail (queue, 0);
+  g_return_val_if_fail (queue->ref_count > 0, 0);
+
+  return queue->queue->length - queue->waiting_threads;
+}
+
+gint
+g_async_queue_length(GAsyncQueue* queue)
+{
+  glong retval;
+
+  g_return_val_if_fail (queue, 0);
+  g_return_val_if_fail (queue->ref_count > 0, 0);
+
+  g_mutex_lock (queue->mutex);
+  retval = queue->queue->length - queue->waiting_threads;
+  g_mutex_unlock (queue->mutex);
+
+  return retval;
+}
+
diff --git a/glib/glib.h b/glib/glib.h
index 94bb61a1e..799a07a94 100644
--- a/glib/glib.h
+++ b/glib/glib.h
@@ -1075,7 +1075,6 @@ void     g_queue_push_tail_link (GQueue  *queue,
 GList*   g_queue_pop_head_link  (GQueue  *queue);
 GList*   g_queue_pop_tail_link  (GQueue  *queue);
 
-
 /* Hash tables
  */
 GHashTable* g_hash_table_new		(GHashFunc	 hash_func,
@@ -2957,6 +2956,8 @@ typedef struct _GMutex		GMutex;
 typedef struct _GCond		GCond;
 typedef struct _GPrivate	GPrivate;
 typedef struct _GStaticPrivate	GStaticPrivate;
+typedef struct _GAsyncQueue	GAsyncQueue;
+typedef struct _GThreadPool     GThreadPool;
 
 typedef struct _GThreadFunctions GThreadFunctions;
 struct _GThreadFunctions
@@ -3169,6 +3170,128 @@ extern void glib_dummy_decl (void);
 #  define G_TRYLOCK(name)		(TRUE)
 #endif	/* !G_THREADS_ENABLED */
 
+/* Asyncronous Queues, can be used to communicate between threads
+ */
+
+/* Get a new GAsyncQueue with the ref_count 1 */
+GAsyncQueue*  g_async_queue_new                (void);
+
+/* Lock and unlock an GAsyncQueue, all functions lock the queue for
+ * themselves, but in certain cirumstances you want to hold the lock longer,
+ * thus you lock the queue, call the *_unlocked functions and unlock it again
+ */
+void          g_async_queue_lock               (GAsyncQueue *queue);
+void          g_async_queue_unlock             (GAsyncQueue *queue);
+
+/* Ref and unref the GAsyncQueue. g_async_queue_unref_unlocked makes
+ * no sense, as after the unreffing the Queue might be gone and can't
+ * be unlocked. So you have a function to call, if you don't hold the
+ * lock (g_async_queue_unref) and one to call, when you already hold
+ * the lock (g_async_queue_unref_and_unlock). After that however, you
+ * don't hold the lock anymore and the Queue might in fact be
+ * destroyed, if you unrefed to zero */
+void          g_async_queue_ref                (GAsyncQueue *queue);
+void          g_async_queue_ref_unlocked       (GAsyncQueue *queue);
+void          g_async_queue_unref              (GAsyncQueue *queue);
+void          g_async_queue_unref_and_unlock   (GAsyncQueue *queue);
+
+/* Push data into the async queue. Must not be NULL */
+void          g_async_queue_push               (GAsyncQueue *queue,
+						gpointer     data);
+void          g_async_queue_push_unlocked      (GAsyncQueue *queue,
+						gpointer     data);
+
+/* Pop data from the async queue, when no data is there, the thread is blocked
+ * until data arrives */
+gpointer      g_async_queue_pop                (GAsyncQueue *queue);
+gpointer      g_async_queue_pop_unlocked       (GAsyncQueue *queue);
+
+/* Try to pop data, NULL is returned in case of empty queue */
+gpointer      g_async_queue_try_pop            (GAsyncQueue *queue);
+gpointer      g_async_queue_try_pop_unlocked   (GAsyncQueue *queue);
+
+/* Wait for data until at maximum until end_time is reached, NULL is returned
+ * in case of empty queue*/
+gpointer      g_async_queue_timed_pop          (GAsyncQueue *queue, 
+						GTimeVal    *end_time);
+gpointer      g_async_queue_timed_pop_unlocked (GAsyncQueue *queue, 
+						GTimeVal    *end_time);
+
+/* Return the length of the queue, negative values mean, that threads
+ * are waiting, positve values mean, that there are entries in the
+ * queue. Actually this function returns the length of the queue minus
+ * the number of waiting threads, g_async_queue_length == 0 could also
+ * mean 'n' entries in the queue and 'n' thread waiting, such can
+ * happen due to locking of the queue or due to scheduling. */
+gint          g_async_queue_length             (GAsyncQueue *queue);
+gint          g_async_queue_length_unlocked    (GAsyncQueue *queue);
+
+/* Thread Pools
+ */
+
+/* The real GThreadPool is bigger, so you may only create a thread
+ * pool with the constructor function */
+struct _GThreadPool
+{
+  GFunc thread_func;
+  gulong stack_size;
+  gboolean bound; 
+  GThreadPriority priority;
+  gboolean exclusive;
+  gpointer user_data;
+};
+
+/* Get a thread pool with the function thread_func, at most max_threads may
+ * run at a time (max_threads == -1 means no limit), stack_size, bound,
+ * priority like in g_thread_create, exclusive == TRUE means, that the threads
+ * shouldn't be shared and that they will be prestarted (otherwise they are
+ * started, as needed) user_data is the 2nd argument to the thread_func */
+GThreadPool*    g_thread_pool_new             (GFunc            thread_func,
+					       gint             max_threads,
+					       gulong           stack_size,
+					       gboolean         bound,
+					       GThreadPriority  priority,
+					       gboolean         exclusive,
+					       gpointer         user_data);
+
+/* Push new data into the thread pool. This task is assigned to a thread later
+ * (when the maximal number of threads is reached for that pool) or now
+ * (otherwise). If necessary a new thread will be started. The function
+ * returns immediatly */
+void            g_thread_pool_push            (GThreadPool     *pool,
+					       gpointer         data);
+
+/* Set the number of threads, which can run concurrently for that pool, -1
+ * means no limit. 0 means has the effect, that the pool won't process
+ * requests until the limit is set higher again */
+void            g_thread_pool_set_max_threads (GThreadPool     *pool,
+					       gint             max_threads);
+gint            g_thread_pool_get_max_threads (GThreadPool     *pool);
+
+/* Get the number of threads assigned to that pool. This number doesn't
+ * necessarily represent the number of working threads in that pool */
+guint           g_thread_pool_get_num_threads (GThreadPool     *pool);
+
+/* Get the number of unprocessed items in the pool */
+guint           g_thread_pool_unprocessed     (GThreadPool     *pool);
+
+/* Free the pool, immediate means, that all unprocessed items in the queue
+ * wont be processed, wait means, that the function doesn't return immediatly,
+ * but after all threads in the pool are ready processing items. immediate
+ * does however not mean, that threads are killed. */
+void            g_thread_pool_free            (GThreadPool     *pool,
+                                               gboolean         immediate,
+					       gboolean         wait);
+
+/* Set the maximal number of unused threads before threads will be stopped by
+ * GLib, -1 means no limit */
+void            g_thread_pool_set_max_unused_threads (gint      max_threads);
+gint            g_thread_pool_get_max_unused_threads (void);
+guint           g_thread_pool_get_num_unused_threads (void);
+
+/* Stop all currently unused threads, but leave the limit untouched */
+void            g_thread_pool_stop_unused_threads    (void);
+
 #ifdef __cplusplus
 }
 #endif /* __cplusplus */
diff --git a/glib/gthreadpool.c b/glib/gthreadpool.c
new file mode 100644
index 000000000..0ca36444c
--- /dev/null
+++ b/glib/gthreadpool.c
@@ -0,0 +1,520 @@
+/* GLIB - Library of useful routines for C programming
+ * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
+ *
+ * GAsyncQueue: thread pool implementation.
+ * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/*
+ * MT safe
+ */
+
+#include "glib.h"
+
+typedef struct _GRealThreadPool GRealThreadPool;
+
+struct _GRealThreadPool
+{
+  GThreadPool pool;
+  GAsyncQueue* queue;
+  gint max_threads;
+  gint num_threads;
+  gboolean running;
+  gboolean immediate;
+  gboolean waiting;
+};
+
+/* The following is just an address to mark the stop order for a
+ * thread, it could be any address (as long, as it isn;t a valid
+ * GThreadPool address) */
+static const gpointer stop_this_thread_marker = &g_thread_pool_new;
+
+/* Here all unused threads are waiting, depending on their priority */
+static GAsyncQueue *unused_thread_queue[G_THREAD_PRIORITY_URGENT + 1];
+static gint unused_threads = 0;
+static gint max_unused_threads = 0;
+G_LOCK_DEFINE_STATIC (unused_threads);
+
+static GMutex *inform_mutex = NULL;
+static GCond *inform_cond = NULL;
+
+static void g_thread_pool_free_internal (GRealThreadPool* pool);
+static void g_thread_pool_thread_proxy (gpointer data);
+static void g_thread_pool_start_thread (GRealThreadPool* pool);
+static void g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool);
+
+#define g_thread_should_run(pool, len) \
+  ((pool)->running || (!(pool)->immediate && (len) > 0))
+
+static void 
+g_thread_pool_thread_proxy (gpointer data)
+{
+  GRealThreadPool *pool = data;
+  GThread* self = g_thread_self ();
+
+  g_async_queue_lock (pool->queue);
+  while (TRUE)
+    {
+      gpointer task; 
+      gboolean goto_global_pool = !pool->pool.exclusive;
+      gint len = g_async_queue_length_unlocked (pool->queue);
+      
+      if (g_thread_should_run (pool, len))
+	{
+	  task = g_async_queue_pop_unlocked (pool->queue);
+
+	  if (pool->num_threads > pool->max_threads && pool->max_threads != -1)
+	    /* We are in fact a superfluous threads, so we go to the
+	     * global pool and just hand the data further to the next one
+	     * waiting in the queue */
+	    {
+	      g_async_queue_push_unlocked (pool->queue, task);
+	      goto_global_pool = TRUE;
+	    }
+	  else if (pool->running || !pool->immediate)
+	    {
+	      g_async_queue_unlock (pool->queue);
+	      pool->pool.thread_func (task, pool->pool.user_data);
+	      g_async_queue_lock (pool->queue);
+	    }
+
+	  len = g_async_queue_length_unlocked (pool->queue);
+	}
+
+      if (!g_thread_should_run (pool, len))
+	g_cond_broadcast (inform_cond);
+      
+      if (!pool->running && (pool->immediate || len <= 0))
+	goto_global_pool = TRUE;
+      else if (len >= 0)
+	/* At this pool there is no thread waiting */
+	goto_global_pool = FALSE; 
+      
+      if (goto_global_pool)
+	{
+	  GThreadPriority priority = pool->pool.priority;
+	  pool->num_threads--; 
+
+	  if (!pool->running && !pool->waiting)
+	    {
+	      if (pool->num_threads == 0)
+		{
+		  g_async_queue_unlock (pool->queue);
+		  g_thread_pool_free_internal (pool);
+		}		
+	      else if (len == - pool->num_threads)
+		g_thread_pool_wakeup_and_stop_all (pool);
+	    }
+	  else
+	    g_async_queue_unlock (pool->queue);
+	  
+	  g_async_queue_lock (unused_thread_queue[priority]);
+
+	  G_LOCK (unused_threads);
+	  if (unused_threads >= max_unused_threads)
+	    {
+	      G_UNLOCK (unused_threads);
+	      g_async_queue_unlock (unused_thread_queue[priority]);
+	      /* Stop this thread */
+	      return;      
+	    }
+	  unused_threads++;
+	  G_UNLOCK (unused_threads);
+
+	  pool = 
+	    g_async_queue_pop_unlocked (unused_thread_queue[priority]);
+
+	  G_LOCK (unused_threads);
+	  unused_threads--;
+	  G_UNLOCK (unused_threads);
+
+	  g_async_queue_unlock (unused_thread_queue[priority]);
+	  
+	  if (pool == stop_this_thread_marker)
+	    /* Stop this thread */
+	    return;
+	  
+	  g_async_queue_lock (pool->queue);
+
+	  if (pool->pool.priority != self->priority)
+	    g_thread_set_priority (self, pool->pool.priority);
+
+	  /* pool->num_threads++ is not done here, but in
+           * g_thread_pool_start_thread to make the new started thread
+           * known to the pool, before itself can do it. */
+	}
+    }
+}
+
+static void
+g_thread_pool_start_thread (GRealThreadPool* pool)
+{
+  gboolean success = FALSE;
+  GThreadPriority priority = pool->pool.priority;
+  GAsyncQueue *queue = unused_thread_queue[priority];
+  
+  if (pool->num_threads >= pool->max_threads && pool->max_threads != -1)
+    /* Enough threads are already running */
+    return;
+
+  g_async_queue_lock (queue);
+
+  if (g_async_queue_length_unlocked (queue) < 0)
+    {
+      /* First we try a thread with the right priority */
+      g_async_queue_push_unlocked (queue, pool);
+      success = TRUE;
+    }
+
+  g_async_queue_unlock (queue);
+
+  if (!success)
+    {
+      /* Now we search for threads with other priorities too, but
+       * only, when there is more than one unused thread with that
+       * priority. */
+      GThreadPriority priority;
+      for (priority = G_THREAD_PRIORITY_LOW; 
+	   priority < G_THREAD_PRIORITY_URGENT + 1; priority++)
+	{
+	  queue = unused_thread_queue[priority];
+	  g_async_queue_lock (queue);
+	  
+	  if (g_async_queue_length_unlocked (queue) < -1)
+	    {
+	      g_async_queue_push_unlocked (queue, pool);
+	      success = TRUE;
+	    }
+	  
+	  g_async_queue_unlock (queue);
+	}      
+    }
+
+  if (!success)
+    /* No thread was found, we have to start one new */
+    g_thread_create (g_thread_pool_thread_proxy, pool, pool->pool.stack_size, 
+		     FALSE, pool->pool.bound, priority);
+
+  /* See comment in g_thread_pool_thread_proxy as to why this is done
+   * here and not there */
+  pool->num_threads++;
+}
+
+GThreadPool* 
+g_thread_pool_new (GFunc            thread_func,
+		   gint             max_threads,
+		   gulong           stack_size,
+		   gboolean         bound,
+		   GThreadPriority  priority,
+		   gboolean         exclusive,
+		   gpointer         user_data)
+{
+  GRealThreadPool *retval;
+
+  g_return_val_if_fail (thread_func, NULL);
+  g_return_val_if_fail (!exclusive || max_threads != -1, NULL);
+  g_return_val_if_fail (max_threads >= -1, NULL);
+  g_return_val_if_fail (g_thread_supported (), NULL);
+
+  retval = g_new (GRealThreadPool, 1);
+
+  retval->pool.thread_func = thread_func;
+  retval->pool.stack_size = stack_size;
+  retval->pool.bound = bound;
+  retval->pool.priority = priority;
+  retval->pool.exclusive = exclusive;
+  retval->pool.user_data = user_data;
+
+  retval->queue = g_async_queue_new ();
+  retval->max_threads = max_threads;
+  retval->num_threads = 0;
+  retval->running = TRUE;
+
+  if (!inform_mutex)
+    {
+      inform_mutex = g_mutex_new ();
+      inform_cond = g_cond_new ();
+      for (priority = G_THREAD_PRIORITY_LOW; 
+	   priority < G_THREAD_PRIORITY_URGENT + 1; priority++)
+	unused_thread_queue[priority] = g_async_queue_new ();
+    }
+
+  if (retval->pool.exclusive)
+    {
+      g_async_queue_lock (retval->queue);
+
+      while (retval->num_threads < retval->max_threads)
+	g_thread_pool_start_thread (retval);
+
+      g_async_queue_unlock (retval->queue);
+    }
+
+  return (GThreadPool*) retval;
+}
+
+void 
+g_thread_pool_push (GThreadPool     *pool,
+		    gpointer         data)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+
+  g_return_if_fail (real);
+
+  g_async_queue_lock (real->queue);
+  
+  if (!real->running)
+    {
+      g_async_queue_unlock (real->queue);
+      g_return_if_fail (real->running);
+    }
+
+  if (!pool->exclusive && g_async_queue_length_unlocked (real->queue) >= 0)
+    {
+      /* No thread is waiting in the queue */
+      g_thread_pool_start_thread (real);
+    }
+  g_async_queue_push_unlocked (real->queue, data);
+  g_async_queue_unlock (real->queue);
+}
+
+void
+g_thread_pool_set_max_threads (GThreadPool     *pool,
+			       gint             max_threads)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+  gint to_start;
+
+  g_return_if_fail (real);
+  g_return_if_fail (real->running);
+  g_return_if_fail (!real->pool.exclusive || max_threads != -1);
+  g_return_if_fail (max_threads >= -1);
+
+  g_async_queue_lock (real->queue);
+
+  real->max_threads = max_threads;
+  
+  if (pool->exclusive)
+    to_start = real->max_threads - real->num_threads;
+  else
+    to_start = g_async_queue_length_unlocked (real->queue);
+  
+  for ( ; to_start > 0; to_start--)
+    g_thread_pool_start_thread (real);
+    
+  g_async_queue_unlock (real->queue);
+}
+
+gint
+g_thread_pool_get_max_threads (GThreadPool     *pool)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+  gint retval;
+
+  g_return_val_if_fail (real, 0);
+  g_return_val_if_fail (real->running, 0);
+
+  g_async_queue_lock (real->queue);
+
+  retval = real->max_threads;
+    
+  g_async_queue_unlock (real->queue);
+
+  return retval;
+}
+
+guint
+g_thread_pool_get_num_threads (GThreadPool     *pool)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+  guint retval;
+
+  g_return_val_if_fail (real, 0);
+  g_return_val_if_fail (real->running, 0);
+
+  g_async_queue_lock (real->queue);
+
+  retval = real->num_threads;
+    
+  g_async_queue_unlock (real->queue);
+
+  return retval;
+}
+
+guint
+g_thread_pool_unprocessed (GThreadPool     *pool)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+  gint unprocessed;
+
+  g_return_val_if_fail (real, 0);
+  g_return_val_if_fail (real->running, 0);
+
+  unprocessed = g_async_queue_length (real->queue);
+
+  return MAX (unprocessed, 0);
+}
+
+void
+g_thread_pool_free (GThreadPool     *pool,
+		    gboolean         immediate,
+		    gboolean         wait)
+{
+  GRealThreadPool *real = (GRealThreadPool*) pool;
+
+  g_return_if_fail (real);
+  g_return_if_fail (real->running);
+  /* It there's no thread allowed here, there is not much sense in
+   * not stopping this pool immediatly, when it's not empty */
+  g_return_if_fail (immediate || real->max_threads != 0 || 
+		    g_async_queue_length (real->queue) == 0);
+
+  g_async_queue_lock (real->queue);
+
+  real->running = FALSE;
+  real->immediate = immediate;
+  real->waiting = wait;
+
+  if (wait)
+    {
+      g_mutex_lock (inform_mutex);
+      while (g_async_queue_length_unlocked (real->queue) != -real->num_threads)
+	{
+	  g_async_queue_unlock (real->queue); 
+	  g_cond_wait (inform_cond, inform_mutex); 
+	  g_async_queue_lock (real->queue); 
+	}
+      g_mutex_unlock (inform_mutex); 
+    }
+
+  if (g_async_queue_length_unlocked (real->queue) == -real->num_threads)
+    {
+      /* No thread is currently doing something (and nothing is left
+       * to process in the queue) */
+      if (real->num_threads == 0) /* No threads left, we clean up */
+	{
+	  g_async_queue_unlock (real->queue);
+	  g_thread_pool_free_internal (real);
+	  return;
+	}
+
+      g_thread_pool_wakeup_and_stop_all (real);
+    }
+  
+  real->waiting = FALSE; /* The last thread should cleanup the pool */
+  g_async_queue_unlock (real->queue);
+}
+
+static void
+g_thread_pool_free_internal (GRealThreadPool* pool)
+{
+  g_return_if_fail (pool);
+  g_return_if_fail (!pool->running);
+  g_return_if_fail (pool->num_threads == 0);
+
+  g_async_queue_unref (pool->queue);
+
+  g_free (pool);
+}
+
+static void
+g_thread_pool_wakeup_and_stop_all (GRealThreadPool* pool)
+{
+  guint i;
+  
+  g_return_if_fail (pool);
+  g_return_if_fail (!pool->running);
+  g_return_if_fail (pool->num_threads != 0);
+  g_return_if_fail (g_async_queue_length_unlocked (pool->queue) == 
+		    -pool->num_threads);
+
+  pool->immediate = TRUE; 
+  for (i = 0; i < pool->num_threads; i++)
+    g_async_queue_push_unlocked (pool->queue, GUINT_TO_POINTER (1));
+}
+
+void
+g_thread_pool_set_max_unused_threads (gint max_threads)
+{
+  g_return_if_fail (max_threads >= -1);  
+
+  G_LOCK (unused_threads);
+  
+  max_unused_threads = max_threads;
+
+  if (max_unused_threads < unused_threads && max_unused_threads != -1)
+    {
+      guint close_down_num = unused_threads - max_unused_threads;
+      GThreadPriority priority;
+
+      while (close_down_num > 0)
+	{
+	  guint old_close_down_num = close_down_num;
+	  for (priority = G_THREAD_PRIORITY_LOW; 
+	       priority < G_THREAD_PRIORITY_URGENT + 1 && close_down_num > 0; 
+	       priority++)
+	    {
+	      GAsyncQueue *queue = unused_thread_queue[priority];
+	      g_async_queue_lock (queue);
+	      
+	      if (g_async_queue_length_unlocked (queue) < 0)
+		{
+		  g_async_queue_push_unlocked (queue, 
+					       stop_this_thread_marker);
+		  close_down_num--;
+		}
+	      
+	      g_async_queue_unlock (queue);
+	    }
+
+	  /* Just to make sure, there are no counting problems */
+	  g_assert (old_close_down_num != close_down_num);
+	}
+    }
+    
+  G_UNLOCK (unused_threads);
+}
+
+gint
+g_thread_pool_get_max_unused_threads (void)
+{
+  gint retval;
+  
+  G_LOCK (unused_threads);
+  retval = max_unused_threads;
+  G_UNLOCK (unused_threads);
+
+  return retval;
+}
+
+guint g_thread_pool_get_num_unused_threads (void)
+{
+  guint retval;
+  
+  G_LOCK (unused_threads);
+  retval = unused_threads;
+  G_UNLOCK (unused_threads);
+
+  return retval;
+}
+
+void g_thread_pool_stop_unused_threads (void)
+{ 
+  guint oldval = g_thread_pool_get_max_unused_threads ();
+  g_thread_pool_set_max_unused_threads (0);
+  g_thread_pool_set_max_unused_threads (oldval);
+}