19 files changed, 940 insertions(+), 242 deletions(-)

M gnot/dat.h
M gnot/devpipe.c
M gnot/errno.h
M gnot/fns.h
M gnot/stasync.c
M gnot/stream.c
M gnot/sysfile.c
M gnot/sysproc.c
M port/chan.c
M port/devpipe.c
A port/stasync.c
M port/stream.c
M port/sysfile.c
M port/sysproc.c
M power/conf.h
M power/dat.h
M power/errno.h
M power/fns.h
M power/main.c

@@ 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 */
 };

@@ 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;


@@ 118,27 199,56 @@ pipewstat(Chan *c, char *db)
 }
 
 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)
 {

@@ 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 */
 };

@@ 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*);

@@ 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] = {

@@ 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; cp<bp->wptr && cp<bp->rptr+10; cp++)
 			print(" %uo", *cp);
 		print("\n");

@@ 56,6 56,45 @@ openmode(ulong 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)
 {
 	int fd;


@@ 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()){


@@ 383,43 422,6 @@ sysmount(ulong *arg)
 }
 
 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)
 {
 	int fd;

@@ 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;

@@ 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)

@@ 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;


@@ 118,27 199,56 @@ pipewstat(Chan *c, char *db)
 }
 
 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)
 {

@@ 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);
+}

@@ 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; cp<bp->wptr && cp<bp->rptr+10; cp++)
 			print(" %uo", *cp);
 		print("\n");

@@ 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;
 }


@@ 56,6 56,45 @@ openmode(ulong 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)
 {
 	int fd;


@@ 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()){


@@ 383,43 422,6 @@ sysmount(ulong *arg)
 }
 
 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)
 {
 	int fd;

@@ 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;

@@ 24,6 24,8 @@ Conftab conftab[] = {
 	{"nnoifc", &conf.nnoifc },
 	{"nnoconv", &conf.nnoconv },
 	{"nurp", &conf.nurp },
+	{"nasync", &conf.nasync },
+	{"npipe", &conf.npipe },
 	{ 0, 0 },
 };
 

@@ 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 */
 };

@@ 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 */
 };

@@ 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*);

@@ 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();