/* mpipefs - userspace multipipe implementation Copyright (C) 2010, IBM Corporation, Eric Van Hensbergen (bergevan@us.ibm.com) Description: This is a user space implementation of our multipipe concept intended to more easily experiment with ideas prior to attempting a more efficient kernel version. Notes: more common use case for this is to just post at a srv point versus to actually mount itself due to nature of instance per attach. Bugs: Flushes and interruptions are probably going to break everything right now. Need to add code to handle such situations in a reasonable fashion.*/ #include <u.h>#include <libc.h>#include <fcall.h>#include <thread.h>#include <9p.h>#include <mp.h>#include <libsec.h> enum{ STACK = 8192, MSIZE = 8192, MAXARGS = 8, MAXHDRARGS = 4,}; Channel *iochan; /* dispatch channel */typedef struct Fidaux Fidaux;char defaultsrvpath[] = "mpipe"; char Enotdir[] = "walk in non-directory";char Enotfound[] = "directory entry not found";char Eoneorother[] = "open read or write, not both";char Eshort[] = "short write";char Ehangup[] = "pipe hang up";char Eremain[] = "new header when bytes still remaining";char Eother[] = "Problem with peer";char Ebcastoverflow[] = "broadcast read underflow";char Ebcast[] = "broadcast failed (perhaps partially)"; enum { MPTnormal = 1, /* strictly a normal pipe */ MPTbcast=2, /* enumerated pipe */}; typedef struct Mpipe Mpipe;struct Mpipe { QLock l; int ref; int mode; /* pipe type */ ulong path; char *name; char *uid; char *gid; ulong len; int writers; /* number of writers */ int readers; /* number of readers */ int slots; /* number of enumerations */ Channel **rrchan; /* ready to read */ int *numr; Fidaux *bcastr; /* broadcast readers */}; struct Fidaux { QLock l; Mpipe *mp; ulong which; /* for enumerations */ /* only for writer */ Fid *other; /* peer Fid for long writes */ int remain; /* bytes remaining in multi-block */ /* only for reader */ Req *r; /* for partial reads */ Channel *chan; /* for bcast */ Fidaux *next;}; static struct{ Lock l; ulong path;} pipealloc; typedef struct Splicearg Splicearg;struct Splicearg { Mpipe *mp; int which; int fd;}; static voidusage(void){ fprint(2, "mpipefs [-D] [-s <srvpt>] [mtpt]\n"); threadexits("usage");} voidkillall(Srv*){ chanclose(iochan); chanfree(iochan); threadexitsall("killall");} #define NETTYPE(x) ((ulong)(x)&0x1f)#define NETID(x) (((ulong)(x))>>5)#define NETQID(i,t) (((i)<<5)|(t)) enum{ Qdir, Qdata,}; char MPdefault[] = "data"; /* tiny helper, why don't we have this generically? */static void *emalloc9pz(int sz, int zero){ void *v = emalloc9p(sz); if(zero) memset(v, 0, sz); return v;} static voidmyrespond(Req *r, char *err){ if(r->aux) sendp(r->aux, err); else respond(r, err);} static voidmyresponderror(Req *r){ if(r->aux) { char *err = emalloc9p(ERRMAX); snprint(err, ERRMAX, "%r"); sendp(r->aux, err); } else responderror(r);} static Mpipe *setfidmp(Fid *fid){ if(fid->qid.type&QTDIR) return fid->aux; if(fid->omode == -1) return fid->aux; else return ((Fidaux *)fid->aux)->mp;} static Fidaux *setfidaux(Fid *fid) /* blech */{ if(fid->qid.type&QTDIR) return nil; if(fid->omode == -1) return nil; else return fid->aux;} /* MAYBE: allow splice-to to specify which by setting aux->which */static voidaddnewreader(Mpipe *mp, Fidaux *aux){ int count; int min = mp->numr[0]; mp->readers++; aux->which = 0; for(count=0; count<mp->slots; count++) if(mp->numr[count] < min) aux->which = count; mp->numr[aux->which]++; aux->chan = chancreate(sizeof(Req *), 0); qlock(&mp->l); if(mp->mode == MPTbcast) { aux->next=mp->bcastr; mp->bcastr=aux; } qunlock(&mp->l);} static voidrmreader(Mpipe *mp, Fidaux *aux){ Fidaux *c; qlock(&mp->l); if(mp->bcastr == aux) mp->bcastr = aux->next; else { for(c = mp->bcastr; c != nil; c=c->next) { if(c->next = aux) { c->next = aux->next; break; } } } aux->next = nil; qunlock(&mp->l);} static voidclosebcasts(Mpipe *mp){ Fidaux *c; for(c=mp->bcastr; c != nil; c=c->next) chanclose(c->chan);} static voidparsespec(Mpipe *mp, int argc, char **argv){ if(argc >= 1) mp->name = estrdup9p(argv[0]); ARGBEGIN { case 'e': mp->slots = atoi(ARGF()); break; case 'b': mp->mode = MPTbcast; break; default: print(" badflag('%c')", ARGC()); }ARGEND} static voidfsattach(Req *r){ Mpipe *mp = emalloc9pz(sizeof(Mpipe), 1); char *argv[MAXARGS]; int argc; int count; /* need to parse aname */ argc = tokenize(r->ifcall.aname, argv, MAXARGS); parsespec(mp, argc, argv); if(mp->slots == 0) mp->slots = 1; if(mp->name == nil) { mp->name = estrdup9p(MPdefault); } lock(&pipealloc.l); mp->path = ++pipealloc.path; unlock(&pipealloc.l); r->fid->qid = (Qid){NETQID(2*mp->path, Qdir), 0, QTDIR}; r->ofcall.qid = r->fid->qid; r->fid->aux = mp; mp->uid = estrdup9p(getuser()); mp->gid = estrdup9p(mp->uid); mp->rrchan = emalloc9p(sizeof(Channel *)*mp->slots); mp->numr = emalloc9p(sizeof(int)*mp->slots); for(count = 0; count < mp->slots; count++) { mp->rrchan[count] = chancreate(sizeof(Fid *), 0); mp->numr[count] = 0; } mp->ref++; respond(r, nil);} static char*fswalk1(Fid *fid, char *name, Qid *qid){ ulong path; Mpipe *mp = setfidmp(fid); path = fid->qid.path; if(!(fid->qid.type&QTDIR)) return Enotdir; /* there's only one dir, so just compare names */ if(strcmp(name, mp->name) == 0) { qid->path = NETQID(path, Qdata); qid->type = QTFILE; return nil; } return Enotfound;} static char*fsclone(Fid* fid, Fid* newfid){ Mpipe *mp = setfidmp(fid); newfid->aux = fid->aux; mp->ref++; return nil;} static voidfsclunk(Fid *fid){ Mpipe *mp = setfidmp(fid); int count; if(fid->aux) { Fidaux *aux = setfidaux(fid); mp->ref--; if(mp->ref == 0) { free(mp->name); free(mp->uid); free(mp->gid); if(mp->rrchan) { for(count=0;count < mp->slots; count++) { if(mp->rrchan[count]) { chanclose(mp->rrchan[count]); chanfree(mp->rrchan[count]); } } free(mp->rrchan); } free(mp); } if((fid->omode&OMASK) == OWRITE) { /* MAYBE: mark outstanding requests crap */ mp->writers--; if(mp->writers == 0) { for(count=0;count < mp->slots; count++) if(mp->rrchan[count]) chanclose(mp->rrchan[count]); } if(mp->mode == MPTbcast) closebcasts(mp); free(aux); fid->aux = nil; } if(((fid->omode&OMASK) == 0)&&(fid->qid.type==QTFILE)) { mp->readers--; if(mp->mode == MPTbcast) rmreader(mp, aux); chanclose(aux->chan); chanfree(aux->chan); mp->numr[aux->which]--; aux->chan = nil; if(aux->r) {/* crap! we still have data - return short write */ respond(aux->r, "short write"); aux->r = nil; } if((mp->writers == 0)&&(mp->readers == 0)) { for(count=0;count < mp->slots; count++) { if(mp->rrchan[count]) chanfree(mp->rrchan[count]); mp->rrchan[count] = chancreate(sizeof(Fid *), 0); } } } } fid->aux = nil;} static voidfsopen(Req *r){ Fid* fid = r->fid; Qid q = fid->qid; Mpipe *mp = setfidmp(fid); Fidaux *aux; if (q.type&QTDIR){ respond(r, nil); return; } if((r->ifcall.mode&OMASK) == ORDWR) { respond(r, Eoneorother); return; } aux = emalloc9pz(sizeof(Fidaux), 1); aux->mp = mp; fid->aux = aux; if((r->ifcall.mode&OMASK) == OWRITE) mp->writers++; if((r->ifcall.mode&OMASK)==0) /* READER */ addnewreader(mp, aux); respond(r, nil);} static intgetdirent(int n, Dir* d, void *aux){ Mpipe *mp = aux; if(aux == nil) return -1; d->atime= time(nil); d->mtime= d->atime; d->uid = estrdup9p(mp->uid); d->gid = estrdup9p(mp->gid); d->muid= estrdup9p(d->uid); if(n == -1) { d->qid = (Qid){NETQID(mp->path, Qdir), 0, QTDIR}; d->mode = 0775; d->name = estrdup9p("/"); d->length = 0; } else if (n==0) { d->qid = (Qid){NETQID(mp->path, Qdata), 0, QTFILE}; d->mode = 0666; d->name = estrdup9p(mp->name); d->length = mp->len; } else { return -1; } return 0;} /* copy from transmitter to receiver */static intfscopy(Req *tr, Req *r, int offset){ int count = tr->ifcall.count - offset; if(count > r->ifcall.count) count = r->ifcall.count; /* truncate */ memcpy(r->ofcall.data+offset, tr->ifcall.data, count); r->ofcall.count = count; return count;} static voidfsread(void *arg){ Req *r = arg; Fid* fid = r->fid; Fidaux *aux = setfidaux(fid); Fidaux *raux; Mpipe *mp = setfidmp(fid); int count; Req *tr; int offset; Qid q = fid->qid; if (q.type&QTDIR) { dirread9p(r, getdirent, mp); respond(r, nil); threadexits(nil); } if(aux->r) { /* If remainder, serve that first */ tr = aux->r; offset = tr->ofcall.count;Lines 1–500 of 957. Load more
/* mpipefs - userspace multipipe implementation Copyright (C) 2010, IBM Corporation, Eric Van Hensbergen (bergevan@us.ibm.com) Description: This is a user space implementation of our multipipe concept intended to more easily experiment with ideas prior to attempting a more efficient kernel version. Notes: more common use case for this is to just post at a srv point versus to actually mount itself due to nature of instance per attach. Bugs: Flushes and interruptions are probably going to break everything right now. Need to add code to handle such situations in a reasonable fashion. */ #include <u.h> #include <libc.h> #include <fcall.h> #include <thread.h> #include <9p.h> #include <mp.h> #include <libsec.h> enum { STACK = 8192, MSIZE = 8192, MAXARGS = 8, MAXHDRARGS = 4, }; Channel *iochan; /* dispatch channel */ typedef struct Fidaux Fidaux; char defaultsrvpath[] = "mpipe"; char Enotdir[] = "walk in non-directory"; char Enotfound[] = "directory entry not found"; char Eoneorother[] = "open read or write, not both"; char Eshort[] = "short write"; char Ehangup[] = "pipe hang up"; char Eremain[] = "new header when bytes still remaining"; char Eother[] = "Problem with peer"; char Ebcastoverflow[] = "broadcast read underflow"; char Ebcast[] = "broadcast failed (perhaps partially)"; enum { MPTnormal = 1, /* strictly a normal pipe */ MPTbcast=2, /* enumerated pipe */ }; typedef struct Mpipe Mpipe; struct Mpipe { QLock l; int ref; int mode; /* pipe type */ ulong path; char *name; char *uid; char *gid; ulong len; int writers; /* number of writers */ int readers; /* number of readers */ int slots; /* number of enumerations */ Channel **rrchan; /* ready to read */ int *numr; Fidaux *bcastr; /* broadcast readers */ }; struct Fidaux { QLock l; Mpipe *mp; ulong which; /* for enumerations */ /* only for writer */ Fid *other; /* peer Fid for long writes */ int remain; /* bytes remaining in multi-block */ /* only for reader */ Req *r; /* for partial reads */ Channel *chan; /* for bcast */ Fidaux *next; }; static struct { Lock l; ulong path; } pipealloc; typedef struct Splicearg Splicearg; struct Splicearg { Mpipe *mp; int which; int fd; }; static void usage(void) { fprint(2, "mpipefs [-D] [-s <srvpt>] [mtpt]\n"); threadexits("usage"); } void killall(Srv*) { chanclose(iochan); chanfree(iochan); threadexitsall("killall"); } #define NETTYPE(x) ((ulong)(x)&0x1f) #define NETID(x) (((ulong)(x))>>5) #define NETQID(i,t) (((i)<<5)|(t)) enum { Qdir, Qdata, }; char MPdefault[] = "data"; /* tiny helper, why don't we have this generically? */ static void * emalloc9pz(int sz, int zero) { void *v = emalloc9p(sz); if(zero) memset(v, 0, sz); return v; } static void myrespond(Req *r, char *err) { if(r->aux) sendp(r->aux, err); else respond(r, err); } static void myresponderror(Req *r) { if(r->aux) { char *err = emalloc9p(ERRMAX); snprint(err, ERRMAX, "%r"); sendp(r->aux, err); } else responderror(r); } static Mpipe * setfidmp(Fid *fid) { if(fid->qid.type&QTDIR) return fid->aux; if(fid->omode == -1) return fid->aux; else return ((Fidaux *)fid->aux)->mp; } static Fidaux * setfidaux(Fid *fid) /* blech */ { if(fid->qid.type&QTDIR) return nil; if(fid->omode == -1) return nil; else return fid->aux; } /* MAYBE: allow splice-to to specify which by setting aux->which */ static void addnewreader(Mpipe *mp, Fidaux *aux) { int count; int min = mp->numr[0]; mp->readers++; aux->which = 0; for(count=0; count<mp->slots; count++) if(mp->numr[count] < min) aux->which = count; mp->numr[aux->which]++; aux->chan = chancreate(sizeof(Req *), 0); qlock(&mp->l); if(mp->mode == MPTbcast) { aux->next=mp->bcastr; mp->bcastr=aux; } qunlock(&mp->l); } static void rmreader(Mpipe *mp, Fidaux *aux) { Fidaux *c; qlock(&mp->l); if(mp->bcastr == aux) mp->bcastr = aux->next; else { for(c = mp->bcastr; c != nil; c=c->next) { if(c->next = aux) { c->next = aux->next; break; } } } aux->next = nil; qunlock(&mp->l); } static void closebcasts(Mpipe *mp) { Fidaux *c; for(c=mp->bcastr; c != nil; c=c->next) chanclose(c->chan); } static void parsespec(Mpipe *mp, int argc, char **argv) { if(argc >= 1) mp->name = estrdup9p(argv[0]); ARGBEGIN { case 'e': mp->slots = atoi(ARGF()); break; case 'b': mp->mode = MPTbcast; break; default: print(" badflag('%c')", ARGC()); }ARGEND } static void fsattach(Req *r) { Mpipe *mp = emalloc9pz(sizeof(Mpipe), 1); char *argv[MAXARGS]; int argc; int count; /* need to parse aname */ argc = tokenize(r->ifcall.aname, argv, MAXARGS); parsespec(mp, argc, argv); if(mp->slots == 0) mp->slots = 1; if(mp->name == nil) { mp->name = estrdup9p(MPdefault); } lock(&pipealloc.l); mp->path = ++pipealloc.path; unlock(&pipealloc.l); r->fid->qid = (Qid){NETQID(2*mp->path, Qdir), 0, QTDIR}; r->ofcall.qid = r->fid->qid; r->fid->aux = mp; mp->uid = estrdup9p(getuser()); mp->gid = estrdup9p(mp->uid); mp->rrchan = emalloc9p(sizeof(Channel *)*mp->slots); mp->numr = emalloc9p(sizeof(int)*mp->slots); for(count = 0; count < mp->slots; count++) { mp->rrchan[count] = chancreate(sizeof(Fid *), 0); mp->numr[count] = 0; } mp->ref++; respond(r, nil); } static char* fswalk1(Fid *fid, char *name, Qid *qid) { ulong path; Mpipe *mp = setfidmp(fid); path = fid->qid.path; if(!(fid->qid.type&QTDIR)) return Enotdir; /* there's only one dir, so just compare names */ if(strcmp(name, mp->name) == 0) { qid->path = NETQID(path, Qdata); qid->type = QTFILE; return nil; } return Enotfound; } static char* fsclone(Fid* fid, Fid* newfid) { Mpipe *mp = setfidmp(fid); newfid->aux = fid->aux; mp->ref++; return nil; } static void fsclunk(Fid *fid) { Mpipe *mp = setfidmp(fid); int count; if(fid->aux) { Fidaux *aux = setfidaux(fid); mp->ref--; if(mp->ref == 0) { free(mp->name); free(mp->uid); free(mp->gid); if(mp->rrchan) { for(count=0;count < mp->slots; count++) { if(mp->rrchan[count]) { chanclose(mp->rrchan[count]); chanfree(mp->rrchan[count]); } } free(mp->rrchan); } free(mp); } if((fid->omode&OMASK) == OWRITE) { /* MAYBE: mark outstanding requests crap */ mp->writers--; if(mp->writers == 0) { for(count=0;count < mp->slots; count++) if(mp->rrchan[count]) chanclose(mp->rrchan[count]); } if(mp->mode == MPTbcast) closebcasts(mp); free(aux); fid->aux = nil; } if(((fid->omode&OMASK) == 0)&&(fid->qid.type==QTFILE)) { mp->readers--; if(mp->mode == MPTbcast) rmreader(mp, aux); chanclose(aux->chan); chanfree(aux->chan); mp->numr[aux->which]--; aux->chan = nil; if(aux->r) {/* crap! we still have data - return short write */ respond(aux->r, "short write"); aux->r = nil; } if((mp->writers == 0)&&(mp->readers == 0)) { for(count=0;count < mp->slots; count++) { if(mp->rrchan[count]) chanfree(mp->rrchan[count]); mp->rrchan[count] = chancreate(sizeof(Fid *), 0); } } } } fid->aux = nil; } static void fsopen(Req *r) { Fid* fid = r->fid; Qid q = fid->qid; Mpipe *mp = setfidmp(fid); Fidaux *aux; if (q.type&QTDIR){ respond(r, nil); return; } if((r->ifcall.mode&OMASK) == ORDWR) { respond(r, Eoneorother); return; } aux = emalloc9pz(sizeof(Fidaux), 1); aux->mp = mp; fid->aux = aux; if((r->ifcall.mode&OMASK) == OWRITE) mp->writers++; if((r->ifcall.mode&OMASK)==0) /* READER */ addnewreader(mp, aux); respond(r, nil); } static int getdirent(int n, Dir* d, void *aux) { Mpipe *mp = aux; if(aux == nil) return -1; d->atime= time(nil); d->mtime= d->atime; d->uid = estrdup9p(mp->uid); d->gid = estrdup9p(mp->gid); d->muid= estrdup9p(d->uid); if(n == -1) { d->qid = (Qid){NETQID(mp->path, Qdir), 0, QTDIR}; d->mode = 0775; d->name = estrdup9p("/"); d->length = 0; } else if (n==0) { d->qid = (Qid){NETQID(mp->path, Qdata), 0, QTFILE}; d->mode = 0666; d->name = estrdup9p(mp->name); d->length = mp->len; } else { return -1; } return 0; } /* copy from transmitter to receiver */ static int fscopy(Req *tr, Req *r, int offset) { int count = tr->ifcall.count - offset; if(count > r->ifcall.count) count = r->ifcall.count; /* truncate */ memcpy(r->ofcall.data+offset, tr->ifcall.data, count); r->ofcall.count = count; return count; } static void fsread(void *arg) { Req *r = arg; Fid* fid = r->fid; Fidaux *aux = setfidaux(fid); Fidaux *raux; Mpipe *mp = setfidmp(fid); int count; Req *tr; int offset; Qid q = fid->qid; if (q.type&QTDIR) { dirread9p(r, getdirent, mp); respond(r, nil); threadexits(nil); } if(aux->r) { /* If remainder, serve that first */ tr = aux->r; offset = tr->ofcall.count;