tIn non-blocking recv functions in libmux and libdraw, distinguish between "cannot receive without blocking" and "EOF on connection". - plan9port - [fork] Plan 9 from user space
(HTM) git clone git://src.adamsgaard.dk/plan9port
(DIR) Log
(DIR) Files
(DIR) Refs
(DIR) README
(DIR) LICENSE
---
(DIR) commit 3a19470202c5c0f6e9375e5d57535c3d508f2edf
(DIR) parent d3864abaee496db2f97dff1dbf5c7766d4439c7b
(HTM) Author: rsc <devnull@localhost>
Date: Sat, 4 Nov 2006 18:46:00 +0000
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.
Diffstat:
M include/mux.h | 9 +++++----
M src/cmd/devdraw/x11-init.c | 3 ++-
M src/libdraw/drawclient.c | 34 ++++++++++++++++++-------------
M src/libdraw/event.c | 12 +++++++++++-
M src/libmux/io.c | 43 ++++++++++++++-----------------
M src/libmux/mux.c | 38 +++++++++++++++++++------------
M src/libmux/queue.c | 10 ++++++----
7 files changed, 88 insertions(+), 61 deletions(-)
---
(DIR) diff --git a/include/mux.h b/include/mux.h
t@@ -19,6 +19,7 @@ struct Muxrpc
uint tag;
void *p;
int waiting;
+ int async;
};
struct Mux
t@@ -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 */
t@@ -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)
}
(DIR) diff --git a/src/cmd/devdraw/x11-init.c b/src/cmd/devdraw/x11-init.c
t@@ -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;
}
(DIR) diff --git a/src/libdraw/drawclient.c b/src/libdraw/drawclient.c
t@@ -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);
t@@ -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
(DIR) diff --git a/src/libdraw/event.c b/src/libdraw/event.c
t@@ -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);
t@@ -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)
t@@ -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)
(DIR) diff --git a/src/libmux/io.c b/src/libmux/io.c
t@@ -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;
t@@ -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;
t@@ -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
(DIR) diff --git a/src/libmux/mux.c b/src/libmux/mux.c
t@@ -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 */
/*
t@@ -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*
t@@ -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);
t@@ -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);
t@@ -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;
t@@ -191,7 +200,8 @@ muxrpccanfinish(Muxrpc *r)
if(p)
puttag(mux, r);
qunlock(&mux->lk);
- return p;
+ *vp = p;
+ return ret;
}
static void
(DIR) diff --git a/src/libmux/queue.c b/src/libmux/queue.c
t@@ -81,8 +81,8 @@ _muxqrecv(Muxqueue *q)
return p;
}
-void*
-_muxnbqrecv(Muxqueue *q)
+int
+_muxnbqrecv(Muxqueue *q, void **vp)
{
void *p;
Qel *e;
t@@ -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