Index: libedataserver/e-msgport.c

===================================================================

RCS file: /cvs/gnome/evolution-data-server/libedataserver/e-msgport.c,v

retrieving revision 1.10

diff -u -p -r1.10 e-msgport.c

--- libedataserver/e-msgport.c	26 Jul 2006 18:12:08 -0000	1.10

+++ libedataserver/e-msgport.c	21 Sep 2006 16:49:41 -0000

@@ -486,29 +486,19 @@ em_cache_clear(EMCache *emc)

 }

 struct _EMsgPort {

-	EDList queue;

-	int condwait;		/* how many waiting in condwait */

-	union {

-		int pipe[2];	/* On Win32 actually a pair of SOCKETs */

-		struct {

-			int read;

-			int write;

-		} fd;

-	} pipe;

+	GAsyncQueue *queue;

+	EMsg *cache;

+	gint pipe[2];  /* on Win32, actually a pair of SOCKETs */

 #ifdef HAVE_NSS

-	union {

-		PRFileDesc *pipe[2];

-		struct {

-			PRFileDesc *read;

-			PRFileDesc *write;

-		} fd;

-	} prpipe;

-#endif

-	/* @#@$#$ glib stuff */

-	GCond *cond;

-	GMutex *lock;

+	PRFileDesc *prpipe[2];

+#endif

 };

+/* message flags */

+enum {

+	MSG_FLAG_SYNC_WITH_PIPE    = 1 << 0,

+	MSG_FLAG_SYNC_WITH_PR_PIPE = 1 << 1

+};

 #ifdef HAVE_NSS

 static int

@@ -529,189 +519,237 @@ e_prpipe (PRFileDesc **fds)

 }

 #endif

-EMsgPort *e_msgport_new(void)

+static void

+msgport_sync_with_pipe (gint fd)

+{

+	gchar buffer[1];

+

+	while (fd >= 0) {

+		if (E_READ (fd, buffer, 1) > 0)

+			break;

+		else if (!E_IS_STATUS_INTR ()) {

+			g_warning ("%s: Failed to read from pipe: %s",

+				G_STRFUNC, g_strerror (errno));

+			break;

+		}

+	}

+}

+

+#ifdef HAVE_NSS

+static void

+msgport_sync_with_prpipe (PRFileDesc *prfd)

+{

+	gchar buffer[1];

+

+	while (prfd != NULL) {

+		if (PR_Read (prfd, buffer, 1) > 0)

+			break;

+		else if (PR_GetError () != PR_PENDING_INTERRUPT_ERROR) {

+			gchar *text = g_alloca (PR_GetErrorTextLength ());

+			PR_GetErrorText (text);

+			g_warning ("%s: Failed to read from NSPR pipe: %s",

+				G_STRFUNC, text);

+			break;

+		}

+	}

+}

+#endif

+

+EMsgPort *

+e_msgport_new (void)

 {

-	EMsgPort *mp;

+	EMsgPort *msgport;

-	mp = g_malloc(sizeof(*mp));

-	e_dlist_init(&mp->queue);

-	mp->lock = g_mutex_new();

-	mp->cond = g_cond_new();

-	e_pipe (mp->pipe.pipe);

+	msgport = g_slice_new (EMsgPort);

+	msgport->queue = g_async_queue_new ();

+	msgport->cache = NULL;

+	msgport->pipe[0] = -1;

+	msgport->pipe[1] = -1;

 #ifdef HAVE_NSS

-	e_prpipe (mp->prpipe.pipe);

+	msgport->prpipe[0] = NULL;

+	msgport->prpipe[1] = NULL;

 #endif

-	mp->condwait = 0;

-	return mp;

+	return msgport;

 }

-void e_msgport_destroy(EMsgPort *mp)

+void

+e_msgport_destroy (EMsgPort *msgport)

 {

-	g_mutex_free(mp->lock);

-	g_cond_free(mp->cond);

-	if (mp->pipe.fd.read != -1) {

-		E_CLOSE(mp->pipe.fd.read);

-		E_CLOSE(mp->pipe.fd.write);

+	g_return_if_fail (msgport != NULL);

+

+	if (msgport->pipe[0] >= 0) {

+		E_CLOSE (msgport->pipe[0]);

+		E_CLOSE (msgport->pipe[1]);

 	}

 #ifdef HAVE_NSS

-	if (mp->prpipe.fd.read) {

-		PR_Close(mp->prpipe.fd.read);

-		PR_Close(mp->prpipe.fd.write);

+	if (msgport->prpipe[0] != NULL) {

+		PR_Close (msgport->prpipe[0]);

+		PR_Close (msgport->prpipe[1]);

 	}

 #endif

-	g_free(mp);

+

+	g_async_queue_unref (msgport->queue);

+	g_slice_free (EMsgPort, msgport);

 }

-/* get a fd that can be used to wait on the port asynchronously */

-int e_msgport_fd(EMsgPort *mp)

+int

+e_msgport_fd (EMsgPort *msgport)

 {

-	return mp->pipe.fd.read;

+	gint fd;

+

+	g_return_val_if_fail (msgport != NULL, -1);

+

+	g_async_queue_lock (msgport->queue);

+	fd = msgport->pipe[0];

+	if (fd < 0 && e_pipe (msgport->pipe) == 0)

+		fd = msgport->pipe[0];

+	g_async_queue_unlock (msgport->queue);

+

+	return fd;

 }

 #ifdef HAVE_NSS

-PRFileDesc *e_msgport_prfd(EMsgPort *mp)

+PRFileDesc *

+e_msgport_prfd (EMsgPort *msgport)

 {

-	return mp->prpipe.fd.read;

+	PRFileDesc *prfd;

+

+	g_return_val_if_fail (msgport != NULL, NULL);

+

+	g_async_queue_lock (msgport->queue);

+	prfd = msgport->prpipe[0];

+	if (prfd == NULL && e_prpipe (msgport->prpipe) == 0)

+		prfd = msgport->prpipe[0];

+	g_async_queue_unlock (msgport->queue);

+

+	return prfd;

 }

 #endif

-void e_msgport_put(EMsgPort *mp, EMsg *msg)

+void

+e_msgport_put (EMsgPort *msgport, EMsg *msg)

 {

+	gint fd;

 #ifdef HAVE_NSS

 	PRFileDesc *prfd;

 #endif

-	ssize_t w;

-	int fd;

-	

-	m(printf("put:\n"));

-	g_mutex_lock(mp->lock);

-	e_dlist_addtail(&mp->queue, &msg->ln);

-	if (mp->condwait > 0) {

-		m(printf("put: condwait > 0, waking up\n"));

-		g_cond_signal(mp->cond);

+

+	g_return_if_fail (msgport != NULL);

+	g_return_if_fail (msg != NULL);

+

+	g_async_queue_lock (msgport->queue);

+

+	msg->flags = 0;

+

+	fd = msgport->pipe[1];

+	while (fd >= 0) {

+		if (E_WRITE (fd, "E", 1) > 0) {

+			msg->flags |= MSG_FLAG_SYNC_WITH_PIPE;

+			break;

+		} else if (!E_IS_STATUS_INTR ()) {

+			g_warning ("%s: Failed to write to pipe: %s",

+				G_STRFUNC, g_strerror (errno));

+			break;

+		}

 	}

-	

-	fd = mp->pipe.fd.write;

-#ifdef HAVE_NSS

-	prfd = mp->prpipe.fd.write;

-#endif

-	g_mutex_unlock(mp->lock);

 #ifdef HAVE_NSS

-	if (prfd != NULL) {

-		m(printf("put: have pr pipe, writing notification to it\n"));

-		do {

-			w = PR_Write (prfd, "E", 1);

-		} while (w == -1 && PR_GetError () == PR_PENDING_INTERRUPT_ERROR);

+	prfd = msgport->prpipe[1];

+	while (prfd != NULL) {

+		if (PR_Write (prfd, "E", 1) > 0) {

+			msg->flags |= MSG_FLAG_SYNC_WITH_PR_PIPE;

+			break;

+		} else if (PR_GetError () != PR_PENDING_INTERRUPT_ERROR) {

+			gchar *text = g_alloca (PR_GetErrorTextLength ());

+			PR_GetErrorText (text);

+			g_warning ("%s: Failed to write to NSPR pipe: %s",

+				G_STRFUNC, text);

+			break;

+		}

 	}

 #endif

-	if (fd != -1) {

-		m(printf("put: have pipe, writing notification to it\n"));

-		do {

-			w = E_WRITE (fd, "E", 1);

-		} while (w == -1 && E_IS_STATUS_INTR ());

-	}

-	m(printf("put: done\n"));

+	g_async_queue_push_unlocked (msgport->queue, msg);

+	g_async_queue_unlock (msgport->queue);

 }

-static void

-msgport_cleanlock(void *data)

+EMsg *

+e_msgport_wait (EMsgPort *msgport)

 {

-	EMsgPort *mp = data;

+	EMsg *msg;

-	g_mutex_unlock(mp->lock);

-}

+	g_return_val_if_fail (msgport != NULL, NULL);

-EMsg *e_msgport_wait(EMsgPort *mp)

-{

-	EMsg *msg;

+	g_async_queue_lock (msgport->queue);

-	m(printf("wait:\n"));

-	g_mutex_lock(mp->lock);

-	while (e_dlist_empty(&mp->queue)) {

-		if (mp->pipe.fd.read != -1) {

-			fd_set rfds;

-			int retry;

-

-			m(printf("wait: waiting on pipe\n"));

-			g_mutex_unlock(mp->lock);

-			do {

-				FD_ZERO(&rfds);

-				FD_SET(mp->pipe.fd.read, &rfds);

-				retry = E_IS_SOCKET_ERROR(select(mp->pipe.fd.read+1, &rfds, NULL, NULL, NULL)) && E_IS_STATUS_INTR();

-				pthread_testcancel();

-			} while (retry);

-			g_mutex_lock(mp->lock);

-			m(printf("wait: got pipe\n"));

-#ifdef HAVE_NSS

-		} else if (mp->prpipe.fd.read != NULL) {

-			PRPollDesc rfds[1];

-			int retry;

-

-			m(printf("wait: waitng on pr pipe\n"));

-			g_mutex_unlock(mp->lock);

-			do {

-				rfds[0].fd = mp->prpipe.fd.read;

-				rfds[0].in_flags = PR_POLL_READ|PR_POLL_ERR;

-				retry = PR_Poll(rfds, 1, PR_INTERVAL_NO_TIMEOUT) == -1 && PR_GetError() == PR_PENDING_INTERRUPT_ERROR;

-				pthread_testcancel();

-			} while (retry);

-			g_mutex_lock(mp->lock);

-			m(printf("wait: got pr pipe\n"));

-#endif /* HAVE_NSS */

-		} else {

-			m(printf("wait: waiting on condition\n"));

-			mp->condwait++;

-			/* if we are cancelled in the cond-wait, then we need to unlock our lock when we cleanup */

-			pthread_cleanup_push(msgport_cleanlock, mp);

-			g_cond_wait(mp->cond, mp->lock);

-			pthread_cleanup_pop(0);

-			m(printf("wait: got condition\n"));

-			mp->condwait--;

-		}

+	/* check the cache first */

+	if (msgport->cache != NULL) {

+		msg = msgport->cache;

+		/* don't clear the cache */

+		g_async_queue_unlock (msgport->queue);

+		return msg;

 	}

-	msg = (EMsg *)mp->queue.head;

-	m(printf("wait: message = %p\n", msg));

-	g_mutex_unlock(mp->lock);

-	m(printf("wait: done\n"));

+

+	msg = g_async_queue_pop_unlocked (msgport->queue);

+

+	g_assert (msg != NULL);

+

+	/* The message is not actually "removed" from the EMsgPort until

+ 	 * e_msgport_get() is called.  So we cache the popped message. */

+	msgport->cache = msg;

+

+	if (msg->flags & MSG_FLAG_SYNC_WITH_PIPE)

+		msgport_sync_with_pipe (msgport->pipe[0]);

+#ifdef HAVE_NSS

+	if (msg->flags & MSG_FLAG_SYNC_WITH_PR_PIPE)

+		msgport_sync_with_prpipe (msgport->prpipe[0]);

+#endif

+

+	g_async_queue_unlock (msgport->queue);

+

 	return msg;

 }

-EMsg *e_msgport_get(EMsgPort *mp)

+EMsg *

+e_msgport_get (EMsgPort *msgport)

 {

 	EMsg *msg;

-	char dummy[1];

-	ssize_t n;

-	

-	g_mutex_lock(mp->lock);

-	msg = (EMsg *)e_dlist_remhead(&mp->queue);

-	if (msg) {

-		if (mp->pipe.fd.read != -1) {

-			do {

-				n = E_READ (mp->pipe.fd.read, dummy, 1);

-			} while (n == -1 && E_IS_STATUS_INTR ());

-		}

+

+	g_return_val_if_fail (msgport != NULL, NULL);

+

+	g_async_queue_lock (msgport->queue);

+

+	/* check the cache first */

+	if (msgport->cache != NULL) {

+		msg = msgport->cache;

+		msgport->cache = NULL;

+		g_async_queue_unlock (msgport->queue);

+		return msg;

+	}

+

+	msg = g_async_queue_try_pop_unlocked (msgport->queue);

+

+	if (msg != NULL && msg->flags & MSG_FLAG_SYNC_WITH_PIPE)

+		msgport_sync_with_pipe (msgport->pipe[0]);

 #ifdef HAVE_NSS

-		if (mp->prpipe.fd.read != NULL) {

-			do {

-				n = PR_Read (mp->prpipe.fd.read, dummy, 1);

-			} while (n == -1 && PR_GetError () == PR_PENDING_INTERRUPT_ERROR);

-		}

+	if (msg != NULL && msg->flags & MSG_FLAG_SYNC_WITH_PR_PIPE)

+		msgport_sync_with_prpipe (msgport->prpipe[0]);

 #endif

-	}

-	m(printf("get: message = %p\n", msg));

-	g_mutex_unlock(mp->lock);

+

+	g_async_queue_unlock (msgport->queue);

 	return msg;

 }

-void e_msgport_reply(EMsg *msg)

+void

+e_msgport_reply (EMsg *msg)

 {

-	if (msg->reply_port) {

-		e_msgport_put(msg->reply_port, msg);

-	}

+	g_return_if_fail (msg != NULL);

+

+	if (msg->reply_port)

+		e_msgport_put (msg->reply_port, msg);

+

 	/* else lost? */

 }

@@ -1099,7 +1137,7 @@ void e_thread_put(EThread *e, EMsg *msg)

 	switch(e->type) {

 	case E_THREAD_QUEUE:

 		/* if the queue is full, lose this new addition */

-		if (e_dlist_length(&e->server_port->queue) < e->queue_limit) {

+		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");

@@ -1108,7 +1146,7 @@ void e_thread_put(EThread *e, EMsg *msg)

 		break;

 	case E_THREAD_DROP:

 		/* if the queue is full, lose the oldest (unprocessed) message */

-		if (e_dlist_length(&e->server_port->queue) < e->queue_limit) {

+		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");

Index: libedataserver/e-msgport.h

===================================================================

RCS file: /cvs/gnome/evolution-data-server/libedataserver/e-msgport.h,v

retrieving revision 1.3

diff -u -p -r1.3 e-msgport.h

--- libedataserver/e-msgport.h	3 Dec 2004 03:33:06 -0000	1.3

+++ libedataserver/e-msgport.h	21 Sep 2006 16:49:41 -0000

@@ -56,6 +56,7 @@ typedef struct _EMsgPort EMsgPort;

 typedef struct _EMsg {

 	EDListNode ln;

 	EMsgPort *reply_port;

+	gint flags;

 } EMsg;

 EMsgPort *e_msgport_new(void);