Index: camel/camel-private.h
===================================================================
RCS file: /cvs/gnome/evolution-data-server/camel/camel-private.h,v
retrieving revision 1.37
diff -u -p -r1.37 camel-private.h
--- camel/camel-private.h	8 Dec 2005 08:59:32 -0000	1.37
+++ camel/camel-private.h	16 Oct 2006 16:34:05 -0000
@@ -82,7 +82,7 @@ struct _CamelSessionPrivate {
 
 	int thread_id;
 	GHashTable *thread_active;
-	EThread *thread_queue;
+	GThreadPool *thread_pool;
 
 	GHashTable *thread_msg_op;
 };
Index: camel/camel-session.c
===================================================================
RCS file: /cvs/gnome/evolution-data-server/camel/camel-session.c,v
retrieving revision 1.117
diff -u -p -r1.117 camel-session.c
--- camel/camel-session.c	10 Jan 2006 07:56:46 -0000	1.117
+++ camel/camel-session.c	16 Oct 2006 16:34:05 -0000
@@ -76,17 +76,22 @@ camel_session_init (CamelSession *sessio
 	session->priv->thread_lock = g_mutex_new();
 	session->priv->thread_id = 1;
 	session->priv->thread_active = g_hash_table_new(NULL, NULL);
-	session->priv->thread_queue = NULL;
+	session->priv->thread_pool = NULL;
 }
 
 static void
 camel_session_finalise (CamelObject *o)
 {
 	CamelSession *session = (CamelSession *)o;
+	GThreadPool *thread_pool = session->priv->thread_pool;
 	
 	g_hash_table_destroy(session->priv->thread_active);
-	if (session->priv->thread_queue)
-		e_thread_destroy(session->priv->thread_queue);
+
+	if (thread_pool != NULL) {
+		/* there should be no unprocessed tasks */
+		g_assert(g_thread_pool_unprocessed (thread_pool) == 0);
+		g_thread_pool_free(thread_pool, FALSE, FALSE);
+	}
 
 	g_free(session->storage_path);
 	
@@ -515,15 +520,9 @@ static void session_thread_msg_free(Came
 	g_free(msg);
 }
 
-static void session_thread_destroy(EThread *thread, CamelSessionThreadMsg *msg, CamelSession *session)
-{
-	d(printf("destroy message %p session %p\n", msg, session));
-	camel_session_thread_msg_free(session, msg);
-}
-
-static void session_thread_received(EThread *thread, CamelSessionThreadMsg *msg, CamelSession *session)
+static void
+session_thread_proxy(CamelSessionThreadMsg *msg, CamelSession *session)
 {
-	d(printf("receive message %p session %p\n", msg, session));
 	if (msg->ops->receive) {
 		CamelOperation *oldop;
 
@@ -531,22 +530,27 @@ static void session_thread_received(EThr
 		msg->ops->receive(session, msg);
 		camel_operation_register(oldop);
 	}
+
+	camel_session_thread_msg_free(session, msg);
 }
 
 static int session_thread_queue(CamelSession *session, CamelSessionThreadMsg *msg, int flags)
 {
+	GThreadPool *thread_pool;
 	int id;
 
 	CAMEL_SESSION_LOCK(session, thread_lock);
-	if (session->priv->thread_queue == NULL) {
-		session->priv->thread_queue = e_thread_new(E_THREAD_QUEUE);
-		e_thread_set_msg_destroy(session->priv->thread_queue, (EThreadFunc)session_thread_destroy, session);
-		e_thread_set_msg_received(session->priv->thread_queue, (EThreadFunc)session_thread_received, session);
+	thread_pool = session->priv->thread_pool;
+	if (thread_pool == NULL) {
+		thread_pool = g_thread_pool_new (
+			(GFunc) session_thread_proxy,
+			session, 1, FALSE, NULL);
+		session->priv->thread_pool = thread_pool;
 	}
 	CAMEL_SESSION_UNLOCK(session, thread_lock);
 
 	id = msg->id;
-	e_thread_put(session->priv->thread_queue, &msg->msg);
+	g_thread_pool_push(thread_pool, msg, NULL);
 
 	return id;
 }
Index: libedataserver/e-msgport.c
===================================================================
RCS file: /cvs/gnome/evolution-data-server/libedataserver/e-msgport.c,v
retrieving revision 1.11
diff -u -p -r1.11 e-msgport.c
--- libedataserver/e-msgport.c	16 Oct 2006 12:35:48 -0000	1.11
+++ libedataserver/e-msgport.c	16 Oct 2006 16:34:09 -0000
@@ -47,7 +47,6 @@
 #include "e-data-server-util.h"
 
 #define m(x)			/* msgport debug */
-#define t(x)			/* thread debug */
 #define c(x)			/* cache debug */
 
 #ifdef G_OS_WIN32
@@ -751,444 +750,6 @@ e_msgport_reply (EMsg *msg)
 		e_msgport_put (msg->reply_port, msg);
 
 	/* else lost? */
-}
-
-struct _thread_info {
-	pthread_t id;
-	int busy;
-};
-
-struct _EThread {
-	struct _EThread *next;
-	struct _EThread *prev;
-
-	EMsgPort *server_port;
-	EMsgPort *reply_port;
-	pthread_mutex_t mutex;
-	e_thread_t type;
-	int queue_limit;
-
-	int waiting;		/* if we are waiting for a new message, count of waiting processes */
-	pthread_t id;		/* our running child thread */
-	int have_thread;
-	GList *id_list;		/* if THREAD_NEW, then a list of our child threads in thread_info structs */
-
-	EThreadFunc destroy;
-	void *destroy_data;
-
-	EThreadFunc received;
-	void *received_data;
-
-	EThreadFunc lost;
-	void *lost_data;
-};
-
-/* All active threads */
-static EDList ethread_list = E_DLIST_INITIALISER(ethread_list);
-static pthread_mutex_t ethread_lock = PTHREAD_MUTEX_INITIALIZER;
-
-#define E_THREAD_QUIT_REPLYPORT ((struct _EMsgPort *)~0)
-
-static void thread_destroy_msg(EThread *e, EMsg *m);
-
-static struct _thread_info *thread_find(EThread *e, pthread_t id)
-{
-	GList *node;
-	struct _thread_info *info;
-
-	node = e->id_list;
-	while (node) {
-		info = node->data;
-		if (pthread_equal (info->id, id))
-			return info;
-		node = node->next;
-	}
-	return NULL;
-}
-
-#if 0
-static void thread_remove(EThread *e, pthread_t id)
-{
-	GList *node;
-	struct _thread_info *info;
-
-	node = e->id_list;
-	while (node) {
-		info = node->data;
-		if (pthread_equal (info->id, id)) {
-			e->id_list = g_list_remove(e->id_list, info);
-			g_free(info);
-		}
-		node = node->next;
-	}
-}
-#endif
-
-EThread *e_thread_new(e_thread_t type)
-{
-	EThread *e;
-
-	e = g_malloc0(sizeof(*e));
-	pthread_mutex_init(&e->mutex, 0);
-	e->type = type;
-	e->server_port = e_msgport_new();
-	e->have_thread = FALSE;
-	e->queue_limit = INT_MAX;
-
-	pthread_mutex_lock(&ethread_lock);
-	e_dlist_addtail(&ethread_list, (EDListNode *)e);
-	pthread_mutex_unlock(&ethread_lock);
-
-	return e;
-}
-
-/* close down the threads & resources etc */
-void e_thread_destroy(EThread *e)
-{
-	int busy = FALSE;
-	EMsg *msg;
-	struct _thread_info *info;
-	GList *l;
-
-	/* make sure we soak up all the messages first */
-	while ( (msg = e_msgport_get(e->server_port)) ) {
-		thread_destroy_msg(e, msg);
-	}
-
-	pthread_mutex_lock(&e->mutex);
-
-	switch(e->type) {
-	case E_THREAD_QUEUE:
-	case E_THREAD_DROP:
-		/* if we have a thread, 'kill' it */
-		if (e->have_thread) {
-			pthread_t id = e->id;
-			t(printf("Sending thread '%" G_GUINT64_FORMAT "' quit message\n", e_util_pthread_id(id)));
-
-			e->have_thread = FALSE;
-
-			msg = g_malloc0(sizeof(*msg));
-			msg->reply_port = E_THREAD_QUIT_REPLYPORT;
-			e_msgport_put(e->server_port, msg);
-
-			pthread_mutex_unlock(&e->mutex);
-			t(printf("Joining thread '%" G_GUINT64_FORMAT "'\n", e_util_pthread_id(id)));
-			pthread_join(id, 0);
-			t(printf("Joined thread '%" G_GUINT64_FORMAT "'!\n", e_util_pthread_id(id)));
-			pthread_mutex_lock(&e->mutex);
-		}
-		busy = e->have_thread;
-		break;
-	case E_THREAD_NEW:
-		/* first, send everyone a quit message */
-		l = e->id_list;
-		while (l) {
-			info = l->data;
-			t(printf("Sending thread '%" G_GUINT64_FORMAT "' quit message\n", e_util_pthread_id(info->id)));
-			msg = g_malloc0(sizeof(*msg));
-			msg->reply_port = E_THREAD_QUIT_REPLYPORT;
-			e_msgport_put(e->server_port, msg);
-			l = l->next;			
-		}
-
-		/* then, wait for everyone to quit */
-		while (e->id_list) {
-			info = e->id_list->data;
-			e->id_list = g_list_remove(e->id_list, info);
-			pthread_mutex_unlock(&e->mutex);
-			t(printf("Joining thread '%" G_GUINT64_FORMAT "'\n", e_util_pthread_id(info->id)));
-			pthread_join(info->id, 0);
-			t(printf("Joined thread '%" G_GUINT64_FORMAT "'!\n", e_util_pthread_id(info->id)));
-			pthread_mutex_lock(&e->mutex);
-			g_free(info);
-		}
-		busy = g_list_length(e->id_list) != 0;
-		break;
-	}
-
-	pthread_mutex_unlock(&e->mutex);
-
-	/* and clean up, if we can */
-	if (busy) {
-		g_warning("threads were busy, leaked EThread");
-		return;
-	}
-
-	pthread_mutex_lock(&ethread_lock);
-	e_dlist_remove((EDListNode *)e);
-	pthread_mutex_unlock(&ethread_lock);
-
-	pthread_mutex_destroy(&e->mutex);
-	e_msgport_destroy(e->server_port);
-	g_free(e);
-}
-
-/* set the queue maximum depth, what happens when the queue
-   fills up depends on the queue type */
-void e_thread_set_queue_limit(EThread *e, int limit)
-{
-	e->queue_limit = limit;
-}
-
-/* set a msg destroy callback, this can not call any e_thread functions on @e */
-void e_thread_set_msg_destroy(EThread *e, EThreadFunc destroy, void *data)
-{
-	pthread_mutex_lock(&e->mutex);
-	e->destroy = destroy;
-	e->destroy_data = data;
-	pthread_mutex_unlock(&e->mutex);
-}
-
-/* set a message lost callback, called if any message is discarded */
-void e_thread_set_msg_lost(EThread *e, EThreadFunc lost, void *data)
-{
-	pthread_mutex_lock(&e->mutex);
-	e->lost = lost;
-	e->lost_data = lost;
-	pthread_mutex_unlock(&e->mutex);
-}
-
-/* set a reply port, if set, then send messages back once finished */
-void e_thread_set_reply_port(EThread *e, EMsgPort *reply_port)
-{
-	e->reply_port = reply_port;
-}
-
-/* set a received data callback */
-void e_thread_set_msg_received(EThread *e, EThreadFunc received, void *data)
-{
-	pthread_mutex_lock(&e->mutex);
-	e->received = received;
-	e->received_data = data;
-	pthread_mutex_unlock(&e->mutex);
-}
-
-/* find out if we're busy doing any work, e==NULL, check for all work */
-int e_thread_busy(EThread *e)
-{
-	int busy = FALSE;
-
-	if (e == NULL) {
-		pthread_mutex_lock(&ethread_lock);
-		e = (EThread *)ethread_list.head;
-		while (e->next && !busy) {
-			busy = e_thread_busy(e);
-			e = e->next;
-		}
-		pthread_mutex_unlock(&ethread_lock);
-	} else {
-		pthread_mutex_lock(&e->mutex);
-		switch (e->type) {
-		case E_THREAD_QUEUE:
-		case E_THREAD_DROP:
-			busy = e->waiting != 1 && e->have_thread;
-			break;
-		case E_THREAD_NEW:
-			busy = e->waiting != g_list_length(e->id_list);
-			break;
-		}
-		pthread_mutex_unlock(&e->mutex);
-	}
-
-	return busy;
-}
-
-static void
-thread_destroy_msg(EThread *e, EMsg *m)
-{
-	EThreadFunc func;
-	void *func_data;
-
-	/* we do this so we never get an incomplete/unmatched callback + data */
-	pthread_mutex_lock(&e->mutex);
-	func = e->destroy;
-	func_data = e->destroy_data;
-	pthread_mutex_unlock(&e->mutex);
-	
-	if (func)
-		func(e, m, func_data);
-}
-
-static void
-thread_received_msg(EThread *e, EMsg *m)
-{
-	EThreadFunc func;
-	void *func_data;
-
-	/* we do this so we never get an incomplete/unmatched callback + data */
-	pthread_mutex_lock(&e->mutex);
-	func = e->received;
-	func_data = e->received_data;
-	pthread_mutex_unlock(&e->mutex);
-	
-	if (func)
-		func(e, m, func_data);
-	else
-		g_warning("No processing callback for EThread, message unprocessed");
-}
-
-static void
-thread_lost_msg(EThread *e, EMsg *m)
-{
-	EThreadFunc func;
-	void *func_data;
-
-	/* we do this so we never get an incomplete/unmatched callback + data */
-	pthread_mutex_lock(&e->mutex);
-	func = e->lost;
-	func_data = e->lost_data;
-	pthread_mutex_unlock(&e->mutex);
-	
-	if (func)
-		func(e, m, func_data);
-}
-
-/* the actual thread dispatcher */
-static void *
-thread_dispatch(void *din)
-{
-	EThread *e = din;
-	EMsg *m;
-	struct _thread_info *info;
-	pthread_t self = pthread_self();
-
-	t(printf("dispatch thread started: %" G_GUINT64_FORMAT "\n", e_util_pthread_id(self)));
-
-	while (1) {
-		pthread_mutex_lock(&e->mutex);
-		m = e_msgport_get(e->server_port);
-		if (m == NULL) {
-			/* nothing to do?  If we are a 'new' type thread, just quit.
-			   Otherwise, go into waiting (can be cancelled here) */
-			info = NULL;
-			switch (e->type) {
-			case E_THREAD_NEW:
-			case E_THREAD_QUEUE:
-			case E_THREAD_DROP:
-				info = thread_find(e, self);
-				if (info)
-					info->busy = FALSE;
-				e->waiting++;
-				pthread_mutex_unlock(&e->mutex);
-				e_msgport_wait(e->server_port);
-				pthread_mutex_lock(&e->mutex);
-				e->waiting--;
-				pthread_mutex_unlock(&e->mutex);
-				break;
-#if 0
-			case E_THREAD_NEW:
-				e->id_list = g_list_remove(e->id_list, (void *)pthread_self());
-				pthread_mutex_unlock(&e->mutex);
-				return 0;
-#endif
-			}
-
-			continue;
-		} else if (m->reply_port == E_THREAD_QUIT_REPLYPORT) {
-			t(printf("Thread %" G_GUINT64_FORMAT " got quit message\n", e_util_pthread_id(self)));
-			/* Handle a quit message, say we're quitting, free the message, and break out of the loop */
-			info = thread_find(e, self);
-			if (info)
-				info->busy = 2;
-			pthread_mutex_unlock(&e->mutex);
-			g_free(m);
-			break;
-		} else {
-			info = thread_find(e, self);
-			if (info)
-				info->busy = TRUE;
-		}
-		pthread_mutex_unlock(&e->mutex);
-
-		t(printf("got message in dispatch thread\n"));
-
-		/* process it */
-		thread_received_msg(e, m);
-
-		/* if we have a reply port, send it back, otherwise, lose it */
-		if (m->reply_port) {
-			e_msgport_reply(m);
-		} else {
-			thread_destroy_msg(e, m);
-		}
-	}
-
-	return NULL;
-}
-
-/* send a message to the thread, start thread if necessary */
-void e_thread_put(EThread *e, EMsg *msg)
-{
-	pthread_t id;
-	EMsg *dmsg = NULL;
-
-	pthread_mutex_lock(&e->mutex);
-
-	/* the caller forgot to tell us what to do, well, we can't do anything can we */
-	if (e->received == NULL) {
-		pthread_mutex_unlock(&e->mutex);
-		g_warning("EThread called with no receiver function, no work to do!");
-		thread_destroy_msg(e, msg);
-		return;
-	}
-
-	msg->reply_port = e->reply_port;
-
-	switch(e->type) {
-	case E_THREAD_QUEUE:
-		/* if the queue is full, lose this new addition */
-		if (g_async_queue_length(e->server_port->queue) < e->queue_limit) {
-			e_msgport_put(e->server_port, msg);
-		} else {
-			printf("queue limit reached, dropping new message\n");
-			dmsg = msg;
-		}
-		break;
-	case E_THREAD_DROP:
-		/* if the queue is full, lose the oldest (unprocessed) message */
-		if (g_async_queue_length(e->server_port->queue) < e->queue_limit) {
-			e_msgport_put(e->server_port, msg);
-		} else {
-			printf("queue limit reached, dropping old message\n");
-			e_msgport_put(e->server_port, msg);
-			dmsg = e_msgport_get(e->server_port);
-		}
-		break;
-	case E_THREAD_NEW:
-		/* it is possible that an existing thread can catch this message, so
-		   we might create a thread with no work to do.
-		   but that doesn't matter, the other alternative that it be lost is worse */
-		e_msgport_put(e->server_port, msg);
-		if (e->waiting == 0
-		    && g_list_length(e->id_list) < e->queue_limit
-		    && pthread_create(&id, NULL, thread_dispatch, e) == 0) {
-			struct _thread_info *info = g_malloc0(sizeof(*info));
-			t(printf("created NEW thread %" G_GUINT64_FORMAT "\n", e_util_pthread_id(id)));
-			info->id = id;
-			info->busy = TRUE;
-			e->id_list = g_list_append(e->id_list, info);
-		}
-		pthread_mutex_unlock(&e->mutex);
-		return;
-	}
-
-	/* create the thread, if there is none to receive it yet */
-	if (!e->have_thread) {
-		int err;
-
-		if ((err = pthread_create(&e->id, NULL, thread_dispatch, e)) != 0) {
-			g_warning("Could not create dispatcher thread, message queued?: %s", strerror(err));
-		} else {
-			e->have_thread = TRUE;
-		}
-	}
-
-	pthread_mutex_unlock(&e->mutex);
-
-	if (dmsg) {
-		thread_lost_msg(e, dmsg);
-		thread_destroy_msg(e, dmsg);
-	}
 }
 
 /* yet-another-mutex interface */
Index: libedataserver/e-msgport.h
===================================================================
RCS file: /cvs/gnome/evolution-data-server/libedataserver/e-msgport.h,v
retrieving revision 1.4
diff -u -p -r1.4 e-msgport.h
--- libedataserver/e-msgport.h	16 Oct 2006 12:35:48 -0000	1.4
+++ libedataserver/e-msgport.h	16 Oct 2006 16:34:09 -0000
@@ -71,28 +71,6 @@ void e_msgport_reply(EMsg *msg);
 struct PRFileDesc *e_msgport_prfd(EMsgPort *mp);
 #endif
 
-/* e threads, a server thread with a message based request-response, and flexible queuing */
-typedef struct _EThread EThread;
-
-typedef enum {
-	E_THREAD_QUEUE = 0,	/* run one by one, until done, if the queue_limit is reached, discard new request */
-	E_THREAD_DROP,		/* run one by one, until done, if the queue_limit is reached, discard oldest requests */
-	E_THREAD_NEW,		/* always run in a new thread, if the queue limit is reached, new requests are
-				   stored in the queue until a thread becomes available for it, creating a thread pool */
-} e_thread_t;
-
-typedef void (*EThreadFunc)(EThread *, EMsg *, void *data);
-
-EThread *e_thread_new(e_thread_t type);
-void e_thread_destroy(EThread *e);
-void e_thread_set_queue_limit(EThread *e, int limit);
-void e_thread_set_msg_lost(EThread *e, EThreadFunc destroy, void *data);
-void e_thread_set_msg_destroy(EThread *e, EThreadFunc destroy, void *data);
-void e_thread_set_reply_port(EThread *e, EMsgPort *reply_port);
-void e_thread_set_msg_received(EThread *e, EThreadFunc received, void *data);
-void e_thread_put(EThread *e, EMsg *msg);
-int e_thread_busy(EThread *e);
-
 /* sigh, another mutex interface, this one allows different mutex types, portably */
 typedef struct _EMutex EMutex;
 
