1990/1009/port/devpipe.c (diff list | history)

1990/0801/sys/src/9/port/devpipe.c:8,45 – 1990/1009/sys/src/9/port/devpipe.c:8,84 (short | long | prev | next)
1990/0227		#include "devtab.h" #include "fcall.h"
1990/1009		typedef struct Pipe Pipe; struct Pipe { Ref; int debug; Pipe *next; }; struct Pipealloc { Lock; Pipe *pipe; Pipe *free; } pipealloc;
1990/0227		static void pipeiput(Queue*, Block*); static void pipeoput(Queue*, Block*); static void pipestclose(Queue *);
1990/0403		Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "pipe" };
1990/0227
1990/1009		Dirtab pipedir[]={ "data", Sdataqid, 0, 0600, "ctl", Sctlqid, 0, 0600, "data1", Sdataqid, 0, 0600, "ctl1", Sctlqid, 0, 0600, }; #define NPIPEDIR 4
1990/0227		void pipeinit(void) { }
1990/1009		/* * allocate structures for conf.npipe pipes */
1990/0227		void pipereset(void) {
1990/1009		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;
1990/0227		} /* * allocate both streams * * a subsequent clone will get them the second stream
1990/1009		* create a pipe, no streams are created until an open
1990/0227		*/ Chan* pipeattach(char *spec) {
1990/1009		Pipe *p;
1990/0227		Chan *c;
1990/0629		int i;
1990/0227		/* * make the first stream */ c = devattach('|', spec); c->qid = STREAMQID(0, Sdataqid); streamnew(c, &pipeinfo);
1990/1009		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);
1990/0227		return c; }
1990/0801/sys/src/9/port/devpipe.c:46,87 – 1990/1009/sys/src/9/port/devpipe.c:85,118
1990/0227		Chan* pipeclone(Chan *c, Chan *nc) { /* * make the second stream */
1990/1009		Pipe *p; p = &pipealloc.pipe[STREAMID(c->qid)/2];
1990/0227		nc = devclone(c, nc); if(waserror()){ close(nc); nexterror(); } nc->qid = STREAMQID(1, Sdataqid); streamnew(nc, &pipeinfo); poperror();
1990/1009		incref(p); return nc; }
1990/0227		/* * attach it to the first */
1990/0629		c->stream->devq->ptr = (Stream *)nc->stream; nc->stream->devq->ptr = (Stream *)c->stream;
1990/0227		c->stream->devq->other->next = nc->stream->devq; nc->stream->devq->other->next = c->stream->devq;
1990/1009		int pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp) { int id;
1990/0629		/* * 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");
1990/0227		return nc;
1990/1009		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;
1990/0227		}
1990/1009
1990/0227		int pipewalk(Chan *c, char *name) { print("pipewalk\n"); error(0, Egreg);
1990/1009		return devwalk(c, name, pipedir, NPIPEDIR, pipegen);
1990/0227		} void
1990/0801/sys/src/9/port/devpipe.c:90,99 – 1990/1009/sys/src/9/port/devpipe.c:121,180
1990/0801		streamstat(c, db, "pipe");
1990/0227		}
1990/1009		/* * if the stream doesn't exist, create it */
1990/0227		Chan * pipeopen(Chan *c, int omode) { c->mode = omode;
1990/1009		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;
1990/0227		c->flag |= COPEN; c->offset = 0; return c;
1990/0801/sys/src/9/port/devpipe.c:118,144 – 1990/1009/sys/src/9/port/devpipe.c:199,254
1990/0227		} void
1990/1009		pipeexit(Pipe *p) { decref(p); if(p->ref <= 0){ lock(&pipealloc); p->next = pipealloc.free; pipealloc.free = p; unlock(&pipealloc); } } void
1990/0227		pipeclose(Chan *c) {
1990/0629		Stream *other;
1990/1009		Stream *remote; Stream *local; Pipe *p;
1990/0629		other = (Stream *)c->stream->devq->ptr;
1990/1009		p = &pipealloc.pipe[STREAMID(c->qid)/2];
1990/0629		if(waserror()){ streamexit(other, 0); nexterror();
1990/1009		/* * 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();
1990/0629		} streamclose(c); /* close this stream */ streamexit(other, 0); /* release stream for other half of pipe */ poperror();
1990/1009		pipeexit(p);
1990/0227		} long piperead(Chan *c, void *va, long n) { return streamread(c, va, n);
1990/1009		if(CHDIR&c->qid) return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen); else return streamread(c, va, n);
1990/0227		}
1990/1009		/* * a write to a closed pipe causes a note to be sent to * the process. */
1990/0227		long pipewrite(Chan *c, void *va, long n) {