/* 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. Todo: * explicit hangup modes * buffered mpipes * internal splice optimization * reads of ~(0) yielding ctl blocks * ability to forward headers * network tight coupling mode (distributed mpipes) * direct connects bypassing 9P * broadcast mpipe tight coupling * collective mpipe tight coupling on BG/P * MAYBE: RDMA support * filter builtins for common scenarios (whitespace, EOL, etc) See Also: 9p(2) 9pfid(2) 9pfile(2)*/ #include <u.h>#include <libc.h>#include <fcall.h>#include <thread.h>#include <9p.h>#include <mp.h>#include <libsec.h>#include <debug.h> enum{ STACK = 8192, MSIZE = 8192, MAXARGS = 8, /* maximum spec arguments */ MAXHDRARGS = 4, /* maximum header entries */}; Channel *iochan; /* dispatch channel */typedef struct Fidaux Fidaux;char defaultsrvpath[] = "mpipe";int iothread_id; 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)";char Ebadspec[] = "bad mount specification arguments";char Esplice[] = "splice couldnt open target";char Ebadhdr[] = "unknown packet header"; /* * Right now there are two types of pipe: * 1) normal - writers match to readers * 2) broadcast - writers send to all readers * */enum { MPTnormal = 1, /* strictly a normal pipe */ MPTbcast = 2, /* broadcast pipe */ MPThangingup, /* terminating */ MPThungup, MPTcleanup,}; /* * Mpipe - primary data structure * * There is one of these per pipe. Mostly it contains * the information necessary for file system queries on the * pipe, but also maintains a number of reference count values * relating to accounting for readers and writers. * * for non-enumerated pipes, there is only a single slot * */typedef struct Mpipe Mpipe;struct Mpipe { QLock l; /* protect pipe */ int ref; /* reference count */ int mode; /* pipe type */ ulong path; /* qid.path */ char *name; /* name in directory */ char *uid; /* owner id */ char *gid; /* owner group */ ulong len; /* current length in bytes of data */ int writers; /* number of writers */ int readers; /* number of readers */ int slots; /* number of enumerations */ /* normal only, one entry per slot: */ Channel **rrchan; /* ready to read channels */ int *numr; /* number of readers on slot */ /* broadcast only */ Fidaux *bcastr; /* list of readers */}; /* states for Fidaux */enum { FID_IDLE, FID_CTLONLY, FID_WRITEN, FID_FLUSHED,}; /* * Fidaux - per particpant descriptor accounting structure * * This is typically hung off the fid private value and contains * the accounting for a particular "participant" (reader or writer) * */struct Fidaux { QLock l; /* Protect it */ int state; /* if participant has been written yet */ Mpipe *mp; /* Multipipe we are dealing with */ 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; /* cached request for partial reads */ Channel *chan; /* Response channel */ /* for bcast */ Fidaux *next; /* Linked list of broadcasters */}; /* * pipealloc - qid.path accounting * * we must guarentee unique qid.paths, this helps with that * */static struct{ Lock l; ulong path;} pipealloc; /* * splicearg - parameter encapsulation * * since splices get spawned off as sub-procs, this allows * us to pass all the arguments in a structure */ typedef struct Splicearg Splicearg;struct Splicearg { Mpipe *mp; /* the pipe in question */ int which; /* enumeration, 0 for don't care */ int fd; /* file descriptor to read from/write to */ char *path; /* name of target of splice */ Channel *chan; /* error/sync channel */}; /* * Bcastr - Broadcast argument * * since broadcasts get spawned off as sub-threads, this allows * us to pass all the arguments in a structure */typedef struct Bcastr Bcastr;struct Bcastr { Channel *chan; /* channel to send on */ Req *r; /* request to send */}; static voidusage(void){ fprint(2, "mpipefs [-D] [-v debuglevel] [-s <srvpt>] [mtpt]\n"); threadexits("usage");} static voidkillall(Srv*){ DPRINT(DFID, "killing iochan and all thread\n"); chanfree(iochan); threadexitsall("killall");} /* FIXME: prepending/adding the PID with pipealloc's path will not fit * within the lower 32 bits * * This might not scale well. Need to reexamine how the qid's are * generated and acounted for not only the local machine and the * synthetic file systems, but also for its scalability. */#define MAGIC ((uvlong) 0x6163 << 32)#define NETTYPE(x) ((ulong)(x)&0x1f)#define NETID(x) ((((ulong)(x))>>5)&0xffff)#define NETQID(i,t) (MAGIC|((((i)&0xffff)<<5)|((getpid()&0x7ff)<<21)|(t))) enum{ Qdir, Qdata,}; static 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;} /* * We need to send our responses down a channel * if one exists so that errors are propagated properly -- * particularly for broadcasts. */static voidmyrespond(Req *r, char *err){ if(r->aux) sendp(r->aux, err); else respond(r, err);} /* * This only gets called from spliceto. */static voidmyresponderror(Req *r){ if(r->aux) { sendp(r->aux, smprint("%r")); } 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 *//* register a new reader with the infrastructure */static voidaddnewreader(Mpipe *mp, Fidaux *aux){ int count; int min = mp->numr[0]; aux->chan = chancreate(sizeof(Req *), 0); if(aux->chan == nil) { DPRINT(DERR, "*ERROR*: addnewreader attempting to operate on a null chanel\n"); return ; } if(mp->mode == MPThangingup || mp->mode == MPThungup || mp->mode == MPTcleanup) { DPRINT(DBCA,"[%s](%p) WARNING: mp mode set to not running\n"); return; } qlock(&mp->l); mp->readers++; DPRINT(DRDR, "[%s](%p) addnewreader total now: %d\n", mp->name, mp, mp->readers); /* find enumeration slot with fewest readers */ aux->which = 0; for(count=0; count<mp->slots; count++) if(mp->numr[count] < min) aux->which = count; mp->numr[aux->which]++; if(mp->mode == MPTbcast) { DPRINT(DBCA,"[%s](%p) adding %p to broadcast list next: %p\n", mp->name, mp, aux, mp->bcastr); aux->next=mp->bcastr; mp->bcastr=aux; } qunlock(&mp->l);} /* safely remove a reader from the broadcast list */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);} /* cleanup broadcast pipes on last write close */static voidclosebcasts(Mpipe *mp){ Fidaux *c; DPRINT(DFID, "closebcasts: pid=(%d)\n", getpid()); for(c=mp->bcastr; c != nil; c=c->next) { chanclose(c->chan); }} /* close everyone the mpipe has slotted */static voidclosempipe(Mpipe *mp){ int count; DPRINT(DFID, "closempipe: pid=(%d)\n", getpid()); for(count=0;count < mp->slots; count++) if(chanclosing(mp->rrchan[count])) chanclose(mp->rrchan[count]);} /* process mount options */static char *parsespec(Mpipe *mp, int argc, char **argv){ char *f; char *err=nil; ARGBEGIN { case 'e': /* enumerated mode */ f = ARGF(); if(f) { char *x = f; if(x) mp->slots = atoi(x); DPRINT(DARG, "(%p) enumerated mode (%d)\n", mp, mp->slots); } else { DPRINT(DERR, "(%p) *ERROR*: usage: invalid enumeration argument\n", mp); err = Ebadspec; } break; case 'b': /* broadcast mode */ DPRINT(DARG, "(%p) parsespec: BROADCAST mode\n", mp); mp->mode = MPTbcast; break; default: DPRINT(DERR, "(%p) *ERROR*: badflag('%c')", mp, ARGC()); err = Ebadspec; }ARGEND DPRINT(DARG, "(%p) argc: %d\n", mp, argc); if(argc>0) mp->name = estrdup9p(*argv); return err;} static voidfsattach(Req *r){ Mpipe *mp = emalloc9pz(sizeof(Mpipe), 1); char *argv[MAXARGS]; int argc; int count; char *err; char *scratch; /* need to parse aname */ DPRINT(DARG, "args: (%s) len=%d\n", r->ifcall.aname, r->ifcall.count); /* add a stupid extra field so we can use args(2) */ scratch = smprint("mount %s\n", r->ifcall.aname); argc = tokenize(scratch, argv, MAXARGS); DPRINT(DARG, "argc: %d\n", argc); err = parsespec(mp, argc, argv); free(scratch); if(err) { respond(r, err); return; } 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++; Lines 1–500 of 1307. 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. Todo: * explicit hangup modes * buffered mpipes * internal splice optimization * reads of ~(0) yielding ctl blocks * ability to forward headers * network tight coupling mode (distributed mpipes) * direct connects bypassing 9P * broadcast mpipe tight coupling * collective mpipe tight coupling on BG/P * MAYBE: RDMA support * filter builtins for common scenarios (whitespace, EOL, etc) See Also: 9p(2) 9pfid(2) 9pfile(2) */ #include <u.h> #include <libc.h> #include <fcall.h> #include <thread.h> #include <9p.h> #include <mp.h> #include <libsec.h> #include <debug.h> enum { STACK = 8192, MSIZE = 8192, MAXARGS = 8, /* maximum spec arguments */ MAXHDRARGS = 4, /* maximum header entries */ }; Channel *iochan; /* dispatch channel */ typedef struct Fidaux Fidaux; char defaultsrvpath[] = "mpipe"; int iothread_id; 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)"; char Ebadspec[] = "bad mount specification arguments"; char Esplice[] = "splice couldnt open target"; char Ebadhdr[] = "unknown packet header"; /* * Right now there are two types of pipe: * 1) normal - writers match to readers * 2) broadcast - writers send to all readers * */ enum { MPTnormal = 1, /* strictly a normal pipe */ MPTbcast = 2, /* broadcast pipe */ MPThangingup, /* terminating */ MPThungup, MPTcleanup, }; /* * Mpipe - primary data structure * * There is one of these per pipe. Mostly it contains * the information necessary for file system queries on the * pipe, but also maintains a number of reference count values * relating to accounting for readers and writers. * * for non-enumerated pipes, there is only a single slot * */ typedef struct Mpipe Mpipe; struct Mpipe { QLock l; /* protect pipe */ int ref; /* reference count */ int mode; /* pipe type */ ulong path; /* qid.path */ char *name; /* name in directory */ char *uid; /* owner id */ char *gid; /* owner group */ ulong len; /* current length in bytes of data */ int writers; /* number of writers */ int readers; /* number of readers */ int slots; /* number of enumerations */ /* normal only, one entry per slot: */ Channel **rrchan; /* ready to read channels */ int *numr; /* number of readers on slot */ /* broadcast only */ Fidaux *bcastr; /* list of readers */ }; /* states for Fidaux */ enum { FID_IDLE, FID_CTLONLY, FID_WRITEN, FID_FLUSHED, }; /* * Fidaux - per particpant descriptor accounting structure * * This is typically hung off the fid private value and contains * the accounting for a particular "participant" (reader or writer) * */ struct Fidaux { QLock l; /* Protect it */ int state; /* if participant has been written yet */ Mpipe *mp; /* Multipipe we are dealing with */ 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; /* cached request for partial reads */ Channel *chan; /* Response channel */ /* for bcast */ Fidaux *next; /* Linked list of broadcasters */ }; /* * pipealloc - qid.path accounting * * we must guarentee unique qid.paths, this helps with that * */ static struct { Lock l; ulong path; } pipealloc; /* * splicearg - parameter encapsulation * * since splices get spawned off as sub-procs, this allows * us to pass all the arguments in a structure */ typedef struct Splicearg Splicearg; struct Splicearg { Mpipe *mp; /* the pipe in question */ int which; /* enumeration, 0 for don't care */ int fd; /* file descriptor to read from/write to */ char *path; /* name of target of splice */ Channel *chan; /* error/sync channel */ }; /* * Bcastr - Broadcast argument * * since broadcasts get spawned off as sub-threads, this allows * us to pass all the arguments in a structure */ typedef struct Bcastr Bcastr; struct Bcastr { Channel *chan; /* channel to send on */ Req *r; /* request to send */ }; static void usage(void) { fprint(2, "mpipefs [-D] [-v debuglevel] [-s <srvpt>] [mtpt]\n"); threadexits("usage"); } static void killall(Srv*) { DPRINT(DFID, "killing iochan and all thread\n"); chanfree(iochan); threadexitsall("killall"); } /* FIXME: prepending/adding the PID with pipealloc's path will not fit * within the lower 32 bits * * This might not scale well. Need to reexamine how the qid's are * generated and acounted for not only the local machine and the * synthetic file systems, but also for its scalability. */ #define MAGIC ((uvlong) 0x6163 << 32) #define NETTYPE(x) ((ulong)(x)&0x1f) #define NETID(x) ((((ulong)(x))>>5)&0xffff) #define NETQID(i,t) (MAGIC|((((i)&0xffff)<<5)|((getpid()&0x7ff)<<21)|(t))) enum { Qdir, Qdata, }; static 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; } /* * We need to send our responses down a channel * if one exists so that errors are propagated properly -- * particularly for broadcasts. */ static void myrespond(Req *r, char *err) { if(r->aux) sendp(r->aux, err); else respond(r, err); } /* * This only gets called from spliceto. */ static void myresponderror(Req *r) { if(r->aux) { sendp(r->aux, smprint("%r")); } 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 */ /* register a new reader with the infrastructure */ static void addnewreader(Mpipe *mp, Fidaux *aux) { int count; int min = mp->numr[0]; aux->chan = chancreate(sizeof(Req *), 0); if(aux->chan == nil) { DPRINT(DERR, "*ERROR*: addnewreader attempting to operate on a null chanel\n"); return ; } if(mp->mode == MPThangingup || mp->mode == MPThungup || mp->mode == MPTcleanup) { DPRINT(DBCA,"[%s](%p) WARNING: mp mode set to not running\n"); return; } qlock(&mp->l); mp->readers++; DPRINT(DRDR, "[%s](%p) addnewreader total now: %d\n", mp->name, mp, mp->readers); /* find enumeration slot with fewest readers */ aux->which = 0; for(count=0; count<mp->slots; count++) if(mp->numr[count] < min) aux->which = count; mp->numr[aux->which]++; if(mp->mode == MPTbcast) { DPRINT(DBCA,"[%s](%p) adding %p to broadcast list next: %p\n", mp->name, mp, aux, mp->bcastr); aux->next=mp->bcastr; mp->bcastr=aux; } qunlock(&mp->l); } /* safely remove a reader from the broadcast list */ 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); } /* cleanup broadcast pipes on last write close */ static void closebcasts(Mpipe *mp) { Fidaux *c; DPRINT(DFID, "closebcasts: pid=(%d)\n", getpid()); for(c=mp->bcastr; c != nil; c=c->next) { chanclose(c->chan); } } /* close everyone the mpipe has slotted */ static void closempipe(Mpipe *mp) { int count; DPRINT(DFID, "closempipe: pid=(%d)\n", getpid()); for(count=0;count < mp->slots; count++) if(chanclosing(mp->rrchan[count])) chanclose(mp->rrchan[count]); } /* process mount options */ static char * parsespec(Mpipe *mp, int argc, char **argv) { char *f; char *err=nil; ARGBEGIN { case 'e': /* enumerated mode */ f = ARGF(); if(f) { char *x = f; if(x) mp->slots = atoi(x); DPRINT(DARG, "(%p) enumerated mode (%d)\n", mp, mp->slots); } else { DPRINT(DERR, "(%p) *ERROR*: usage: invalid enumeration argument\n", mp); err = Ebadspec; } break; case 'b': /* broadcast mode */ DPRINT(DARG, "(%p) parsespec: BROADCAST mode\n", mp); mp->mode = MPTbcast; break; default: DPRINT(DERR, "(%p) *ERROR*: badflag('%c')", mp, ARGC()); err = Ebadspec; }ARGEND DPRINT(DARG, "(%p) argc: %d\n", mp, argc); if(argc>0) mp->name = estrdup9p(*argv); return err; } static void fsattach(Req *r) { Mpipe *mp = emalloc9pz(sizeof(Mpipe), 1); char *argv[MAXARGS]; int argc; int count; char *err; char *scratch; /* need to parse aname */ DPRINT(DARG, "args: (%s) len=%d\n", r->ifcall.aname, r->ifcall.count); /* add a stupid extra field so we can use args(2) */ scratch = smprint("mount %s\n", r->ifcall.aname); argc = tokenize(scratch, argv, MAXARGS); DPRINT(DARG, "argc: %d\n", argc); err = parsespec(mp, argc, argv); free(scratch); if(err) { respond(r, err); return; } 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++;