2002/0109/port/devpipe.c (diff list | history)

port/devpipe.c on 1990/0227
1990/0227		#include "u.h"
1992/0321		#include "../port/lib.h"
1990/0227		#include "mem.h" #include "dat.h" #include "fns.h"
1992/0111		#include "../port/error.h"
1990/0227
1993/0528		#include "netif.h"
1990/0227
1990/1009		typedef struct Pipe Pipe; struct Pipe {
1991/1227		QLock;
1990/1009		Pipe *next;
1993/0528		int ref;
1992/0621		ulong path;
1993/0528		Queue *q[2]; int qref[2];
1990/1009		};
1992/0621		struct
1990/1009		{ Lock;
1992/0621		ulong path;
1990/1009		} pipealloc;
1993/0528		enum
1990/1113		{
1993/0528		Qdir, Qdata0, Qdata1,
1990/1113		};
1990/0227
1992/0621		Dirtab pipedir[] = {
2001/0527		".", {Qdir,0,QTDIR}, 0, DMDIR|0500, "data", {Qdata0}, 0, 0600, "data1", {Qdata1}, 0, 0600,
1990/1009		};
2001/0527		#define NPIPEDIR 3
1990/1009
1997/0327		static void
1990/0227		pipeinit(void) {
1994/0503		if(conf.pipeqsize == 0){ if(conf.nmach > 1) conf.pipeqsize = 256*1024; else conf.pipeqsize = 32*1024; }
1990/0227		} /*
1990/1009		* create a pipe, no streams are created until an open
1990/0227		*/
1997/0327		static Chan*
1990/0227		pipeattach(char *spec) {
1990/1009		Pipe *p;
1990/0227		Chan *c;
1990/0629
1990/0227		c = devattach('|', spec);
1993/0528		p = malloc(sizeof(Pipe)); if(p == 0) exhausted("memory");
1992/0621		p->ref = 1;
1990/1009
1994/0503		p->q[0] = qopen(conf.pipeqsize, 0, 0, 0);
1993/0528		if(p->q[0] == 0){ free(p); exhausted("memory"); }
1994/0503		p->q[1] = qopen(conf.pipeqsize, 0, 0, 0);
1993/0528		if(p->q[1] == 0){ free(p->q[0]); free(p); exhausted("memory"); }
1990/1009		lock(&pipealloc);
1992/0621		p->path = ++pipealloc.path;
1990/1009		unlock(&pipealloc);
2001/0527		mkqid(&c->qid, NETQID(2*p->path, Qdir), 0, QTDIR);
1993/0528		c->aux = p;
1991/1227		c->dev = 0;
1990/0227		return c; }
1997/0327		static int
2001/0527		pipegen(Chan *c, char*, Dirtab *tab, int ntab, int i, Dir *dp)
1990/1009		{
2001/0813		Qid q;
2000/0516		int len; Pipe *p;
1990/0629
1999/1230		if(i == DEVDOTDOT){
2001/0527		devdir(c, c->qid, "#|", 0, eve, DMDIR|0555, dp);
1999/1230		return 1; }
2001/1017		i++; /* skip . */
1990/1009		if(tab==0 || i>=ntab) return -1;
2001/0527
1990/1009		tab += i;
2000/0516		p = c->aux;
2001/0527		switch((ulong)tab->qid.path){
2000/0516		case Qdata0: len = qlen(p->q[0]); break; case Qdata1: len = qlen(p->q[1]); break; default: len = tab->length; break; }
2001/0527		mkqid(&q, NETQID(NETID(c->qid.path), tab->qid.path), 0, QTFILE); devdir(c, q, tab->name, len, eve, tab->perm, dp);
1990/1009		return 1;
1990/0227		}
1990/1009
2001/0527		static Walkqid* pipewalk(Chan *c, Chan *nc, char **name, int nname)
1990/0227		{
2001/0527		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); 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); } return wq;
1990/0227		}
2001/0527		static int pipestat(Chan *c, uchar *db, int n)
1990/0227		{
1993/0528		Pipe *p; Dir dir; p = c->aux; switch(NETTYPE(c->qid.path)){ case Qdir:
2001/0527		devdir(c, c->qid, ".", 0, eve, DMDIR|0555, &dir);
1993/0528		break; case Qdata0:
1993/0530		devdir(c, c->qid, "data", qlen(p->q[0]), eve, 0660, &dir);
1993/0528		break; case Qdata1:
1993/0530		devdir(c, c->qid, "data1", qlen(p->q[1]), eve, 0660, &dir);
1993/0528		break; default: panic("pipestat"); }
2001/0527		n = convD2M(&dir, db, n); if(n < BIT16SZ) error(Eshortstat); return n;
1990/0227		}
1990/1009		/* * if the stream doesn't exist, create it */
1997/0327		static Chan*
1990/0227		pipeopen(Chan *c, int omode) {
1990/1009		Pipe *p;
2001/0527		if(c->qid.type & QTDIR){
1990/1009		if(omode != OREAD)
1990/11211		error(Ebadarg);
1990/1009		c->mode = omode; c->flag |= COPEN; c->offset = 0; return c; }
1993/0528		p = c->aux;
1991/1227		qlock(p);
1993/0528		switch(NETTYPE(c->qid.path)){ case Qdata0: p->qref[0]++; break; case Qdata1: p->qref[1]++; break;
1990/1009		}
1991/1227		qunlock(p);
1990/1009
1993/1106		c->mode = openmode(omode);
1990/0227		c->flag |= COPEN; c->offset = 0;
2001/1207		c->iounit = qiomaxatomic;
1990/0227		return c; }
1997/0327		static void
1990/0227		pipeclose(Chan *c) {
1994/0804		Pipe *p;
1990/0629
1993/0528		p = c->aux; qlock(p);
1990/0629
1993/1111		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){
1997/0327		qhangup(p->q[1], 0);
1994/0804		qclose(p->q[0]);
1993/1111		} break; case Qdata1: p->qref[1]--; if(p->qref[1] == 0){
1997/0327		qhangup(p->q[0], 0);
1994/0804		qclose(p->q[1]);
1993/1111		} break;
1991/0314		}
1990/11211		}
1990/11161
1998/0512
1990/11161		/*
1993/0528		* if both sides are closed, they are reusable
1990/11161		*/
1993/0528		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); free(p->q[0]); free(p->q[1]);
1992/0621		free(p);
1994/0804		} else qunlock(p);
1990/0227		}
1997/0327		static long
1998/0319		piperead(Chan *c, void *va, long n, vlong)
1990/0227		{
1993/0528		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 */
1990/0227		}
1997/0327		static Block*
1995/0108		pipebread(Chan *c, long n, ulong offset) {
1995/0714		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); }
1995/0108		return devbread(c, n, offset); }
1990/1009		/* * a write to a closed pipe causes a note to be sent to * the process. */
1997/0327		static long
1998/0319		pipewrite(Chan *c, void *va, long n, vlong)
1990/0227		{
1993/0528		Pipe *p;
1997/0327		if(!islo())
1999/0501		print("pipewrite hi %lux\n", getcallerpc(&c));
1993/0725		if(waserror()) {
1993/0911		/* avoid notes when pipe is a mounted queue */ if((c->flag & CMSG) == 0) postnote(up, 1, "sys: write on closed pipe", NUser);
1999/0219		nexterror();
1993/0725		}
1993/0528		p = c->aux;
1992/0821
1993/0528		switch(NETTYPE(c->qid.path)){ case Qdata0:
1994/0902		n = qwrite(p->q[1], va, n);
1993/0725		break;
1993/0528		case Qdata1:
1994/0902		n = qwrite(p->q[0], va, n);
1993/0725		break;
1993/0528		default:
1995/0714		panic("pipewrite");
1990/0227		}
1993/0725		poperror();
1990/0513		return n;
1995/0108		}
1997/0327		static long
1995/0804		pipebwrite(Chan *c, Block *bp, ulong)
1995/0108		{
1995/0714		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);
1999/0219		nexterror();
1995/0714		} 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:
1995/0804		n = 0;
1995/0714		panic("pipebwrite"); } poperror(); return n;
1990/0227		}
1997/0327		Dev pipedevtab = {
1997/0408		'|', "pipe",
1997/0327		devreset, pipeinit,
2002/0109		devshutdown,
1997/0327		pipeattach, pipewalk, pipestat, pipeopen, devcreate, pipeclose, piperead, pipebread, pipewrite, pipebwrite, devremove, devwstat, };