ref: c90981972dd33870d49e8a36a7baf382849a7758
dir: /kern/devpipe.c/
#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 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; 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 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; 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 long pipewrite(Chan *c, void *va, long n, vlong offset) { Pipe *p; USED(offset); if(!islo()) print("pipewrite hi %lux\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; 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, };