From bfa01b256f39d14bdfc22ecb00187ab2dcc4e0fc Mon Sep 17 00:00:00 2001 From: David du Colombier <0intro@gmail.com> Date: Fri, 28 May 1993 00:00:00 +0000 Subject: [PATCH] Plan 9 from Bell Labs 1993-05-28 --- boot/boot.c | 98 ++++++++++++- port/devpipe.c | 258 ++++++++++++++++----------------- port/netif.c | 4 + port/portdat.h | 16 +-- port/portfns.h | 3 + port/proc.c | 2 + port/qio.c | 379 +++++++++++++++++++++---------------------------- port/taslock.c | 2 +- 8 files changed, 396 insertions(+), 366 deletions(-) diff --git a/boot/boot.c b/boot/boot.c index 323bb9dc948a7117dacb2f672dad10571b6e6282..db5d477f11534bef2ace30e1045d4111b1d4d039 100644 --- a/boot/boot.c +++ b/boot/boot.c @@ -24,10 +24,97 @@ static void swapproc(void); static void recover(Method*); static Method *rootserver(char*); +void +pipetest(void) +{ + int i, n, p[2]; + ulong buf[1]; + + if(bind("#c", "/dev", MREPL) < 0) + print("can't bind #c\n"); + + if(pipe(p) < 0){ + print("pipe fails\n"); + exits(0); + } + switch(fork()){ + case -1: + print("fork fails\n"); + exits(0); + case 0: + close(p[0]); + for(;;){ + n = read(p[1], buf, 1); + if(n <= 0){ + print("1 exiting %d\n", time(0)); + exits(0); + } + write(p[1], buf, 1); + } + default: + close(p[1]); + print("start %d\n", time(0)); + for(i = 0; i < 100000; i++){ + if(write(p[0], buf, 1) != 1){ + print("sndr broke\n"); + exits(0); + } + n = read(p[0], buf, 1); + if(n <= 0){ + print("0 exiting %d\n", time(0)); + exits(0); + } + } + close(p[0]); + sleep(1000); + } +} + +void +pipetest2(void) +{ + int i, n, p[2]; + uchar buf[4096]; + + if(bind("#c", "/dev", MREPL) < 0) + print("can't bind #c\n"); + + if(pipe(p) < 0){ + print("pipe fails\n"); + exits(0); + } + switch(fork()){ + case -1: + print("fork fails\n"); + exits(0); + case 0: + close(p[0]); + for(;;){ + n = read(p[1], buf, sizeof(buf)); + if(n <= 0){ + print("1 exiting %d\n", time(0)); + exits(0); + } + } + default: + close(p[1]); + print("start %d\n", time(0)); + for(i = 0; i < 40000; i++){ + if(write(p[0], buf, sizeof(buf)) != sizeof(buf)){ + print("sndr broke\n"); + exits(0); + } + } + close(p[0]); + sleep(1000); + } + +} + void ethertest(void) { - int cf, df, n, t; + int cf, df, n, t, i; char buf[64]; struct Etherpkt { @@ -58,7 +145,7 @@ ethertest(void) exits(0); } close(cf); - for(;;){ + for(i=0;i<20;i++){ n = read(df, &p, sizeof(p)); if(n <= 0){ print("read returns %d: %r\n", n); @@ -67,6 +154,7 @@ ethertest(void) t = (p.type[0]<<8) | p.type[1]; print("%d %2.2ux%2.2ux%2.2ux%2.2ux%2.2ux%2.2ux -> %2.2ux%2.2ux%2.2ux%2.2ux%2.2ux%2.2ux %ux\n", n, p.s[0], p.s[1], p.s[2], p.s[3], p.s[4], p.s[5], p.d[0], p.d[1], p.d[2], p.d[3], p.d[4], p.d[5], t); } + close(df); } void @@ -83,11 +171,15 @@ boot(int argc, char *argv[]) open("#c/cons", OREAD); open("#c/cons", OWRITE); open("#c/cons", OWRITE); +#ifdef DEBUG print("argc=%d\n", argc); for(fd = 0; fd < argc; fd++) print("%s ", argv[fd]); - print("\n");/**/ + print("\n"); +#endif DEBUG + pipetest(); + pipetest2(); ethertest(); if(argc <= 1) diff --git a/port/devpipe.c b/port/devpipe.c index 94a9ad98f68725c2a9836a7e59457c0f89d46c3b..38a4d0a46646f039f414f984d35d7b1112c0cfb7 100644 --- a/port/devpipe.c +++ b/port/devpipe.c @@ -6,14 +6,17 @@ #include "../port/error.h" #include "devtab.h" +#include "netif.h" typedef struct Pipe Pipe; struct Pipe { - Ref; QLock; Pipe *next; + int ref; ulong path; + Queue *q[2]; + int qref[2]; }; struct @@ -23,26 +26,17 @@ struct ulong path; } pipealloc; -static Pipe *getpipe(ulong); -static void pipeiput(Queue*, Block*); -static void pipeoput(Queue*, Block*); -static void pipestclose(Queue *); - -Qinfo pipeinfo = +enum { - pipeiput, - pipeoput, - 0, - pipestclose, - "pipe" + Qdir, + Qdata0, + Qdata1, }; Dirtab pipedir[] = { - "data", {Sdataqid}, 0, 0600, - "ctl", {Sctlqid}, 0, 0600, - "data1", {Sdataqid}, 0, 0600, - "ctl1", {Sctlqid}, 0, 0600, + "data", {Qdata0}, 0, 0600, + "data1", {Qdata1}, 0, 0600, }; #define NPIPEDIR 4 @@ -66,16 +60,33 @@ pipeattach(char *spec) Chan *c; c = devattach('|', spec); - p = smalloc(sizeof(Pipe)); + p = malloc(sizeof(Pipe)); + if(p == 0) + exhausted("memory"); p->ref = 1; + p->q[0] = qopen(64*1024, 0, 0); + if(p->q[0] == 0){ + free(p); + exhausted("memory"); + } + p->q[0]->state &= ~Qmsg; + p->q[1] = qopen(32*1024, 0, 0); + if(p->q[1] == 0){ + free(p->q[0]); + free(p); + exhausted("memory"); + } + p->q[1]->state &= ~Qmsg; + lock(&pipealloc); p->path = ++pipealloc.path; p->next = pipealloc.pipe; pipealloc.pipe = p; unlock(&pipealloc); - c->qid = (Qid){CHDIR|STREAMQID(2*p->path, 0), 0}; + c->qid = (Qid){CHDIR|NETQID(2*p->path, Qdir), 0}; + c->aux = p; c->dev = 0; return c; } @@ -85,10 +96,11 @@ pipeclone(Chan *c, Chan *nc) { Pipe *p; - p = getpipe(STREAMID(c->qid.path)/2); + p = c->aux; nc = devclone(c, nc); - if(incref(p) <= 1) - panic("pipeclone"); + qlock(p); + p->ref++; + qunlock(p); return nc; } @@ -97,13 +109,13 @@ pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp) { int id; - id = STREAMID(c->qid.path); + id = NETID(c->qid.path); if(i > 1) id++; if(tab==0 || i>=ntab) return -1; tab += i; - devdir(c, (Qid){STREAMQID(id, tab->qid.path),0}, tab->name, tab->length, eve, tab->perm, dp); + devdir(c, (Qid){NETQID(id, tab->qid.path),0}, tab->name, tab->length, eve, tab->perm, dp); return 1; } @@ -117,7 +129,25 @@ pipewalk(Chan *c, char *name) void pipestat(Chan *c, char *db) { - streamstat(c, db, "pipe", 0666); + Pipe *p; + Dir dir; + + p = c->aux; + + switch(NETTYPE(c->qid.path)){ + case Qdir: + devdir(c, c->qid, ".", 2*DIRLEN, eve, CHDIR|0555, &dir); + break; + case Qdata0: + devdir(c, c->qid, "data", p->q[0]->len, eve, 0660, &dir); + break; + case Qdata1: + devdir(c, c->qid, "data1", p->q[1]->len, eve, 0660, &dir); + break; + default: + panic("pipestat"); + } + convD2M(&dir, db); } /* @@ -127,8 +157,6 @@ Chan * pipeopen(Chan *c, int omode) { Pipe *p; - int other; - Stream *local, *remote; if(c->qid.path & CHDIR){ if(omode != OREAD) @@ -139,36 +167,17 @@ pipeopen(Chan *c, int omode) return c; } - p = getpipe(STREAMID(c->qid.path)/2); - if(waserror()){ - qunlock(p); - nexterror(); - } + p = c->aux; qlock(p); - streamopen(c, &pipeinfo); - local = c->stream; - if(local->devq->ptr == 0){ - /* - * first open, create the other end also - */ - other = STREAMID(c->qid.path)^1; - remote = streamnew(c->type, c->dev, other, &pipeinfo,1); - - /* - * connect the device ends of both streams - */ - local->devq->ptr = remote; - remote->devq->ptr = local; - local->devq->other->next = remote->devq; - remote->devq->other->next = local->devq; - } else if(local->opens == 1){ - /* - * keep other side around till last close of this side - */ - streamenter(local->devq->ptr); + switch(NETTYPE(c->qid.path)){ + case Qdata0: + p->qref[0]++; + break; + case Qdata1: + p->qref[1]++; + break; } qunlock(p); - poperror(); c->mode = omode&~OTRUNC; c->flag |= COPEN; @@ -201,27 +210,44 @@ void pipeclose(Chan *c) { Pipe *p, *f, **l; - Stream *remote; - p = getpipe(STREAMID(c->qid.path)/2); + p = c->aux; + qlock(p); /* - * take care of local and remote streams + * closing either side hangs up the stream */ - if(c->stream){ - qlock(p); - remote = c->stream->devq->ptr; - if(streamclose(c) == 0){ - if(remote) - streamexit(remote); + switch(NETTYPE(c->qid.path)){ + case Qdata0: + p->qref[0]--; + if(p->qref[0] == 0){ + qclose(p->q[0]); + qhangup(p->q[1]); } - qunlock(p); + break; + case Qdata1: + p->qref[1]--; + if(p->qref[1] == 0){ + qclose(p->q[1]); + qhangup(p->q[0]); + } + break; } /* - * free the structure + * if both sides are closed, they are reusable */ - if(decref(p) == 0){ + if(p->qref[0] == 0 && p->qref[1] == 0){ + qreopen(p->q[0]); + qreopen(p->q[1]); + } + + /* + * free the structure on last close + */ + p->ref--; + if(p->ref == 0){ + qunlock(p); lock(&pipealloc); l = &pipealloc.pipe; for(f = *l; f; f = f->next) { @@ -232,18 +258,34 @@ pipeclose(Chan *c) l = &f->next; } unlock(&pipealloc); + free(p->q[0]); + free(p->q[1]); free(p); } + + qunlock(p); } long piperead(Chan *c, void *va, long n, ulong offset) { + Pipe *p; + USED(offset); - if(c->qid.path & CHDIR) - return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen); - return streamread(c, va, n); + p = c->aux; + + switch(NETTYPE(c->qid.path)){ + case Qdir: + return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen); + case Qdata0: + return qread(p->q[0], va, n); + case Qdata1: + return qread(p->q[1], va, n); + default: + panic("piperead"); + } + return -1; /* not reached */ } /* @@ -253,79 +295,19 @@ piperead(Chan *c, void *va, long n, ulong offset) long pipewrite(Chan *c, void *va, long n, ulong offset) { + Pipe *p; + USED(offset); - /* avoid notes when pipe is a mounted stream */ - if(c->flag & CMSG) - return streamwrite(c, va, n, 0); + p = c->aux; - if(waserror()) { - postnote(up, 1, "sys: write on closed pipe", NUser); - error(Egreg); + switch(NETTYPE(c->qid.path)){ + case Qdata0: + return qwrite(p->q[1], va, n); + case Qdata1: + return qwrite(p->q[0], va, n); + default: + panic("piperead"); } - n = streamwrite(c, va, n, 0); - poperror(); return n; } - -/* - * send a block upstream to the process. - * sleep until there's room upstream. - */ -static void -pipeiput(Queue *q, Block *bp) -{ - FLOWCTL(q, bp); -} - -/* - * send the block to the other side - */ -static void -pipeoput(Queue *q, Block *bp) -{ - PUTNEXT(q, bp); -} - -/* - * send a hangup and disconnect the streams - */ -static void -pipestclose(Queue *q) -{ - Block *bp; - - /* - * point to the bit-bucket and let any in-progress - * write's finish. - */ - q->put = nullput; - wakeup(&q->r); - - /* - * send a hangup - */ - q = q->other; - if(q->next == 0) - return; - bp = allocb(0); - bp->type = M_HANGUP; - PUTNEXT(q, bp); -} - -Pipe* -getpipe(ulong path) -{ - Pipe *p; - - lock(&pipealloc); - for(p = pipealloc.pipe; p; p = p->next) { - if(path == p->path) { - unlock(&pipealloc); - return p; - } - } - unlock(&pipealloc); - panic("getpipe"); - return 0; /* not reached */ -} diff --git a/port/netif.c b/port/netif.c index 0d46b841f0525b4c1afa90a447ce9a9e2c6452b5..704ba7a771cc39a02f6fb14871f7bc525e8ab04e 100644 --- a/port/netif.c +++ b/port/netif.c @@ -269,6 +269,8 @@ netifclose(Netif *nif, Chan *c) qunlock(nif); } f->owner[0] = 0; + f->type = 0; + qclose(f->in); } qunlock(f); } @@ -320,6 +322,7 @@ openfile(Netif *nif, int id) if(f == 0) error(Enodev); qlock(f); + qreopen(f->in); f->inuse++; qunlock(f); return id; @@ -346,6 +349,7 @@ openfile(Netif *nif, int id) } } f->inuse = 1; + qreopen(f->in); netown(f, up->user, 0); qunlock(f); qunlock(nif); diff --git a/port/portdat.h b/port/portdat.h index 575791b40e64ba884ecfadc140d29795fd131e25..28ffc0742266964f34975ce5e09569bf092664f6 100644 --- a/port/portdat.h +++ b/port/portdat.h @@ -656,11 +656,6 @@ struct Block struct Queue { Lock; - QLock rlock; /* mutex for reading processes */ - QLock wlock; /* mutex for writing processes */ - - Block *rfirst; /* waiting readers */ - Block *rlast; Block *bfirst; /* buffer */ Block *blast; @@ -672,7 +667,10 @@ struct Queue void (*kick)(void*); /* restart output */ void *arg; /* argument to kick */ - Rendez r; + QLock rlock; /* mutex for reading processes */ + Rendez rr; /* process waiting to read */ + QLock wlock; /* mutex for writing processes */ + Rendez wr; /* process waiting to write */ }; enum @@ -681,8 +679,10 @@ enum Bfilled=1, /* block filled */ /* Queue.state */ - Qstarve=1, /* consumer starved */ - Qmsg=2, /* message stream */ + Qstarve= (1<<0), /* consumer starved */ + Qmsg= (1<<1), /* message stream */ + Qclosed= (1<<2), + Qflow= (1<<3), }; diff --git a/port/portfns.h b/port/portfns.h index 444ff9266cc1e9df282c548164dbfc834450ec63..8a3fc9f314db3bd8d562acfb804d33dfa4d2e9c2 100644 --- a/port/portfns.h +++ b/port/portfns.h @@ -170,11 +170,14 @@ void putstr(char*); void putstrn(char*, long); void putswap(Page*); ulong pwait(Waitmsg*); +void qclose(Queue*); int qconsume(Queue*, uchar*, int); +void qhangup(Queue*); void qlock(QLock*); Queue* qopen(int, void (*)(void*), void*); int qproduce(Queue*, uchar*, int); long qread(Queue*, char*, int); +void qreopen(Queue*); void qunlock(QLock*); long qwrite(Queue*, char*, int); int readnum(ulong, char*, ulong, ulong, int); diff --git a/port/proc.c b/port/proc.c index 36896cc1607365549c916c53a7547aa4abee8bdd..bf47bf007ac69477f0e8d8c7a72bf9515f965333 100644 --- a/port/proc.c +++ b/port/proc.c @@ -160,6 +160,8 @@ loop: /* p->mach==0 only when process state is saved */ if(p == 0 || p->mach){ unlock(&runhiq); + if(conf.nmach > 1) + delay(7); /* keep off the bus (tuned for everest) */ goto loop; } if(p->rnext == 0) diff --git a/port/qio.c b/port/qio.c index ca99719a791e671fd506f836e22ee531812a46f9..37318161de1fbde3f3714b1ee159727e8ae43279 100644 --- a/port/qio.c +++ b/port/qio.c @@ -150,10 +150,10 @@ iallocb(int size) cl->first = p->next; unlock(cl); b = (Block *)p; + memset(b, 0, sizeof(Block)); b->base = (uchar*)(b+1); b->wp = b->rp = b->base; b->lim = b->base + (1<flag = 0; return b; } panic("iallocb %d\n", size); @@ -186,6 +186,7 @@ allocb(int size) if(b == 0) exhausted("Blocks"); + memset(b, 0, sizeof(Block)); b->base = (uchar*)(b+1); b->rp = b->wp = b->base; b->lim = b->base + size; @@ -202,7 +203,7 @@ int qconsume(Queue *q, uchar *p, int len) { Block *b; - int n; + int n, dowakeup; /* sync with qwrite */ lock(q); @@ -213,6 +214,7 @@ qconsume(Queue *q, uchar *p, int len) unlock(q); return -1; } + n = BLEN(b); if(n < len) len = n; @@ -223,42 +225,34 @@ qconsume(Queue *q, uchar *p, int len) b->rp += len; q->len -= len; - /* wakeup flow controlled writers (with a bit of histeresis) */ - if(q->len+len >= q->limit && q->len < q->limit/2) - wakeup(&q->r); + /* if writer flow controlled, restart */ + if((q->state & Qflow) && q->len < q->limit/2){ + q->state &= ~Qflow; + dowakeup = 1; + } else + dowakeup = 0; unlock(q); + if(dowakeup) + wakeup(&q->wr); + + /* discard the block if we're done with it */ if((q->state & Qmsg) || len == n) ifree(b); return len; } -static int -qproduce0(Queue *q, uchar *p, int len) +int +qproduce(Queue *q, uchar *p, int len) { Block *b; - int n; + int dowakeup; /* sync with qread */ lock(q); - b = q->rfirst; - if(b){ - /* hand to waiting receiver */ - q->rfirst = b->next; - unlock(q); - n = b->lim - b->wp; - if(n < len) - len = n; - memmove(b->wp, p, len); - b->wp += len; - b->flag |= Bfilled; - wakeup(&b->r); - return len; - } - /* no waiting receivers, room in buffer? */ if(q->len >= q->limit){ unlock(q); @@ -279,7 +273,6 @@ qproduce0(Queue *q, uchar *p, int len) } memmove(b->wp, p, len); b->wp += len; - b->flag |= Bfilled; if(q->bfirst) q->blast->next = b; else @@ -287,24 +280,17 @@ qproduce0(Queue *q, uchar *p, int len) q->blast = b; } q->len += len; + if(q->state & Qstarve){ + q->state &= ~Qstarve; + dowakeup = 1; + } else + dowakeup = 0; unlock(q); - return len; -} - -int -qproduce(Queue *q, uchar *p, int len) -{ - int n, sofar; + if(dowakeup) + wakeup(&q->rr); - sofar = 0; - do { - n = qproduce0(q, p + sofar, len - sofar); - if(n < 0) - break; - sofar += n; - } while(sofar < len && (q->state & Qmsg) == 0); - return sofar; + return len; } /* @@ -317,7 +303,9 @@ qopen(int limit, void (*kick)(void*), void *arg) q = malloc(sizeof(Queue)); if(q == 0) - exhausted("Queues"); + return 0; + + memset(q, 0, sizeof(Queue)); q->limit = limit; q->kick = kick; q->arg = arg; @@ -326,248 +314,207 @@ qopen(int limit, void (*kick)(void*), void *arg) return q; } -ulong qrtoomany; -ulong qrtoofew; - static int -bfilled(void *a) +notempty(void *a) { - Block *b = a; + Queue *q = a; - return b->flag & Bfilled; + return q->bfirst != 0; } +/* + * read a queue. if no data is queued, post a Block + * and wait on its Rendez. + */ long qread(Queue *q, char *p, int len) { - Block *b, *bb, **l; - int x, n; + Block *b; + int x, n, dowakeup; qlock(&q->rlock); - b = 0; if(waserror()){ qunlock(&q->rlock); - if(b) - free(b); nexterror(); } - /* - * If there are no buffered blocks, allocate a block - * for the qproducer/qwrite to fill. This is - * optimistic and and we will - * sometimes be wrong: after locking we may either - * have to throw away or allocate one. - * - * We hope to replace the allocb with a kmap later on. - */ -retry: - if(q->bfirst == 0) - b = allocb(len); - - /* sync with qwrite/qproduce */ - x = splhi(); - lock(q); + /* wait for data */ + for(;;){ + /* sync with qwrite/qproduce */ + x = splhi(); + lock(q); - bb = q->bfirst; - if(bb == 0){ - if(b == 0){ - /* we guessed wrong, drop the locks and try again */ + if(q->state & Qclosed){ unlock(q); splx(x); - qrtoofew++; - goto retry; + return 0; } - /* add ourselves to the list of readers */ - if(q->rfirst) - q->rlast->next = b; - else - q->rfirst = b; - q->rlast = b; + b = q->bfirst; + if(b) + break; + q->state |= Qstarve; unlock(q); splx(x); - qunlock(&q->rlock); - poperror(); - - if(waserror()){ - /* on error, unlink us from the chain */ - x = splhi(); - lock(q); - l = &q->rfirst; - for(bb = q->rfirst; bb; bb = bb->next){ - if(b == bb){ - *l = bb->next; - break; - } else - l = &bb->next; - } - unlock(q); - splx(x); - free(b); - nexterror(); - } - - /* wait for the producer */ - sleep(&b->r, bfilled, b); - n = BLEN(b); - memmove(p, b->rp, n); - poperror(); - free(b); - - return n; + sleep(&q->rr, notempty, q); } - /* copy from a buffered block */ - q->bfirst = bb->next; - n = BLEN(bb); - if(n > len) - n = len; + /* remove a buffered block */ + q->bfirst = b->next; + n = BLEN(b); q->len -= n; + + /* if writer flow controlled, restart */ + if((q->state & Qflow) && q->len < q->limit/2){ + q->state &= ~Qflow; + dowakeup = 1; + } else + dowakeup = 0; unlock(q); splx(x); /* do this outside of the lock(q)! */ - memmove(p, bb->rp, n); - bb->rp += n; + if(n > len) + n = len; + memmove(p, b->rp, n); + b->rp += n; - /* free it or put it back on the queue */ - if(bb->rp >= bb->wp || (q->state&Qmsg)) - free(bb); + /* free it or put it what's left on the queue */ + if(b->rp >= b->wp || (q->state&Qmsg)) + free(b); else { x = splhi(); lock(q); - bb->next = q->bfirst; - q->bfirst = bb; + b->next = q->bfirst; + q->bfirst = b; + q->len += BLEN(b); unlock(q); splx(x); } + /* wakeup flow controlled writers (with a bit of histeresis) */ + if(dowakeup) + wakeup(&q->wr); + poperror(); qunlock(&q->rlock); - if(b){ - qrtoomany++; - free(b); - } return n; } -ulong qwtoomany; -ulong qwtoofew; - -static long -qwrite0(Queue *q, char *p, int len, Block *b) +static int +qnotfull(void *a) { - Block *bb; - int x, n, sofar; + Queue *q = a; - /* sync with qconsume/qread */ - x = splhi(); - lock(q); + return q->len < q->limit; +} - sofar = 0; - while(bb = q->rfirst){ - /* hand to waiting receiver */ - q->rfirst = bb->next; - unlock(q); - splx(x); +/* + * write to a queue. if no reader blocks are posted + * queue the data. + */ +long +qwrite(Queue *q, char *p, int len) +{ + int x, dowakeup; + Block *b; - n = bb->lim - bb->wp; - if(n > len-sofar) - n = len - sofar; - memmove(bb->wp, p+sofar, n); - bb->wp += n; - bb->flag |= Bfilled; - wakeup(&bb->r); - - sofar += n; - if(sofar == len){ - if(b){ - free(b); /* we were wrong to allocate */ - qwtoomany++; - } - return len; - } + b = allocb(len); + memmove(b->wp, p, len); + b->wp += len; + + /* flow control */ + while(!qnotfull(q)){ + qlock(&q->wlock); + q->state |= Qflow; + sleep(&q->wr, qnotfull, q); + qunlock(&q->wlock); } - /* buffer what ever is left */ - if(b == 0){ - /* we should have alloc'd, return to qwrite and have it do it */ + x = splhi(); + lock(q); + + if(q->state & Qclosed){ unlock(q); splx(x); - qwtoofew++; - return sofar; + error(Ehungup); } - b->rp += sofar; - x = splhi(); - lock(q); - if(q->bfirst) - q->blast->next = b; - else - q->bfirst = b; - q->blast = b; + b->next = q->bfirst; + q->bfirst = b; q->len += len; - if((q->state & Qstarve) && q->kick){ + + if(q->state & Qstarve){ q->state &= ~Qstarve; - (*q->kick)(q->arg); - } + dowakeup = 1; + } else + dowakeup = 0; + unlock(q); splx(x); + if(dowakeup) + wakeup(&q->rr); + return len; } -static int -qnotfull(void *a) +/* + * Mark a queue as closed. No further IO is permitted. + * All blocks are released. + */ +void +qclose(Queue *q) { - Queue *q = a; + int x; + Block *b, *bfirst; - return q->len < q->limit; -} + /* mark it */ + x = splhi(); + lock(q); + q->state |= Qclosed; + bfirst = q->bfirst; + q->bfirst = 0; + unlock(q); + splx(x); -long -qwrite(Queue *q, char *p, int len) -{ - int n, i; - Block *b; + /* free queued blocks */ + while(b = bfirst){ + bfirst = b->next; + free(b); + } - /* - * If there are no readers, grab a buffer and copy - * into it before locking anything down. This - * provides the highest concurrency but we will - * sometimes be wrong: after locking we may either - * have to throw away or allocate one. - */ - if(q->rfirst == 0){ - b = allocb(len); - memmove(b->wp, p, len); - b->wp += len; - } else - b = 0; + /* wake up readers/writers */ + wakeup(&q->rr); + wakeup(&q->wr); +} - /* ensure atomic writes */ - qlock(&q->wlock); - if(waserror()){ - qunlock(&q->wlock); - nexterror(); - } +/* + * Mark a queue as closed. Wakeup any readers. Don't remove queued + * blocks. + */ +void +qhangup(Queue *q) +{ + int x; - /* flow control */ - sleep(&q->r, qnotfull, q); - - n = qwrite0(q, p, len, b); - if(n != len){ - /* no readers and we need a buffer */ - i = len - n; - b = allocb(i); - memmove(b->wp, p + n, i); - b->wp += n; - n += qwrite0(q, p + n, i, b); - } + /* mark it */ + x = splhi(); + lock(q); + q->state |= Qclosed; + unlock(q); + splx(x); - qunlock(&q->wlock); - poperror(); + /* wake up readers/writers */ + wakeup(&q->rr); + wakeup(&q->wr); +} - return n; +/* + * mark a queue as no longer hung up + */ +void +qreopen(Queue *q) +{ + q->state &= ~Qclosed; } diff --git a/port/taslock.c b/port/taslock.c index ea3226e6107e74ccab2695f0a943b8869fa0925f..ef4364cd14beee384ef75caa5721b2f7a7c4a941 100644 --- a/port/taslock.c +++ b/port/taslock.c @@ -13,7 +13,7 @@ lock(Lock *l) pc = getcallerpc(((uchar*)&l) - sizeof(l)); - for(i = 0; i < 10000000; i++){ + for(i = 0; i < 20000000; i++){ if (tas(&l->key) == 0){ l->pc = pc; return;