#include	"u.h"#include	"lib.h"#include	"dat.h"#include	"fns.h"#include	"error.h" #include	"netif.h" typedef struct Pipe	Pipe;struct Pipe{	QLock lk;	Pipe	*next;	int	ref;	ulong	path;	Queue	*q[2];	int	qref[2];}; struct{	Lock lk;	ulong	path;} pipealloc; enum{	Qdir,	Qdata0,	Qdata1,}; Dirtab pipedir[] ={	".",		{Qdir,0,QTDIR},	0,		DMDIR|0500,	"data",		{Qdata0},	0,		0600,	"data1",	{Qdata1},	0,		0600,};#define NPIPEDIR 3 static voidpipeinit(void){	if(conf.pipeqsize == 0){		if(conf.nmach > 1)			conf.pipeqsize = 256*1024;		else			conf.pipeqsize = 32*1024;	}} /* *  create a pipe, no streams are created until an open */static Chan*pipeattach(char *spec){	Pipe *p;	Chan *c; 	c = devattach('|', spec);	p = malloc(sizeof(Pipe));	if(p == 0)		exhausted("memory");	p->ref = 1; 	p->q[0] = qopen(conf.pipeqsize, 0, 0, 0);	if(p->q[0] == 0){		free(p);		exhausted("memory");	}	p->q[1] = qopen(conf.pipeqsize, 0, 0, 0);	if(p->q[1] == 0){		free(p->q[0]);		free(p);		exhausted("memory");	} 	lock(&pipealloc.lk);	p->path = ++pipealloc.path;	unlock(&pipealloc.lk); 	mkqid(&c->qid, NETQID(2*p->path, Qdir), 0, QTDIR);	c->aux = p;	c->dev = 0;	return c;} static intpipegen(Chan *c, char *name, Dirtab *tab, int ntab, int i, Dir *dp){	Qid q;	int len;	Pipe *p; 	USED(name); 	if(i == DEVDOTDOT){		devdir(c, c->qid, "#|", 0, eve, DMDIR|0555, dp);		return 1;	}	i++;	/* skip . */	if(tab==0 || i>=ntab)		return -1; 	tab += i;	p = c->aux;	switch((ulong)tab->qid.path){	case Qdata0:		len = qlen(p->q[0]);		break;	case Qdata1:		len = qlen(p->q[1]);		break;	default:		len = tab->length;		break;	}	mkqid(&q, NETQID(NETID(c->qid.path), tab->qid.path), 0, QTFILE);	devdir(c, q, tab->name, len, eve, tab->perm, dp);	return 1;}  static Walkqid*pipewalk(Chan *c, Chan *nc, char **name, int nname){	Walkqid *wq;	Pipe *p; 	wq = devwalk(c, nc, name, nname, pipedir, NPIPEDIR, pipegen);	if(wq != nil && wq->clone != nil && wq->clone != c){		p = c->aux;		qlock(&p->lk);		p->ref++;		if(c->flag & COPEN){			print("channel open in pipewalk\n");			switch(NETTYPE(c->qid.path)){			case Qdata0:				p->qref[0]++;				break;			case Qdata1:				p->qref[1]++;				break;			}		}		qunlock(&p->lk);	}	return wq;} static intpipestat(Chan *c, uchar *db, int n){	Pipe *p;	Dir dir; 	p = c->aux; 	switch(NETTYPE(c->qid.path)){	case Qdir:		devdir(c, c->qid, ".", 0, eve, DMDIR|0555, &dir);		break;	case Qdata0:		devdir(c, c->qid, "data", qlen(p->q[0]), eve, 0600, &dir);		break;	case Qdata1:		devdir(c, c->qid, "data1", qlen(p->q[1]), eve, 0600, &dir);		break;	default:		panic("pipestat");	}	n = convD2M(&dir, db, n);	if(n < BIT16SZ)		error(Eshortstat);	return n;} /* *  if the stream doesn't exist, create it */static Chan*pipeopen(Chan *c, int omode){	Pipe *p; 	if(c->qid.type & QTDIR){		if(omode != OREAD)			error(Ebadarg);		c->mode = omode;		c->flag |= COPEN;		c->offset = 0;		return c;	} 	p = c->aux;	qlock(&p->lk);	switch(NETTYPE(c->qid.path)){	case Qdata0:		p->qref[0]++;		break;	case Qdata1:		p->qref[1]++;		break;	}	qunlock(&p->lk); 	c->mode = openmode(omode);	c->flag |= COPEN;	c->offset = 0;	c->iounit = qiomaxatomic;	return c;} static voidpipeclose(Chan *c){	Pipe *p; 	p = c->aux;	qlock(&p->lk); 	if(c->flag & COPEN){		/*		 *  closing either side hangs up the stream		 */		switch(NETTYPE(c->qid.path)){		case Qdata0:			p->qref[0]--;			if(p->qref[0] == 0){				qhangup(p->q[1], 0);				qclose(p->q[0]);			}			break;		case Qdata1:			p->qref[1]--;			if(p->qref[1] == 0){				qhangup(p->q[0], 0);				qclose(p->q[1]);			}			break;		}	}  	/*	 *  if both sides are closed, they are reusable	 */	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->lk);		free(p->q[0]);		free(p->q[1]);		free(p);	} else		qunlock(&p->lk);} static longpiperead(Chan *c, void *va, long n, vlong offset){	Pipe *p; 	USED(offset); 	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 */} static Block*pipebread(Chan *c, long n, ulong offset){	Pipe *p; 	p = c->aux; 	switch(NETTYPE(c->qid.path)){	case Qdata0:		return qbread(p->q[0], n);	case Qdata1:		return qbread(p->q[1], n);	} 	return devbread(c, n, offset);} /* *  a write to a closed pipe causes a note to be sent to *  the process. */static longpipewrite(Chan *c, void *va, long n, vlong offset){	Pipe *p; 	USED(offset); 	if(waserror()) {		/* avoid notes when pipe is a mounted queue */		if((c->flag & CMSG) == 0)			postnote(up, 1, "sys: write on closed pipe", NUser);		nexterror();	} 	p = c->aux; 	switch(NETTYPE(c->qid.path)){	case Qdata0:		n = qwrite(p->q[1], va, n);		break; 	case Qdata1:		n = qwrite(p->q[0], va, n);		break; 	default:		panic("pipewrite");	} 	poperror();	return n;} static longpipebwrite(Chan *c, Block *bp, ulong offset){	long n;	Pipe *p; 	USED(offset); 	if(waserror()) {		/* avoid notes when pipe is a mounted queue */		if((c->flag & CMSG) == 0)			postnote(up, 1, "sys: write on closed pipe", NUser);		nexterror();	} 	p = c->aux;	switch(NETTYPE(c->qid.path)){	case Qdata0:		n = qbwrite(p->q[1], bp);		break; 	case Qdata1:		n = qbwrite(p->q[0], bp);		break; 	default:		n = 0;		panic("pipebwrite");	} 	poperror();	return n;} Dev pipedevtab = {	'|',	"pipe", 	devreset,	pipeinit,	devshutdown,	pipeattach,	pipewalk,	pipestat,	pipeopen,	devcreate,	pipeclose,	piperead,	pipebread,	pipewrite,	pipebwrite,	devremove,	devwstat,};