1992/0305/port/devpipe.c (diff list | history)

port/devpipe.c on 1990/0227
1990/0227		#include "u.h" #include "lib.h" #include "mem.h" #include "dat.h" #include "fns.h"
1992/0111		#include "../port/error.h"
1990/0227		#include "devtab.h" #include "fcall.h"
1990/1009		typedef struct Pipe Pipe; struct Pipe { Ref;
1991/1227		QLock;
1990/1009		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/1113		Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "pipe" };
1990/0227
1990/1009		Dirtab pipedir[]={
1990/11211		"data", {Sdataqid}, 0, 0600, "ctl", {Sctlqid}, 0, 0600, "data1", {Sdataqid}, 0, 0600, "ctl1", {Sctlqid}, 0, 0600,
1990/1009		}; #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		} /*
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
1990/0227		c = devattach('|', spec);
1990/1009		lock(&pipealloc); if(pipealloc.free == 0){ unlock(&pipealloc);
1991/0125		close(c);
1992/0114		exhausted("pipes");
1990/1009		} p = pipealloc.free; pipealloc.free = p->next;
1990/1115		if(incref(p) != 1)
1990/1013		panic("pipeattach");
1990/1009		unlock(&pipealloc);
1990/11211		c->qid = (Qid){CHDIR|STREAMQID(2*(p - pipealloc.pipe), 0), 0};
1991/1227		c->dev = 0;
1990/0227		return c; } Chan* pipeclone(Chan *c, Chan *nc) {
1990/1009		Pipe *p;
1990/11211		p = &pipealloc.pipe[STREAMID(c->qid.path)/2];
1990/0227		nc = devclone(c, nc);
1990/1013		if(incref(p) <= 1) panic("pipeclone");
1990/1009		return nc; }
1990/0227
1990/1009		int pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp) { int id;
1990/0629
1990/11211		id = STREAMID(c->qid.path);
1990/1009		if(i > 1) id++; if(tab==0 || i>=ntab) return -1; tab += i;
1991/1109		devdir(c, (Qid){STREAMQID(id, tab->qid.path),0}, tab->name, tab->length, eve, tab->perm, dp);
1990/1009		return 1;
1990/0227		}
1990/1009
1990/0227		int pipewalk(Chan *c, char *name) {
1990/1009		return devwalk(c, name, pipedir, NPIPEDIR, pipegen);
1990/0227		} void pipestat(Chan *c, char *db) {
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) {
1990/1009		Pipe *p; Stream *local, *remote;
1990/11211		if(c->qid.path & CHDIR){
1990/1009		if(omode != OREAD)
1990/11211		error(Ebadarg);
1990/1009		c->mode = omode; c->flag |= COPEN; c->offset = 0; return c; }
1990/11211		p = &pipealloc.pipe[STREAMID(c->qid.path)/2];
1990/1009		if(waserror()){
1991/1227		qunlock(p);
1990/1009		nexterror(); }
1991/1227		qlock(p);
1990/1009		streamopen(c, &pipeinfo); local = c->stream; if(local->devq->ptr == 0){ /*
1991/1227		* first open, create the other end also
1990/1009		*/
1991/1227		remote = streamnew(c->type, c->dev, STREAMID(c->qid.path)^1, &pipeinfo,1);
1990/1009		/* * 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;
1991/1227		} else if(local->opens == 1){
1990/1009		/*
1991/1227		* keep other side around till last close of this side
1990/1009		*/
1991/1227		streamenter(local->devq->ptr);
1990/1009		}
1991/1227		qunlock(p);
1990/1009		poperror(); c->mode = omode&~OTRUNC;
1990/0227		c->flag |= COPEN; c->offset = 0; return c; } void pipecreate(Chan *c, char *name, int omode, ulong perm) {
1991/1115		USED(c, name, omode, perm);
1990/11211		error(Egreg);
1990/0227		} void piperemove(Chan *c) {
1991/1115		USED(c);
1990/11211		error(Egreg);
1990/0227		} void pipewstat(Chan *c, char *db) {
1991/1115		USED(c, db);
1990/11211		error(Eperm);
1990/0227		} void pipeclose(Chan *c) {
1990/1009		Pipe *p;
1990/11211		Stream *remote;
1990/0629
1990/11211		p = &pipealloc.pipe[STREAMID(c->qid.path)/2];
1990/0629
1990/1009		/*
1991/1227		* take care of local and remote streams
1990/1009		*/
1990/11211		if(c->stream){
1991/1227		qlock(p);
1990/11211		remote = c->stream->devq->ptr;
1991/1227		if(streamclose(c) == 0){
1991/0314		if(remote) streamexit(remote, 0); }
1991/1227		qunlock(p);
1990/11211		}
1990/11161		/* * free the structure */ if(decref(p) == 0){ lock(&pipealloc); p->next = pipealloc.free; pipealloc.free = p; unlock(&pipealloc); } if(p->ref < 0) panic("pipeexit");
1990/0227		} long
1991/0411		piperead(Chan *c, void *va, long n, ulong offset)
1990/0227		{
1990/11211		if(c->qid.path & CHDIR)
1990/1009		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
1991/0411		pipewrite(Chan *c, void *va, long n, ulong offset)
1990/0227		{ if(waserror()){ postnote(u->p, 1, "sys: write on closed pipe", NExit);
1990/11211		error(Egreg);
1990/0227		}
1990/0513		n = streamwrite(c, va, n, 0); poperror(); return n;
1990/0227		} /* * stream stuff */ /* * send a block up stream to the process. * sleep untill there's room upstream. */ static void pipeiput(Queue *q, Block *bp) {
1992/0305		FLOWCTL(q, bp);
1990/0227		} /*
1991/1227		* send the block to the other side
1990/0227		*/ static void pipeoput(Queue *q, Block *bp) {
1990/0629		PUTNEXT(q, bp);
1990/0227		} /* * send a hangup and disconnect the streams */ static void pipestclose(Queue *q) { Block *bp; /* * point to the bit-bucket and let any in-progress * write's finish. */ q->put = nullput; wakeup(&q->r); /* * send a hangup */ q = q->other;
1991/0302		if(q->next == 0) return;
1990/0629		bp = allocb(0); bp->type = M_HANGUP; PUTNEXT(q, bp);
1990/0227		}