In non-blocking recv functions in libmux and libdraw,

distinguish between "cannot receive without blocking"
and "EOF on connection".

In libmux, do not elect async guys muxers, so that
synchronous RPC calls run in the main event loop
(e.g., in eresized) do not get stuck.

Fixes problem reported by Lu Xuxiao, namely that
jpg etc. would spin at 100% cpu usage.

7 files changed, 88 insertions, 61 deletions

diff --git a/include/mux.h b/include/mux.h
index 53321fa5..e9890fc3 100644
--- a/include/mux.h
+++ b/include/mux.h
@@ -19,6 +19,7 @@ struct Muxrpc
 	uint tag;
 	void *p;
 	int waiting;
+	int async;
 };
 
 struct Mux
@@ -27,7 +28,7 @@ struct Mux
 	uint maxtag;
 	int (*send)(Mux*, void*);
 	void *(*recv)(Mux*);
-	void *(*nbrecv)(Mux*);
+	int (*nbrecv)(Mux*, void**);
 	int (*gettag)(Mux*, void*);
 	int (*settag)(Mux*, void*, uint);
 	void *aux;	/* for private use by client */
@@ -52,18 +53,18 @@ void	muxinit(Mux*);
 void*	muxrpc(Mux*, void*);
 void	muxprocs(Mux*);
 Muxrpc*	muxrpcstart(Mux*, void*);
-void*	muxrpccanfinish(Muxrpc*);
+int	muxrpccanfinish(Muxrpc*, void**);
 
 /* private */
 int	_muxsend(Mux*, void*);
-void*	_muxrecv(Mux*, int);
+int	_muxrecv(Mux*, int, void**);
 void	_muxsendproc(void*);
 void	_muxrecvproc(void*);
 Muxqueue *_muxqalloc(void);
 int _muxqsend(Muxqueue*, void*);
 void *_muxqrecv(Muxqueue*);
 void _muxqhangup(Muxqueue*);
-void *_muxnbqrecv(Muxqueue*);
+int _muxnbqrecv(Muxqueue*, void**);
 
 #if defined(__cplusplus)
 }
diff --git a/src/cmd/devdraw/x11-init.c b/src/cmd/devdraw/x11-init.c
index bf1262ac..9ad6054f 100644
--- a/src/cmd/devdraw/x11-init.c
+++ b/src/cmd/devdraw/x11-init.c
@@ -47,7 +47,8 @@ static int
 xioerror(XDisplay *d)
 {
 	/*print("X I/O error\n"); */
-	sysfatal("X I/O error\n");
+	exit(0);
+	/*sysfatal("X I/O error\n");*/
 	abort();
 	return -1;
 }
diff --git a/src/libdraw/drawclient.c b/src/libdraw/drawclient.c
index 62615942..93c56235 100644
--- a/src/libdraw/drawclient.c
+++ b/src/libdraw/drawclient.c
@@ -13,7 +13,7 @@ int chattydrawclient;
 
 static int	drawgettag(Mux *mux, void *vmsg);
 static void*	drawrecv(Mux *mux);
-static void*	drawnbrecv(Mux *mux);
+static int	drawnbrecv(Mux *mux, void**);
 static int	drawsend(Mux *mux, void *vmsg);
 static int	drawsettag(Mux *mux, void *vmsg, uint tag);
 static int canreadfd(int);
@@ -83,40 +83,46 @@ drawsend(Mux *mux, void *vmsg)
 	return write(d->srvfd, msg, n);
 }
 
-static void*
-_drawrecv(Mux *mux, int nb)
+static int
+_drawrecv(Mux *mux, int canblock, void **vp)
 {
 	int n;
 	uchar buf[4], *p;
 	Display *d;
 
 	d = mux->aux;
-	if(nb && !canreadfd(d->srvfd))
-		return nil;
+	*vp = nil;
+	if(!canblock && !canreadfd(d->srvfd))
+		return 0;
 	if((n=readn(d->srvfd, buf, 4)) != 4)
-		return nil;
+		return 1;
 	GET(buf, n);
 	p = malloc(n);
 	if(p == nil){
 		fprint(2, "out of memory allocating %d in drawrecv\n", n);
-		return nil;
+		return 1;
 	}
 	memmove(p, buf, 4);
-	if(readn(d->srvfd, p+4, n-4) != n-4)
-		return nil;
-	return p;
+	if(readn(d->srvfd, p+4, n-4) != n-4){
+		free(p);
+		return 1;
+	}
+	*vp = p;
+	return 1;
 }
 
 static void*
 drawrecv(Mux *mux)
 {
-	return _drawrecv(mux, 0);
+	void *p;
+	_drawrecv(mux, 1, &p);
+	return p;
 }
 
-static void*
-drawnbrecv(Mux *mux)
+static int
+drawnbrecv(Mux *mux, void **vp)
 {
-	return _drawrecv(mux, 1);
+	return _drawrecv(mux, 0, vp);
 }
 
 static int
diff --git a/src/libdraw/event.c b/src/libdraw/event.c
index 1da3fb39..101aa374 100644
--- a/src/libdraw/event.c
+++ b/src/libdraw/event.c
@@ -214,10 +214,14 @@ static int
 finishrpc(Muxrpc *r, Wsysmsg *w)
 {
 	uchar *p;
+	void *v;
 	int n;
 	
-	if((p = muxrpccanfinish(r)) == nil)
+	if(!muxrpccanfinish(r, &v))
 		return 0;
+	p = v;
+	if(p == nil)	/* eof on connection */
+		exit(0);
 	GET(p, n);
 	convM2W(p, n, w);
 	free(p);
@@ -269,6 +273,9 @@ extract(int canblock)
 			if(eslave[i].rpc == nil)
 				eslave[i].rpc = startrpc(Trdmouse);
 			if(eslave[i].rpc){
+				/* if ready, don't block in select */
+				if(eslave[i].rpc->p)
+					canblock = 0;
 				FD_SET(display->srvfd, &rset);
 				FD_SET(display->srvfd, &xset);
 				if(display->srvfd > max)
@@ -278,6 +285,9 @@ extract(int canblock)
 			if(eslave[i].rpc == nil)
 				eslave[i].rpc = startrpc(Trdkbd);
 			if(eslave[i].rpc){
+				/* if ready, don't block in select */
+				if(eslave[i].rpc->p)
+					canblock = 0;
 				FD_SET(display->srvfd, &rset);
 				FD_SET(display->srvfd, &xset);
 				if(display->srvfd > max)
diff --git a/src/libmux/io.c b/src/libmux/io.c
index 4a89ca22..d9d9d8a6 100644
--- a/src/libmux/io.c
+++ b/src/libmux/io.c
@@ -34,7 +34,7 @@ _muxrecvproc(void *v)
 	qunlock(&mux->inlk);
 	qlock(&mux->lk);
 	_muxqhangup(q);
-	while((p = _muxnbqrecv(q)) != nil)
+	while(_muxnbqrecv(q, &p))
 		free(p);
 	free(q);
 	mux->readq = nil;
@@ -64,7 +64,7 @@ _muxsendproc(void *v)
 	qunlock(&mux->outlk);
 	qlock(&mux->lk);
 	_muxqhangup(q);
-	while((p = _muxnbqrecv(q)) != nil)
+	while(_muxnbqrecv(q, &p))
 		free(p);
 	free(q);
 	mux->writeq = nil;
@@ -73,42 +73,39 @@ _muxsendproc(void *v)
 	return;
 }
 
-void*
-_muxrecv(Mux *mux, int canblock)
+int
+_muxrecv(Mux *mux, int canblock, void **vp)
 {
 	void *p;
+	int ret;
 
 	qlock(&mux->lk);
-/*
-	if(mux->state != VtStateConnected){
-		werrstr("not connected");
-		qunlock(&mux->lk);
-		return nil;
-	}
-*/
 	if(mux->readq){
 		qunlock(&mux->lk);
-		if(canblock)
-			return _muxqrecv(mux->readq);
-		return _muxnbqrecv(mux->readq);
+		if(canblock){
+			*vp = _muxqrecv(mux->readq);
+			return 1;
+		}
+		return _muxnbqrecv(mux->readq, vp);
 	}
 
 	qlock(&mux->inlk);
 	qunlock(&mux->lk);
-	if(canblock)
+	if(canblock){
 		p = mux->recv(mux);
-	else{
+		ret = 1;
+	}else{
 		if(mux->nbrecv)
-			p = mux->nbrecv(mux);
-		else
+			ret = mux->nbrecv(mux, &p);
+		else{
+			/* send eof, not "no packet ready" */
 			p = nil;
+			ret = 1;
+		}
 	}
 	qunlock(&mux->inlk);
-/*
-	if(!p && canblock)
-		vthangup(mux);
-*/
-	return p;
+	*vp = p;
+	return ret;
 }
 
 int
diff --git a/src/libmux/mux.c b/src/libmux/mux.c
index bfabb238..8257fb0e 100644
--- a/src/libmux/mux.c
+++ b/src/libmux/mux.c
@@ -1,4 +1,4 @@
-/* Copyright (C) 2003 Russ Cox, Massachusetts Institute of Technology */
+/* Copyright (C) 2003-2006 Russ Cox, Massachusetts Institute of Technology */
 /* See COPYRIGHT */
 
 /*
@@ -100,12 +100,17 @@ muxmsgandqlock(Mux *mux, void *p)
 void
 electmuxer(Mux *mux)
 {
+	Muxrpc *rpc;
+
 	/* if there is anyone else sleeping, wake them to mux */
-	if(mux->sleep.next != &mux->sleep){
-		mux->muxer = mux->sleep.next;
-		rwakeup(&mux->muxer->r);
-	}else
-		mux->muxer = nil;
+	for(rpc=mux->sleep.next; rpc != &mux->sleep; rpc = rpc->next){
+		if(!rpc->async){
+			mux->muxer = rpc;
+			rwakeup(&rpc->r);
+			return;
+		}
+	}
+	mux->muxer = nil;
 }
 
 void*
@@ -133,7 +138,7 @@ muxrpc(Mux *mux, void *tx)
 		mux->muxer = r;
 		while(!r->p){
 			qunlock(&mux->lk);
-			p = _muxrecv(mux, 1);
+			_muxrecv(mux, 1, &p);
 			if(p == nil){
 				/* eof -- just give up and pass the buck */
 				qlock(&mux->lk);
@@ -144,7 +149,6 @@ muxrpc(Mux *mux, void *tx)
 		}
 		electmuxer(mux);
 	}
-/*print("finished %p\n", r); */
 	p = r->p;
 	puttag(mux, r);
 	qunlock(&mux->lk);
@@ -161,24 +165,29 @@ muxrpcstart(Mux *mux, void *tx)
 
 	if((r = allocmuxrpc(mux)) == nil)
 		return nil;
+	r->async = 1;
 	if((tag = tagmuxrpc(r, tx)) < 0)
 		return nil;
 	return r;
 }
 
-void*
-muxrpccanfinish(Muxrpc *r)
+int
+muxrpccanfinish(Muxrpc *r, void **vp)
 {
-	char *p;
+	void *p;
 	Mux *mux;
-	
+	int ret;
+
 	mux = r->mux;
 	qlock(&mux->lk);
+	ret = 1;
 	if(!r->p && !mux->muxer){
 		mux->muxer = r;
 		while(!r->p){
 			qunlock(&mux->lk);
-			p = _muxrecv(mux, 0);
+			p = nil;
+			if(!_muxrecv(mux, 0, &p))
+				ret = 0;
 			if(p == nil){
 				qlock(&mux->lk);
 				break;
@@ -191,7 +200,8 @@ muxrpccanfinish(Muxrpc *r)
 	if(p)
 		puttag(mux, r);
 	qunlock(&mux->lk);
-	return p;
+	*vp = p;
+	return ret;
 }
 
 static void
diff --git a/src/libmux/queue.c b/src/libmux/queue.c
index 1cadbe6c..2151c252 100644
--- a/src/libmux/queue.c
+++ b/src/libmux/queue.c
@@ -81,8 +81,8 @@ _muxqrecv(Muxqueue *q)
 	return p;
 }
 
-void*
-_muxnbqrecv(Muxqueue *q)
+int
+_muxnbqrecv(Muxqueue *q, void **vp)
 {
 	void *p;
 	Qel *e;
@@ -90,14 +90,16 @@ _muxnbqrecv(Muxqueue *q)
 	qlock(&q->lk);
 	if(q->head == nil){
 		qunlock(&q->lk);
-		return nil;
+		*vp = nil;
+		return q->hungup;
 	}
 	e = q->head;
 	q->head = e->next;
 	qunlock(&q->lk);
 	p = e->p;
 	free(e);
-	return p;
+	*vp = p;
+	return 1;
 }
 
 void