vx32

devpipe.c (5881B)

---

 #include	"u.h"
 #include	"lib.h"
 #include	"mem.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 void
 pipeinit(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 int
 pipegen(Chan *c, char *name, Dirtab *tab, int ntab, int i, Dir *dp)
 {
 	Qid q;
 	int len;
 	Pipe *p;
 
 	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 int
 pipestat(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 void
 pipeclose(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 long
 piperead(Chan *c, void *va, long n, vlong offset)
 {
 	Pipe *p;
 
 	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 long
 pipewrite(Chan *c, void *va, long n, vlong offset)
 {
 	Pipe *p;
 
 	if(!islo())
 		print("pipewrite hi %#p\n", getcallerpc(&c));
 	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 long
 pipebwrite(Chan *c, Block *bp, ulong offset)
 {
 	long n;
 	Pipe *p;
 
 	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,
 };