From e6a31db53a342e6cf0431b5c8a547b195ba9b0da Mon Sep 17 00:00:00 2001 From: David du Colombier <0intro@gmail.com> Date: Tue, 9 Oct 1990 00:00:00 +0000 Subject: [PATCH] Plan 9 from Bell Labs 1990-10-09 --- gnot/dat.h | 14 +- gnot/devpipe.c | 204 +++++++++++++++++----- gnot/errno.h | 5 +- gnot/fns.h | 3 +- gnot/stasync.c | 2 +- gnot/stream.c | 50 +++--- gnot/sysfile.c | 82 ++++----- gnot/sysproc.c | 2 + port/chan.c | 2 +- port/devpipe.c | 204 +++++++++++++++++----- port/stasync.c | 453 +++++++++++++++++++++++++++++++++++++++++++++++++ port/stream.c | 50 +++--- port/sysfile.c | 86 +++++----- port/sysproc.c | 2 + power/conf.h | 2 + power/dat.h | 15 +- power/errno.h | 2 + power/fns.h | 3 +- power/main.c | 1 + 19 files changed, 940 insertions(+), 242 deletions(-) create mode 100644 port/stasync.c diff --git a/gnot/dat.h b/gnot/dat.h index 7e63bd79415a7c1c9fcfc93e3dff97b89e24f58f..89be61df5370aceacc43fe45ebc2f189b4f7314c 100644 --- a/gnot/dat.h +++ b/gnot/dat.h @@ -150,6 +150,7 @@ struct Conf int nfont; /* font structs (devbit.c) */ int nurp; /* max urp conversations */ int nasync; /* number of async protocol modules */ + int npipe; /* number of pipes */ }; struct Dev @@ -457,14 +458,13 @@ struct Queue { */ struct Stream { Lock; /* structure lock */ - int inuse; /* number of processes in stream */ - int opens; /* number of processes with stream open */ - int hread; /* number of reads after hangup */ - int type; /* correclation with Chan */ - int dev; /* ... */ - int id; /* ... */ + short inuse; /* number of processes in stream */ + short opens; /* number of processes with stream open */ + ushort hread; /* number of reads after hangup */ + ushort type; /* correlation with Chan */ + ushort dev; /* ... */ + ushort id; /* ... */ QLock rdlock; /* read lock */ - QLock wrlock; /* write lock */ Queue *procq; /* write queue at process end */ Queue *devq; /* read queue at device end */ }; diff --git a/gnot/devpipe.c b/gnot/devpipe.c index 08a1ec1065412cb13fd34ba5a974b2589f37aed6..2183aae2fb553544d5f09b499932704a585539df 100644 --- a/gnot/devpipe.c +++ b/gnot/devpipe.c @@ -8,80 +8,111 @@ #include "devtab.h" #include "fcall.h" +typedef struct Pipe Pipe; + +struct Pipe +{ + Ref; + int debug; + Pipe *next; +}; + +struct Pipealloc +{ + Lock; + Pipe *pipe; + Pipe *free; +} pipealloc; + static void pipeiput(Queue*, Block*); static void pipeoput(Queue*, Block*); static void pipestclose(Queue *); Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "pipe" }; +Dirtab pipedir[]={ + "data", Sdataqid, 0, 0600, + "ctl", Sctlqid, 0, 0600, + "data1", Sdataqid, 0, 0600, + "ctl1", Sctlqid, 0, 0600, +}; +#define NPIPEDIR 4 + void pipeinit(void) { } +/* + * allocate structures for conf.npipe pipes + */ void pipereset(void) { + Pipe *p, *ep; + + pipealloc.pipe = ialloc(conf.npipe * sizeof(Pipe), 0); + ep = &pipealloc.pipe[conf.npipe-1]; + for(p = pipealloc.pipe; p < ep; p++) + p->next = p+1; + pipealloc.free = pipealloc.pipe; } /* - * allocate both streams - * - * a subsequent clone will get them the second stream + * create a pipe, no streams are created until an open */ Chan* pipeattach(char *spec) { + Pipe *p; Chan *c; - int i; - /* - * make the first stream - */ c = devattach('|', spec); - c->qid = STREAMQID(0, Sdataqid); - streamnew(c, &pipeinfo); + + lock(&pipealloc); + if(pipealloc.free == 0){ + unlock(&pipealloc); + error(0, Enopipe); + } + p = pipealloc.free; + pipealloc.free = p->next; + p->ref = 1; + unlock(&pipealloc); + + c->qid = CHDIR|STREAMQID(2*(p - pipealloc.pipe), 0); return c; } Chan* pipeclone(Chan *c, Chan *nc) { - /* - * make the second stream - */ + Pipe *p; + + p = &pipealloc.pipe[STREAMID(c->qid)/2]; nc = devclone(c, nc); - if(waserror()){ - close(nc); - nexterror(); - } - nc->qid = STREAMQID(1, Sdataqid); - streamnew(nc, &pipeinfo); - poperror(); + incref(p); + return nc; +} - /* - * attach it to the first - */ - c->stream->devq->ptr = (Stream *)nc->stream; - nc->stream->devq->ptr = (Stream *)c->stream; - c->stream->devq->other->next = nc->stream->devq; - nc->stream->devq->other->next = c->stream->devq; +int +pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp) +{ + int id; - /* - * up the inuse count of each stream to reflect the - * pointer from the other stream. - */ - if(streamenter(c->stream)<0) - panic("pipeattach"); - if(streamenter(nc->stream)<0) - panic("pipeattach"); - return nc; + id = STREAMID(c->qid); + if(i > 1) + id++; + if(tab==0 || i>=ntab) + return -1; + tab += i; + devdir(c, STREAMQID(id, tab->qid), tab->name, tab->length, tab->perm, dp); + return 1; } + int pipewalk(Chan *c, char *name) { - print("pipewalk\n"); - error(0, Egreg); + return devwalk(c, name, pipedir, NPIPEDIR, pipegen); } void @@ -90,10 +121,60 @@ pipestat(Chan *c, char *db) streamstat(c, db, "pipe"); } +/* + * if the stream doesn't exist, create it + */ Chan * pipeopen(Chan *c, int omode) { - c->mode = omode; + Pipe *p; + Stream *local, *remote; + + if(CHDIR & c->qid){ + if(omode != OREAD) + error(0, Ebadarg); + c->mode = omode; + c->flag |= COPEN; + c->offset = 0; + return c; + } + + p = &pipealloc.pipe[STREAMID(c->qid)/2]; + remote = 0; + if(waserror()){ + unlock(p); + if(remote) + streamclose1(remote); + nexterror(); + } + lock(p); + streamopen(c, &pipeinfo); + local = c->stream; + if(local->devq->ptr == 0){ + /* + * First stream opened, create the other end also + */ + remote = streamnew(c->type, c->dev, STREAMID(c->qid)^1, &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; + + /* + * increment the inuse count to reflect the + * pointer from the other stream. + */ + if(streamenter(local)<0) + panic("pipeattach"); + } + unlock(p); + poperror(); + + c->mode = omode&~OTRUNC; c->flag |= COPEN; c->offset = 0; return c; @@ -117,28 +198,57 @@ pipewstat(Chan *c, char *db) error(0, Eperm); } +void +pipeexit(Pipe *p) +{ + decref(p); + if(p->ref <= 0){ + lock(&pipealloc); + p->next = pipealloc.free; + pipealloc.free = p; + unlock(&pipealloc); + } +} + void pipeclose(Chan *c) { - Stream *other; + Stream *remote; + Stream *local; + Pipe *p; - other = (Stream *)c->stream->devq->ptr; + p = &pipealloc.pipe[STREAMID(c->qid)/2]; - if(waserror()){ - streamexit(other, 0); - nexterror(); + /* + * take care of assosiated streams + */ + if(local = c->stream){ + remote = (Stream *)c->stream->devq->ptr; + if(waserror()){ + streamexit(remote, 0); + pipeexit(p); + nexterror(); + } + streamclose(c); /* close this stream */ + streamexit(remote, 0); /* release stream for other half of pipe */ + poperror(); } - streamclose(c); /* close this stream */ - streamexit(other, 0); /* release stream for other half of pipe */ - poperror(); + pipeexit(p); } long piperead(Chan *c, void *va, long n) { - return streamread(c, va, n); + if(CHDIR&c->qid) + return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen); + else + return streamread(c, va, n); } +/* + * a write to a closed pipe causes a note to be sent to + * the process. + */ long pipewrite(Chan *c, void *va, long n) { diff --git a/gnot/errno.h b/gnot/errno.h index 3f4ddd9a2aaf5d00b8ea12fb9ed98d3a06c031f2..2c6e33643e83e423d382065afbaf79f765b43bb3 100644 --- a/gnot/errno.h +++ b/gnot/errno.h @@ -36,9 +36,9 @@ enum{ Etoosmall, /* read or write too small */ Ehungup, /* write to hungup stream */ Ebadnet, /* illegal network address */ - Enoifc, /* no free interface slots */ + Enoifc, /* bad interface or no free interface slots */ Enodev, /* no free devices */ - Ebadctl, /* bad process control request */ + Ebadctl, /* bad process or stream control request */ Enonote, /* note overflow */ Eintr, /* interrupted */ Edestbusy, /* datakit destination busy */ @@ -57,5 +57,6 @@ enum{ Enofont, /* out of font descriptors */ Enovmem, /* virtual memory allocation failed */ Enoasync, /* out of async stream modules */ + Enopipe, /* out of pipes */ Egreg, /* ken hasn't implemented datakit */ }; diff --git a/gnot/fns.h b/gnot/fns.h index f458eb5cd46284c42e479d6041ddfb5280caad11..4f56834e6e923649e0766f3fefdc2fd001cb5995 100644 --- a/gnot/fns.h +++ b/gnot/fns.h @@ -153,12 +153,13 @@ void splx(int); void spldone(void); Devgen streamgen; void streamclose(Chan*); +void streamclose1(Stream*); int streamenter(Stream*); void streamexit(Stream*, int); void streaminit(void); long streamread(Chan*, void*, long); long streamwrite(Chan*, void*, long, int); -Stream* streamnew(Chan*, Qinfo*); +Stream* streamnew(ushort, ushort, ushort, Qinfo*, int); void streamopen(Chan*, Qinfo*); int streamparse(char*, Block*); void streamstat(Chan*, char*, char*); diff --git a/gnot/stasync.c b/gnot/stasync.c index 52d41118a20eda2ca8e06700c371a182f3558fca..a4d5d19e4bcc3c65d602a83ad07bdffeafeeb6ba 100644 --- a/gnot/stasync.c +++ b/gnot/stasync.c @@ -61,7 +61,7 @@ static void asyncclose(Queue*); static void asyncreset(void); Qinfo asyncinfo = { asynciput, asyncoput, asyncopen, asyncclose, "async", asyncreset }; -int asyncdebug = 10; +int asyncdebug = 0; int asyncerror; static ushort crc_table[256] = { diff --git a/gnot/stream.c b/gnot/stream.c index a9f2201b21d78ea6ebfe254b970671167fa8c429..a8eaacb96529349a3386d9420723f409e15d0944 100644 --- a/gnot/stream.c +++ b/gnot/stream.c @@ -235,6 +235,7 @@ allocq(Qinfo *qi) q->info = qi; q->put = qi->iput; q->len = q->nb = 0; + q->ptr = 0; wq = q->other = q + 1; wq->flag = QINUSE; @@ -242,6 +243,7 @@ allocq(Qinfo *qi) wq->info = qi; wq->put = qi->oput; wq->other = q; + wq->ptr = 0; wq->len = wq->nb = 0; unlock(q); @@ -625,10 +627,10 @@ streamgen(Chan *c, Dirtab *tab, int ntab, int s, Dir *dp) } /* - * create a new stream + * create a new stream, if noopen is non-zero, don't increment the open count */ Stream * -streamnew(Chan *c, Qinfo *qi) +streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen) { Stream *s; Queue *q; @@ -651,26 +653,25 @@ streamnew(Chan *c, Qinfo *qi) } if(waserror()){ unlock(s); - streamclose(c); + streamclose1(s); nexterror(); } /* - * marry a stream and a channel + * identify the stream */ - if(c){ - c->stream = s; - s->type = c->type; - s->dev = c->dev; - s->id = STREAMID(c->qid); - } else - s->type = -1; + s->type = type; + s->dev = dev; + s->id = id; /* * hang a device and process q off the stream */ s->inuse = 1; - s->opens = 1; + if(noopen) + s->opens = 0; + else + s->opens = 1; s->hread = 0; q = allocq(&procinfo); s->procq = WR(q); @@ -699,7 +700,7 @@ streamopen(Chan *c, Qinfo *qi) Queue *q; /* - * if the stream already exists, just up the reference count. + * if the stream already exists, just increment the reference counts. */ for(s = slist; s < &slist[conf.nstream]; s++) { if(s->inuse && s->type == c->type && s->dev == c->dev @@ -721,7 +722,7 @@ streamopen(Chan *c, Qinfo *qi) /* * create a new stream */ - streamnew(c, qi); + c->stream = streamnew(c->type, c->dev, STREAMID(c->qid), qi, 0); } /* @@ -773,17 +774,10 @@ streamexit(Stream *s, int locked) * stream release its blocks and call its close routine. */ void -streamclose(Chan *c) +streamclose1(Stream *s) { Queue *q, *nq; Block *bp; - Stream *s = c->stream; - - /* - * if not open, ignore it - */ - if(!c->stream) - return; /* * decrement the reference count @@ -817,6 +811,16 @@ streamclose(Chan *c) streamexit(s, 1); unlock(s); } +void +streamclose(Chan *c) +{ + /* + * if no stream, ignore it + */ + if(!c->stream) + return; + streamclose1(c->stream); +} /* * put a block to be read into the queue. wakeup any waiting reader @@ -1174,7 +1178,7 @@ dumpblocks(Queue *q, char c) lock(q); for(bp = q->first; bp; bp = bp->next){ - print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)); + print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' '); for(cp = bp->rptr; cpwptr && cprptr+10; cp++) print(" %uo", *cp); print("\n"); diff --git a/gnot/sysfile.c b/gnot/sysfile.c index 231b7ea5600a596318e6a0663d32feefb5b3a909..c45fbe353aed5cb48bcd41d4dbd6e784b931556a 100644 --- a/gnot/sysfile.c +++ b/gnot/sysfile.c @@ -55,6 +55,45 @@ openmode(ulong o) return o; } +long +syspipe(ulong *arg) +{ + int fd[2]; + Chan *c[2]; + Dev *d; + + validaddr(arg[0], 2*BY2WD, 1); + evenaddr(arg[0]); + d = &devtab[devno('|', 0)]; + c[0] = (*d->attach)(0); + c[1] = 0; + fd[0] = -1; + fd[1] = -1; + if(waserror()){ + close(c[0]); + if(c[1]) + close(c[1]); + if(fd[0] >= 0) + u->fd[fd[0]]=0; + if(fd[1] >= 0) + u->fd[fd[1]]=0; + nexterror(); + } + c[1] = (*d->clone)(c[0], 0); + (*d->walk)(c[0], "data"); + (*d->walk)(c[1], "data1"); + c[0] = (*d->open)(c[0], ORDWR); + c[1] = (*d->open)(c[1], ORDWR); + fd[0] = newfd(); + u->fd[fd[0]] = c[0]; + fd[1] = newfd(); + u->fd[fd[1]] = c[1]; + ((long*)arg[0])[0] = fd[0]; + ((long*)arg[0])[1] = fd[1]; + poperror(); + return 0; +} + long sysdup(ulong *arg) { @@ -180,8 +219,8 @@ sysread(ulong *arg) Chan *c; long n; - c = fdtochan(arg[0], 0); - validaddr(arg[1], arg[2], 0); + c = fdtochan(arg[0], OREAD); + validaddr(arg[1], arg[2], 1); qlock(c); if(waserror()){ qunlock(c); @@ -208,7 +247,7 @@ syswrite(ulong *arg) Chan *c; long n; - c = fdtochan(arg[0], 1); + c = fdtochan(arg[0], OWRITE); validaddr(arg[1], arg[2], 0); qlock(c); if(waserror()){ @@ -382,43 +421,6 @@ sysmount(ulong *arg) return bindmount(arg, 1); } -long -syspipe(ulong *arg) -{ - int fd[2]; - Chan *c[2]; - Dev *d; - - validaddr(arg[0], 2*BY2WD, 1); - evenaddr(arg[0]); - d = &devtab[devno('|', 0)]; - c[0] = (*d->attach)(0); - c[1] = 0; - fd[0] = -1; - fd[1] = -1; - if(waserror()){ - close(c[0]); - if(c[1]) - close(c[1]); - if(fd[0] >= 0) - u->fd[fd[0]]=0; - if(fd[1] >= 0) - u->fd[fd[1]]=0; - nexterror(); - } - c[1] = (*d->clone)(c[0], 0); - c[0] = (*d->open)(c[0], ORDWR); - c[1] = (*d->open)(c[1], ORDWR); - fd[0] = newfd(); - u->fd[fd[0]] = c[0]; - fd[1] = newfd(); - u->fd[fd[1]] = c[1]; - ((long*)arg[0])[0] = fd[0]; - ((long*)arg[0])[1] = fd[1]; - poperror(); - return 0; -} - long syscreate(ulong *arg) { diff --git a/gnot/sysproc.c b/gnot/sysproc.c index dc46af05b2badb2fc19fc4f87add841391025053..a2998e765fd4046341b7aa7b314698ec6e33fbbb 100644 --- a/gnot/sysproc.c +++ b/gnot/sysproc.c @@ -500,6 +500,8 @@ sysforkpgrp(ulong *arg) } if(arg[0] == 0) pgrpcpy(pg, u->p->pgrp); + else + memcpy(pg->user, u->p->pgrp->user, NAMELEN); closepgrp(u->p->pgrp); u->p->pgrp = pg; return pg->pgrpid; diff --git a/port/chan.c b/port/chan.c index 955e8bf6d13d93a0e9e31c300eed338f0529af93..335371736d213e8bb51ab77f3d26e5680214d6ab 100644 --- a/port/chan.c +++ b/port/chan.c @@ -479,7 +479,7 @@ namec(char *name, int amode, int omode, ulong perm) name = skipslash(name); }else if(name[0] == '#'){ mntok = 0; - if(name[1]=='|' || name[1]=='M') + if(name[1]=='M') error(0, Enonexist); t = devno(name[1], 1); if(t == -1) diff --git a/port/devpipe.c b/port/devpipe.c index 08a1ec1065412cb13fd34ba5a974b2589f37aed6..2183aae2fb553544d5f09b499932704a585539df 100644 --- a/port/devpipe.c +++ b/port/devpipe.c @@ -8,80 +8,111 @@ #include "devtab.h" #include "fcall.h" +typedef struct Pipe Pipe; + +struct Pipe +{ + Ref; + int debug; + Pipe *next; +}; + +struct Pipealloc +{ + Lock; + Pipe *pipe; + Pipe *free; +} pipealloc; + static void pipeiput(Queue*, Block*); static void pipeoput(Queue*, Block*); static void pipestclose(Queue *); Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "pipe" }; +Dirtab pipedir[]={ + "data", Sdataqid, 0, 0600, + "ctl", Sctlqid, 0, 0600, + "data1", Sdataqid, 0, 0600, + "ctl1", Sctlqid, 0, 0600, +}; +#define NPIPEDIR 4 + void pipeinit(void) { } +/* + * allocate structures for conf.npipe pipes + */ void pipereset(void) { + Pipe *p, *ep; + + pipealloc.pipe = ialloc(conf.npipe * sizeof(Pipe), 0); + ep = &pipealloc.pipe[conf.npipe-1]; + for(p = pipealloc.pipe; p < ep; p++) + p->next = p+1; + pipealloc.free = pipealloc.pipe; } /* - * allocate both streams - * - * a subsequent clone will get them the second stream + * create a pipe, no streams are created until an open */ Chan* pipeattach(char *spec) { + Pipe *p; Chan *c; - int i; - /* - * make the first stream - */ c = devattach('|', spec); - c->qid = STREAMQID(0, Sdataqid); - streamnew(c, &pipeinfo); + + lock(&pipealloc); + if(pipealloc.free == 0){ + unlock(&pipealloc); + error(0, Enopipe); + } + p = pipealloc.free; + pipealloc.free = p->next; + p->ref = 1; + unlock(&pipealloc); + + c->qid = CHDIR|STREAMQID(2*(p - pipealloc.pipe), 0); return c; } Chan* pipeclone(Chan *c, Chan *nc) { - /* - * make the second stream - */ + Pipe *p; + + p = &pipealloc.pipe[STREAMID(c->qid)/2]; nc = devclone(c, nc); - if(waserror()){ - close(nc); - nexterror(); - } - nc->qid = STREAMQID(1, Sdataqid); - streamnew(nc, &pipeinfo); - poperror(); + incref(p); + return nc; +} - /* - * attach it to the first - */ - c->stream->devq->ptr = (Stream *)nc->stream; - nc->stream->devq->ptr = (Stream *)c->stream; - c->stream->devq->other->next = nc->stream->devq; - nc->stream->devq->other->next = c->stream->devq; +int +pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp) +{ + int id; - /* - * up the inuse count of each stream to reflect the - * pointer from the other stream. - */ - if(streamenter(c->stream)<0) - panic("pipeattach"); - if(streamenter(nc->stream)<0) - panic("pipeattach"); - return nc; + id = STREAMID(c->qid); + if(i > 1) + id++; + if(tab==0 || i>=ntab) + return -1; + tab += i; + devdir(c, STREAMQID(id, tab->qid), tab->name, tab->length, tab->perm, dp); + return 1; } + int pipewalk(Chan *c, char *name) { - print("pipewalk\n"); - error(0, Egreg); + return devwalk(c, name, pipedir, NPIPEDIR, pipegen); } void @@ -90,10 +121,60 @@ pipestat(Chan *c, char *db) streamstat(c, db, "pipe"); } +/* + * if the stream doesn't exist, create it + */ Chan * pipeopen(Chan *c, int omode) { - c->mode = omode; + Pipe *p; + Stream *local, *remote; + + if(CHDIR & c->qid){ + if(omode != OREAD) + error(0, Ebadarg); + c->mode = omode; + c->flag |= COPEN; + c->offset = 0; + return c; + } + + p = &pipealloc.pipe[STREAMID(c->qid)/2]; + remote = 0; + if(waserror()){ + unlock(p); + if(remote) + streamclose1(remote); + nexterror(); + } + lock(p); + streamopen(c, &pipeinfo); + local = c->stream; + if(local->devq->ptr == 0){ + /* + * First stream opened, create the other end also + */ + remote = streamnew(c->type, c->dev, STREAMID(c->qid)^1, &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; + + /* + * increment the inuse count to reflect the + * pointer from the other stream. + */ + if(streamenter(local)<0) + panic("pipeattach"); + } + unlock(p); + poperror(); + + c->mode = omode&~OTRUNC; c->flag |= COPEN; c->offset = 0; return c; @@ -117,28 +198,57 @@ pipewstat(Chan *c, char *db) error(0, Eperm); } +void +pipeexit(Pipe *p) +{ + decref(p); + if(p->ref <= 0){ + lock(&pipealloc); + p->next = pipealloc.free; + pipealloc.free = p; + unlock(&pipealloc); + } +} + void pipeclose(Chan *c) { - Stream *other; + Stream *remote; + Stream *local; + Pipe *p; - other = (Stream *)c->stream->devq->ptr; + p = &pipealloc.pipe[STREAMID(c->qid)/2]; - if(waserror()){ - streamexit(other, 0); - nexterror(); + /* + * take care of assosiated streams + */ + if(local = c->stream){ + remote = (Stream *)c->stream->devq->ptr; + if(waserror()){ + streamexit(remote, 0); + pipeexit(p); + nexterror(); + } + streamclose(c); /* close this stream */ + streamexit(remote, 0); /* release stream for other half of pipe */ + poperror(); } - streamclose(c); /* close this stream */ - streamexit(other, 0); /* release stream for other half of pipe */ - poperror(); + pipeexit(p); } long piperead(Chan *c, void *va, long n) { - return streamread(c, va, n); + if(CHDIR&c->qid) + return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen); + else + return streamread(c, va, n); } +/* + * a write to a closed pipe causes a note to be sent to + * the process. + */ long pipewrite(Chan *c, void *va, long n) { diff --git a/port/stasync.c b/port/stasync.c new file mode 100644 index 0000000000000000000000000000000000000000..a4d5d19e4bcc3c65d602a83ad07bdffeafeeb6ba --- /dev/null +++ b/port/stasync.c @@ -0,0 +1,453 @@ +#include "u.h" +#include "lib.h" +#include "mem.h" +#include "dat.h" +#include "fns.h" +#include "errno.h" + +#define DPRINT if(asyncdebug)kprint + +/* + * configuration + */ +enum { + MAXFRAME= 256, /* also known to tsm8 code */ +}; + +/* input states */ +enum { Hunt=0, Framing, Framed, Data, Escape }; + +typedef struct Async { + QLock; + + int inuse; + Queue *wq; + + /* output state */ + + QLock xmit; /* transmit lock */ + int chan; /* current urp channel */ + Block *bp; /* current output buffer */ + int count; + ushort crc; + + /* input state */ + + int state; /* input state */ + uchar buf[MAXFRAME]; /* current input buffer */ + int icount; + ushort icrc; + + /* statistics */ + + ulong chan0; + ulong toolong; + ulong tooshort; + ulong badcrc; + ulong badescape; + ulong in; /* bytes in */ + ulong out; /* bytes out */ +} Async; + +Async *async; + +/* + * async stream module definition + */ +static void asynciput(Queue*, Block*); +static void asyncoput(Queue*, Block*); +static void asyncopen(Queue*, Stream*); +static void asyncclose(Queue*); +static void asyncreset(void); +Qinfo asyncinfo = { asynciput, asyncoput, asyncopen, asyncclose, "async", asyncreset }; + +int asyncdebug = 0; +int asyncerror; + +static ushort crc_table[256] = { +#include "crc_16.h" +}; + +#define BOT 0050 /* begin trailer */ +#define BOTM 0051 /* begin trailer, more data follows */ +#define BOTS 0052 /* seq update alg. on this trailer */ + +#define FRAME 0x7e +#define STUF 0x9d + +#define CRCSTART (crc_table[0xff]) +#define CRCFUNC(crc,x) (crc_table[((crc)^(x))&0xff]^((crc)>>8)) + +/* + * create the async structures + */ +static void +asyncreset(void) +{ + async = (Async *)ialloc(conf.nasync*sizeof(Async), 0); +} + +/* + * allocate an async structure + */ +static void +asyncopen(Queue *q, Stream *s) +{ + Async *ap; + + DPRINT("asyncopen %d\n", s->dev); + + for(ap = async; ap < &async[conf.nasync]; ap++){ + qlock(ap); + if(ap->inuse == 0) + break; + qunlock(ap); + } + if(ap == &async[conf.nasync]) + error(0, Enoasync); + q->ptr = q->other->ptr = ap; + + ap->inuse = 1; + ap->bp = 0; + ap->chan = -1; + ap->count = 0; + ap->toolong = 0; + ap->tooshort = 0; + ap->badcrc = 0; + ap->badescape = 0; + ap->chan0 = 0; + ap->in = 0; + ap->out = 0; + ap->wq = WR(q); + ap->state = Hunt; + qunlock(ap); +} + +static void +asyncclose(Queue * q) +{ + Async *ap = (Async *)q->ptr; + + DPRINT("asyncstclose %d\n", ap-async); + qlock(ap); + ap->inuse = 0; + qunlock(ap); +} + +/* + * free all blocks of a message in `q', `bp' is the first block + * of the message + */ +static void +freemsg(Queue *q, Block *bp) +{ + for(; bp; bp = getq(q)){ + if(bp->flags & S_DELIM){ + freeb(bp); + return; + } + freeb(bp); + } +} + +static void +showframe(char *t, Async *ap, uchar *buf, int n) +{ + kprint("a%d %s [", ap-async, t); + while (--n >= 0) + kprint(" %2.2ux", *buf++); + kprint(" ]\n"); +} + +void +aswrite(Async *ap) +{ + if(ap->bp->rptr == ap->bp->wptr) + return; + FLOWCTL(ap->wq); + PUTNEXT(ap->wq, ap->bp); + ap->bp = 0; +} + +void +asputf(Async *ap, int frame) +{ + uchar *p; + int c; + + p = ap->bp->wptr; + if(ap->count > 0) { + if(asyncerror) + ap->crc^=1, asyncerror=0; + *p++ = c = ap->crc&0xff; + if(c == FRAME) + *p++ = 0x00; + *p++ = c = (ap->crc>>8)&0xff; + if(c == FRAME) + *p++ = 0x00; + ap->count = 0; + } + if(frame) { + *p++ = FRAME; + *p++ = FRAME; + } + ap->bp->wptr = p; + if(asyncdebug > 2) + showframe("out", ap, ap->bp->rptr, BLEN(ap->bp)); + aswrite(ap); +} + +void +asputc(Async *ap, int c) +{ + int d; + uchar *p; + + if(ap->bp == 0) + ap->bp = allocb(MAXFRAME+4); + p = ap->bp->wptr; + if(ap->count <= 0) { + *p++ = FRAME; + *p++ = FRAME; + *p++ = d = 0x80|((ap->chan>>5)&0x7e); + ap->crc = CRCFUNC(CRCSTART, d); + *p++ = d = 0x80|((ap->chan<<1)&0x7e); + ap->crc = CRCFUNC(ap->crc, d); + } + *p++ = c; + if(c == FRAME) + *p++ = 0x00; + ap->crc = CRCFUNC(ap->crc, c); + ap->bp->wptr = p; + if(++ap->count >= MAXFRAME-4) + asputf(ap, 0); + else if(ap->bp->lim - p < 8) + aswrite(ap); +} + +/* + * output a block + * + * the first 2 bytes of every message are the channel number, + * low order byte first. the third is a possible trailing control + * character. + */ +void +asyncoput(Queue *q, Block *bp) +{ + Async *ap = (Async *)q->ptr; + int c, chan, ctl; + + if(bp->type != M_DATA){ + freeb(bp); + return; + } + + /* + * get a whole message before handing bytes to the device + */ + if(!putq(q, bp)) + return; + + /* + * one transmitter at a time + */ + qlock(&ap->xmit); + + /* + * parse message + */ + bp = getq(q); + if(bp->wptr - bp->rptr < 3){ + freemsg(q, bp); + qunlock(&ap->xmit); + return; + } + chan = bp->rptr[0] | (bp->rptr[1]<<8); + ctl = bp->rptr[2]; + bp->rptr += 3; + + /* + * new frame if the channel number has changed + */ + if(chan != ap->chan && ap->count > 0) + asputf(ap, 0); + ap->chan = chan; + + /* + * send the 8 bit data + */ + for(;;){ + /* + * put in next packet + */ + while (bp->rptr < bp->wptr) { + asputc(ap, c = *bp->rptr++); + if(c == STUF) + asputc(ap, 0); + } + + /* + * get next block + */ + if(bp->flags & S_DELIM){ + freeb(bp); + break; + } + freeb(bp); + bp = getq(q); + if(bp==0) + break; + } + + /* + * send the control byte if there is one + */ + if(ctl){ + asputc(ap, STUF); + asputc(ap, ctl); + switch (ctl) { + case BOT: + case BOTM: + case BOTS: + break; + default: + asputf(ap, 1); + } + } + + qunlock(&ap->xmit); + return; +} + +/* + * Read bytes from the raw input. + */ + +void +asdeliver(Queue *q, Async *ap) +{ + int chan, c; + Block *bp = 0; + uchar *p = ap->buf; + int n = ap->icount; + + chan = *p++ & 0x7e; + chan = (chan<<5)|((*p++ & 0x7e)>>1); + if(chan==0) { + DPRINT("a%d deliver chan 0\n", ap-async); + ap->chan0++; + return; + } + for (n-=4; n>0; n--) { + if(!bp) { + bp = allocb(n+2); + bp->flags |= S_DELIM; + bp->wptr[0] = chan; + bp->wptr[1] = chan>>8; + bp->wptr[2] = 0; + bp->wptr += 3; + } + if((c = *p++) == STUF) { + --n; + if((c = *p++) != 0) { + bp->rptr[2] = c; + if(asyncdebug > 1) + kprint("a%d<-(%d)%3.3uo %d\n", + ap-async, chan, bp->rptr[2], + bp->wptr - bp->rptr - 3); + PUTNEXT(q, bp); + bp = 0; + continue; + } else + c = STUF; + } + *bp->wptr++ = c; + } + if(bp) { + if(asyncdebug > 1) + kprint("a%d<-(%d)%3.3uo %d\n", + ap-async, chan, bp->rptr[2], + bp->wptr - bp->rptr - 3); + PUTNEXT(q, bp); + } +} + +static void +asynciput(Queue *q, Block *bp) +{ + int c; + Async *ap = q->ptr; + int state = ap->state; + + while(bp->wptr > bp->rptr){ + c = *bp->rptr++; + switch(state) { + case Hunt: /* wait for framing byte */ + if(c == FRAME) + state = Framing; + break; + + case Framing: /* saw 1 framing byte after Hunt */ + if(c == FRAME) + state = Framed; + else + state = Hunt; + break; + + case Framed: /* saw 2 or more framing bytes */ + if(c == FRAME) + break; + state = Data; + ap->icrc = CRCSTART; + ap->icount = 0; + goto Datachar; + + case Data: /* mid-frame */ + if(c == FRAME) { + state = Escape; + break; + } + Datachar: + if(ap->icount >= MAXFRAME) { + DPRINT("a%d pkt too long\n", ap-async); + ap->toolong++; + state = Hunt; + break; + } + ap->icrc = CRCFUNC(ap->icrc, c); + ap->buf[ap->icount++] = c; + break; + + case Escape: /* saw framing byte in Data */ + switch (c) { + case FRAME: + if(asyncdebug > 2) + showframe("in", ap, ap->buf, ap->icount); + if(ap->icount < 5) { + DPRINT("a%d pkt too short\n", ap-async); + ap->tooshort++; + } else if(ap->icrc != 0) { + DPRINT("a%d bad crc\n", ap-async); + ap->badcrc++; + } else { + asdeliver(q, ap); + } + state = Framed; + break; + case 0: + c = FRAME; + state = Data; + goto Datachar; + default: + DPRINT("a%d bad escape\n", ap-async); + ap->badescape++; + state = Hunt; + break; + } + break; + } + } + ap->state = state; + freeb(bp); +} diff --git a/port/stream.c b/port/stream.c index a9f2201b21d78ea6ebfe254b970671167fa8c429..a8eaacb96529349a3386d9420723f409e15d0944 100644 --- a/port/stream.c +++ b/port/stream.c @@ -235,6 +235,7 @@ allocq(Qinfo *qi) q->info = qi; q->put = qi->iput; q->len = q->nb = 0; + q->ptr = 0; wq = q->other = q + 1; wq->flag = QINUSE; @@ -242,6 +243,7 @@ allocq(Qinfo *qi) wq->info = qi; wq->put = qi->oput; wq->other = q; + wq->ptr = 0; wq->len = wq->nb = 0; unlock(q); @@ -625,10 +627,10 @@ streamgen(Chan *c, Dirtab *tab, int ntab, int s, Dir *dp) } /* - * create a new stream + * create a new stream, if noopen is non-zero, don't increment the open count */ Stream * -streamnew(Chan *c, Qinfo *qi) +streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen) { Stream *s; Queue *q; @@ -651,26 +653,25 @@ streamnew(Chan *c, Qinfo *qi) } if(waserror()){ unlock(s); - streamclose(c); + streamclose1(s); nexterror(); } /* - * marry a stream and a channel + * identify the stream */ - if(c){ - c->stream = s; - s->type = c->type; - s->dev = c->dev; - s->id = STREAMID(c->qid); - } else - s->type = -1; + s->type = type; + s->dev = dev; + s->id = id; /* * hang a device and process q off the stream */ s->inuse = 1; - s->opens = 1; + if(noopen) + s->opens = 0; + else + s->opens = 1; s->hread = 0; q = allocq(&procinfo); s->procq = WR(q); @@ -699,7 +700,7 @@ streamopen(Chan *c, Qinfo *qi) Queue *q; /* - * if the stream already exists, just up the reference count. + * if the stream already exists, just increment the reference counts. */ for(s = slist; s < &slist[conf.nstream]; s++) { if(s->inuse && s->type == c->type && s->dev == c->dev @@ -721,7 +722,7 @@ streamopen(Chan *c, Qinfo *qi) /* * create a new stream */ - streamnew(c, qi); + c->stream = streamnew(c->type, c->dev, STREAMID(c->qid), qi, 0); } /* @@ -773,17 +774,10 @@ streamexit(Stream *s, int locked) * stream release its blocks and call its close routine. */ void -streamclose(Chan *c) +streamclose1(Stream *s) { Queue *q, *nq; Block *bp; - Stream *s = c->stream; - - /* - * if not open, ignore it - */ - if(!c->stream) - return; /* * decrement the reference count @@ -817,6 +811,16 @@ streamclose(Chan *c) streamexit(s, 1); unlock(s); } +void +streamclose(Chan *c) +{ + /* + * if no stream, ignore it + */ + if(!c->stream) + return; + streamclose1(c->stream); +} /* * put a block to be read into the queue. wakeup any waiting reader @@ -1174,7 +1178,7 @@ dumpblocks(Queue *q, char c) lock(q); for(bp = q->first; bp; bp = bp->next){ - print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)); + print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' '); for(cp = bp->rptr; cpwptr && cprptr+10; cp++) print(" %uo", *cp); print("\n"); diff --git a/port/sysfile.c b/port/sysfile.c index 47b648148f97da2254b18c2d6298b5681eac51c1..c45fbe353aed5cb48bcd41d4dbd6e784b931556a 100644 --- a/port/sysfile.c +++ b/port/sysfile.c @@ -31,12 +31,12 @@ fdtochan(int fd, int mode) if(fd<0 || NFD<=fd || (c=u->fd[fd])==0) error(0, Ebadfd); - if(mode<0 || c->mode == 2) + if(mode<0 || c->mode==ORDWR) return c; - if((mode&16) && c->mode==0) + if((mode&OTRUNC) && c->mode==OREAD) err: error(0, Ebadusefd); - if((mode&~16) != c->mode) + if((mode&~OTRUNC) != c->mode) goto err; return c; } @@ -55,6 +55,45 @@ openmode(ulong o) return o; } +long +syspipe(ulong *arg) +{ + int fd[2]; + Chan *c[2]; + Dev *d; + + validaddr(arg[0], 2*BY2WD, 1); + evenaddr(arg[0]); + d = &devtab[devno('|', 0)]; + c[0] = (*d->attach)(0); + c[1] = 0; + fd[0] = -1; + fd[1] = -1; + if(waserror()){ + close(c[0]); + if(c[1]) + close(c[1]); + if(fd[0] >= 0) + u->fd[fd[0]]=0; + if(fd[1] >= 0) + u->fd[fd[1]]=0; + nexterror(); + } + c[1] = (*d->clone)(c[0], 0); + (*d->walk)(c[0], "data"); + (*d->walk)(c[1], "data1"); + c[0] = (*d->open)(c[0], ORDWR); + c[1] = (*d->open)(c[1], ORDWR); + fd[0] = newfd(); + u->fd[fd[0]] = c[0]; + fd[1] = newfd(); + u->fd[fd[1]] = c[1]; + ((long*)arg[0])[0] = fd[0]; + ((long*)arg[0])[1] = fd[1]; + poperror(); + return 0; +} + long sysdup(ulong *arg) { @@ -180,7 +219,7 @@ sysread(ulong *arg) Chan *c; long n; - c = fdtochan(arg[0], 0); + c = fdtochan(arg[0], OREAD); validaddr(arg[1], arg[2], 1); qlock(c); if(waserror()){ @@ -208,7 +247,7 @@ syswrite(ulong *arg) Chan *c; long n; - c = fdtochan(arg[0], 1); + c = fdtochan(arg[0], OWRITE); validaddr(arg[1], arg[2], 0); qlock(c); if(waserror()){ @@ -382,43 +421,6 @@ sysmount(ulong *arg) return bindmount(arg, 1); } -long -syspipe(ulong *arg) -{ - int fd[2]; - Chan *c[2]; - Dev *d; - - validaddr(arg[0], 2*BY2WD, 1); - evenaddr(arg[0]); - d = &devtab[devno('|', 0)]; - c[0] = (*d->attach)(0); - c[1] = 0; - fd[0] = -1; - fd[1] = -1; - if(waserror()){ - close(c[0]); - if(c[1]) - close(c[1]); - if(fd[0] >= 0) - u->fd[fd[0]]=0; - if(fd[1] >= 0) - u->fd[fd[1]]=0; - nexterror(); - } - c[1] = (*d->clone)(c[0], 0); - c[0] = (*d->open)(c[0], ORDWR); - c[1] = (*d->open)(c[1], ORDWR); - fd[0] = newfd(); - u->fd[fd[0]] = c[0]; - fd[1] = newfd(); - u->fd[fd[1]] = c[1]; - ((long*)arg[0])[0] = fd[0]; - ((long*)arg[0])[1] = fd[1]; - poperror(); - return 0; -} - long syscreate(ulong *arg) { diff --git a/port/sysproc.c b/port/sysproc.c index ec664417d23ce326386049a10534ec7d92e446f9..98f004829e77b9cee616df7f3451f82d965d0ee8 100644 --- a/port/sysproc.c +++ b/port/sysproc.c @@ -496,6 +496,8 @@ sysforkpgrp(ulong *arg) } if(arg[0] == 0) pgrpcpy(pg, u->p->pgrp); + else + memcpy(pg->user, u->p->pgrp->user, NAMELEN); closepgrp(u->p->pgrp); u->p->pgrp = pg; return pg->pgrpid; diff --git a/power/conf.h b/power/conf.h index 8f1343efd731898225634c770bf49a6d5f898640..5dc87296cbc5722073d4feb8d395c73ddee18346 100644 --- a/power/conf.h +++ b/power/conf.h @@ -24,6 +24,8 @@ Conftab conftab[] = { {"nnoifc", &conf.nnoifc }, {"nnoconv", &conf.nnoconv }, {"nurp", &conf.nurp }, + {"nasync", &conf.nasync }, + {"npipe", &conf.npipe }, { 0, 0 }, }; diff --git a/power/dat.h b/power/dat.h index 8e8ddd0dbfb252a8468f4024a47c6c3f5acfa1cb..02657abb3774354421c2fd7e6147f32f1d426e83 100644 --- a/power/dat.h +++ b/power/dat.h @@ -148,6 +148,8 @@ struct Conf ulong nnoifc; /* number of nonet interfaces */ ulong nnoconv; /* number of nonet conversations/ifc */ ulong nurp; /* max urp conversations */ + ulong nasync; /* number of async protocol modules */ + ulong npipe; /* number of pipes */ }; struct Dev @@ -450,14 +452,13 @@ struct Queue { */ struct Stream { Lock; /* structure lock */ - int inuse; /* number of processes in stream */ - int opens; /* number of processes with stream open */ - int hread; /* number of reads after hangup */ - int type; /* correclation with Chan */ - int dev; /* ... */ - int id; /* ... */ + short inuse; /* number of processes in stream */ + short opens; /* number of processes with stream open */ + ushort hread; /* number of reads after hangup */ + ushort type; /* correlation with Chan */ + ushort dev; /* ... */ + ushort id; /* ... */ QLock rdlock; /* read lock */ - QLock wrlock; /* write lock */ Queue *procq; /* write queue at process end */ Queue *devq; /* read queue at device end */ }; diff --git a/power/errno.h b/power/errno.h index d23af7433fb18e67d3a2d7d3a04b142353048053..25d777b756b1686c409906e9b85416813044a51a 100644 --- a/power/errno.h +++ b/power/errno.h @@ -51,5 +51,7 @@ enum{ Ebadcnt, /* read count greater than requested */ Enoannounce, /* listening on an unannounced network connection */ Enovmem, /* virtual memory allocation failed */ + Enoasync, /* out of async stream modules */ + Enopipe, /* out of pipes */ Egreg, /* ken hasn't implemented datakit */ }; diff --git a/power/fns.h b/power/fns.h index 9c28dafbdaaf30bde57f09c09b7b241e920bd621..eeee28c01d03295ca22cba99416b2e5ae09e5c45 100644 --- a/power/fns.h +++ b/power/fns.h @@ -169,12 +169,13 @@ void splx(int); void sunmap(int, uchar*); Devgen streamgen; void streamclose(Chan*); +void streamclose1(Stream*); int streamenter(Stream*); void streamexit(Stream*, int); void streaminit(void); long streamread(Chan*, void*, long); long streamwrite(Chan*, void*, long, int); -Stream* streamnew(Chan*, Qinfo*); +Stream* streamnew(ushort, ushort, ushort, Qinfo*, int); void streamopen(Chan*, Qinfo*); int streamparse(char*, Block*); void streamstat(Chan*, char*, char*); diff --git a/power/main.c b/power/main.c index ce17bacbf901e576974297adc1946b8bab1072b9..cd6012bafb434585919841419d30dd62bb0af9af 100644 --- a/power/main.c +++ b/power/main.c @@ -612,6 +612,7 @@ confinit(void) conf.npte = 4 * conf.npage; conf.nqueue = 3 * conf.nstream; conf.nblock = 10 * conf.nstream; + conf.npipe = conf.nstream/2; confread();