exportfs: get rid of limits, cleanup

get rid of the service buffer limit. keep service buffers
on a global freelist protected by lock.

dont fatal when we hit the process limit. instead, just
abort the rpc with an error.

handle rendezvous() interrupts.

diff --git a/sys/src/cmd/exportfs/exportfs.c b/sys/src/cmd/exportfs/exportfs.c
index 821e01718ac394f0cb0171ee6e16d11a5f792240..c16220deeda26a8eba8d3847140af318fe368a00 100644 (file)
--- a/sys/src/cmd/exportfs/exportfs.c
+++ b/sys/src/cmd/exportfs/exportfs.c
@@ -76,9 +76,9 @@ noteconn(int fd)
        nci = getnetconninfo(nil, fd);
        if (nci == nil)
                return;
-       netdir = strdup(nci->dir);
-       local = strdup(nci->lsys);
-       remote = strdup(nci->rsys);
+       netdir = estrdup(nci->dir);
+       local = estrdup(nci->lsys);
+       remote = estrdup(nci->rsys);
        freenetconninfo(nci);
 }
 
@@ -237,8 +237,6 @@ main(int argc, char **argv)
                if(messagesize == 0)
                        messagesize = 8192+IOHDRSZ;
        }
-
-       Workq = emallocz(sizeof(Fsrpc)*Nr_workbufs);
        fhash = emallocz(sizeof(Fid*)*FHASHSIZE);
 
        fmtinstall('F', fcallfmt);
@@ -546,41 +544,49 @@ newfid(int nr)
        return new;     
 }
 
+static struct {
+       Lock;
+       Fsrpc   *free;
+
+       /* statistics */
+       int     nalloc;
+       int     nfree;
+}      sbufalloc;
+
 Fsrpc *
 getsbuf(void)
 {
-       static int ap;
-       int look, rounds;
-       Fsrpc *wb;
-       int small_instead_of_fast = 1;
-
-       if(small_instead_of_fast)
-               ap = 0; /* so we always start looking at the beginning and reuse buffers */
-
-       for(rounds = 0; rounds < 10; rounds++) {
-               for(look = 0; look < Nr_workbufs; look++) {
-                       if(++ap == Nr_workbufs)
-                               ap = 0;
-                       if(Workq[ap].busy == 0)
-                               break;
-               }
-
-               if(look == Nr_workbufs){
-                       sleep(10 * rounds);
-                       continue;
-               }
-
-               wb = &Workq[ap];
-               wb->pid = 0;
-               wb->canint = 0;
-               wb->flushtag = NOTAG;
-               wb->busy = 1;
-               if(wb->buf == nil)      /* allocate buffers dynamically to keep size down */
-                       wb->buf = emallocz(messagesize);
-               return wb;
+       Fsrpc *w;
+
+       lock(&sbufalloc);
+       w = sbufalloc.free;
+       if(w != 0){
+               sbufalloc.free = w->next;
+               w->next = nil;
+               sbufalloc.nfree--;
+               unlock(&sbufalloc);
+       } else {
+               sbufalloc.nalloc++;
+               unlock(&sbufalloc);
+               w = emallocz(sizeof(*w) + messagesize);
        }
-       fatal("No more work buffers");
-       return nil;
+       w->pid = 0;
+       w->canint = 0;
+       w->flushtag = NOTAG;
+       return w;
+}
+
+void
+putsbuf(Fsrpc *w)
+{
+       w->pid = 0;
+       w->canint = 0;
+       w->flushtag = NOTAG;
+       lock(&sbufalloc);
+       w->next = sbufalloc.free;
+       sbufalloc.free = w;
+       sbufalloc.nfree++;
+       unlock(&sbufalloc);
 }
 
 void
@@ -711,9 +717,7 @@ makepath(File *p, char *name)
                seg[i] = p->name;
                n += strlen(p->name)+1;
        }
-       path = malloc(n);
-       if(path == nil)
-               fatal("out of memory");
+       path = emallocz(n);
        s = path;
 
        while(i--) {

diff --git a/sys/src/cmd/exportfs/exportfs.h b/sys/src/cmd/exportfs/exportfs.h
index 0f7ba3e71e0485ed981156ef42567675c6d29fef..77709d0d97add6831df80b8061960cc5194d79c9 100644 (file)
--- a/sys/src/cmd/exportfs/exportfs.h
+++ b/sys/src/cmd/exportfs/exportfs.h
@@ -14,12 +14,12 @@ typedef struct Qidtab Qidtab;
 
 struct Fsrpc
 {
-       int     busy;           /* Work buffer has pending rpc to service */
+       Fsrpc   *next;          /* freelist */
        uintptr pid;            /* Pid of slave process executing the rpc */
        int     canint;         /* Interrupt gate */
        int     flushtag;       /* Tag on which to reply to flush */
        Fcall   work;           /* Plan 9 incoming Fcall */
-       uchar   *buf;           /* Data buffer */
+       uchar   buf[];          /* Data buffer */
 };
 
 struct Fid
@@ -54,7 +54,7 @@ struct File
 struct Proc
 {
        uintptr pid;
-       int     busy;
+       Fsrpc   *busy;
        Proc    *next;
 };
 
@@ -72,7 +72,6 @@ enum
 {
        MAXPROC         = 50,
        FHASHSIZE       = 64,
-       Nr_workbufs     = 50,
        Fidchunk        = 1000,
        Npsmpt          = 32,
        Nqidbits                = 5,
@@ -88,7 +87,6 @@ char Enomem[];
 char Emip[];
 char Enopsmt[];
 
-Extern Fsrpc   *Workq;
 Extern int     dbg;
 Extern File    *root;
 Extern File    *psmpt;
@@ -121,6 +119,7 @@ Fid         *getfid(int);
 int    freefid(int);
 Fid    *newfid(int);
 Fsrpc  *getsbuf(void);
+void   putsbuf(Fsrpc*);
 void   initroot(void);
 void   fatal(char*, ...);
 char*  makepath(File*, char*);

diff --git a/sys/src/cmd/exportfs/exportsrv.c b/sys/src/cmd/exportfs/exportsrv.c
index 058e841934314911442bed83ecb47cdc1ef36aaf..d7a0ad290374e21d9dfd44ae84ba6ed64eaaacda 100644 (file)
--- a/sys/src/cmd/exportfs/exportsrv.c
+++ b/sys/src/cmd/exportfs/exportsrv.c
@@ -17,6 +17,7 @@ char Enopsmt[] = "Out of pseudo mount points";
 char Enomem[] = "No memory";
 char Eversion[] = "Bad 9P2000 version";
 char Ereadonly[] = "File system read only";
+char Enoprocs[] = "Out of processes";
 
 ulong messagesize;
 int readonly;
@@ -28,7 +29,7 @@ Xversion(Fsrpc *t)
 
        if(t->work.msize < 256){
                reply(&t->work, &rhdr, "version: message size too small");
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        if(t->work.msize > messagesize)
@@ -36,13 +37,13 @@ Xversion(Fsrpc *t)
        messagesize = t->work.msize;
        if(strncmp(t->work.version, "9P2000", 6) != 0){
                reply(&t->work, &rhdr, Eversion);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        rhdr.version = "9P2000";
        rhdr.msize = t->work.msize;
        reply(&t->work, &rhdr, 0);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -51,34 +52,31 @@ Xauth(Fsrpc *t)
        Fcall rhdr;
 
        reply(&t->work, &rhdr, "exportfs: authentication not required");
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
 Xflush(Fsrpc *t)
 {
-       Fsrpc *w, *e;
        Fcall rhdr;
+       Fsrpc *w;
+       Proc *m;
 
-       e = &Workq[Nr_workbufs];
-
-       for(w = Workq; w < e; w++) {
-               if(w->work.tag == t->work.oldtag) {
-                       DEBUG(DFD, "\tQ busy %d pid %p can %d\n", w->busy, w->pid, w->canint);
-                       if(w->busy && w->pid) {
-                               w->flushtag = t->work.tag;
-                               DEBUG(DFD, "\tset flushtag %d\n", t->work.tag);
-                               if(w->canint)
-                                       postnote(PNPROC, w->pid, "flush");
-                               t->busy = 0;
-                               return;
-                       }
+       for(m = Proclist; m; m = m->next){
+               w = m->busy;
+               if(w != 0 && w->pid == m->pid && w->work.tag == t->work.oldtag) {
+                       w->flushtag = t->work.tag;
+                       DEBUG(DFD, "\tset flushtag %d\n", t->work.tag);
+                       if(w->canint)
+                               postnote(PNPROC, w->pid, "flush");
+                       putsbuf(t);
+                       return;
                }
        }
 
        reply(&t->work, &rhdr, 0);
        DEBUG(DFD, "\tflush reply\n");
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -92,7 +90,7 @@ Xattach(Fsrpc *t)
        f = newfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -100,8 +98,8 @@ Xattach(Fsrpc *t)
                if(psmpt == 0){
                Nomount:
                        reply(&t->work, &rhdr, Enopsmt);
-                       t->busy = 0;
                        freefid(t->work.fid);
+                       putsbuf(t);
                        return;
                }
                for(i=0; i<Npsmpt; i++)
@@ -118,8 +116,8 @@ Xattach(Fsrpc *t)
                if(amount(nfd, buf, MREPL|MCREATE, t->work.aname) < 0){
                        errstr(buf, sizeof buf);
                        reply(&t->work, &rhdr, buf);
-                       t->busy = 0;
                        freefid(t->work.fid);
+                       putsbuf(t);
                        close(nfd);
                        return;
                }
@@ -132,7 +130,7 @@ Xattach(Fsrpc *t)
 
        rhdr.qid = f->f->qid;
        reply(&t->work, &rhdr, 0);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 Fid*
@@ -169,7 +167,7 @@ Xwalk(Fsrpc *t)
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -215,7 +213,7 @@ Xwalk(Fsrpc *t)
        if(rhdr.nwqid > 0)
                e = nil;
        reply(&t->work, &rhdr, e);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -227,7 +225,7 @@ Xclunk(Fsrpc *t)
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -236,7 +234,7 @@ Xclunk(Fsrpc *t)
 
        freefid(t->work.fid);
        reply(&t->work, &rhdr, 0);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -252,7 +250,7 @@ Xstat(Fsrpc *t)
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        if(f->fid >= 0)
@@ -266,7 +264,7 @@ Xstat(Fsrpc *t)
        if(d == nil) {
                errstr(err, sizeof err);
                reply(&t->work, &rhdr, err);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -279,7 +277,7 @@ Xstat(Fsrpc *t)
        rhdr.stat = statbuf;
        reply(&t->work, &rhdr, 0);
        free(statbuf);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 static int
@@ -303,13 +301,13 @@ Xcreate(Fsrpc *t)
 
        if(readonly) {
                reply(&t->work, &rhdr, Ereadonly);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        
@@ -320,7 +318,7 @@ Xcreate(Fsrpc *t)
        if(f->fid < 0) {
                errstr(err, sizeof err);
                reply(&t->work, &rhdr, err);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -328,7 +326,7 @@ Xcreate(Fsrpc *t)
        if(nf == 0) {
                errstr(err, sizeof err);
                reply(&t->work, &rhdr, err);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -338,7 +336,7 @@ Xcreate(Fsrpc *t)
        rhdr.qid = f->f->qid;
        rhdr.iounit = getiounit(f->fid);
        reply(&t->work, &rhdr, 0);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -350,13 +348,13 @@ Xremove(Fsrpc *t)
 
        if(readonly) {
                reply(&t->work, &rhdr, Ereadonly);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
 
@@ -366,7 +364,7 @@ Xremove(Fsrpc *t)
                free(path);
                errstr(err, sizeof err);
                reply(&t->work, &rhdr, err);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        free(path);
@@ -377,7 +375,7 @@ Xremove(Fsrpc *t)
        freefid(t->work.fid);
 
        reply(&t->work, &rhdr, 0);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 void
@@ -392,20 +390,20 @@ Xwstat(Fsrpc *t)
 
        if(readonly) {
                reply(&t->work, &rhdr, Ereadonly);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        f = getfid(t->work.fid);
        if(f == 0) {
                reply(&t->work, &rhdr, Ebadfid);
-               t->busy = 0;
+               putsbuf(t);
                return;
        }
        strings = emallocz(t->work.nstat);      /* ample */
        if(convM2D(t->work.stat, t->work.nstat, &d, strings) <= BIT16SZ){
                rerrstr(err, sizeof err);
                reply(&t->work, &rhdr, err);
-               t->busy = 0;
+               putsbuf(t);
                free(strings);
                return;
        }
@@ -430,7 +428,7 @@ Xwstat(Fsrpc *t)
                reply(&t->work, &rhdr, 0);
        }
        free(strings);
-       t->busy = 0;
+       putsbuf(t);
 }
 
 /*
@@ -470,12 +468,12 @@ slave(Fsrpc *f)
                switch(f->work.type){
                case Twrite:
                        reply(&f->work, &rhdr, Ereadonly);
-                       f->busy = 0;
+                       putsbuf(f);
                        return;
                case Topen:
                        if((f->work.mode&3) == OWRITE || (f->work.mode&OTRUNC)){
                                reply(&f->work, &rhdr, Ereadonly);
-                               f->busy = 0;
+                               putsbuf(f);
                                return;
                        }
                }
@@ -484,17 +482,23 @@ slave(Fsrpc *f)
                for(p = Proclist; p; p = p->next) {
                        if(p->busy == 0) {
                                f->pid = p->pid;
-                               p->busy = 1;
-                               pid = (uintptr)rendezvous((void*)p->pid, f);
+                               p->busy = f;
+                               do {
+                                       pid = (uintptr)rendezvous((void*)p->pid, f);
+                               }
+                               while(pid == ~0);       /* Interrupted */
                                if(pid != p->pid)
                                        fatal("rendezvous sync fail");
                                return;
                        }       
                }
 
-               if(++nproc > MAXPROC)
-                       fatal("too many procs");
-
+               if(nproc >= MAXPROC){
+                       reply(&f->work, &rhdr, Enoprocs);
+                       putsbuf(f);
+                       return;
+               }
+               nproc++;
                pid = rfork(RFPROC|RFMEM);
                switch(pid) {
                case -1:
@@ -511,16 +515,13 @@ slave(Fsrpc *f)
                        fatal("slave");
 
                default:
-                       p = malloc(sizeof(Proc));
-                       if(p == 0)
-                               fatal("out of memory");
-
+                       p = emallocz(sizeof(Proc));
                        p->busy = 0;
                        p->pid = pid;
                        p->next = Proclist;
                        Proclist = p;
-
-                       rendezvous((void*)pid, p);
+                       while(rendezvous((void*)pid, p) == (void*)~0)
+                               ;
                }
        }
 }
@@ -537,14 +538,17 @@ blockingslave(void)
 
        pid = getpid();
 
-       m = rendezvous((void*)pid, 0);
+       do {
+               m = rendezvous((void*)pid, 0);
+       }
+       while(m == (void*)~0);  /* Interrupted */
        
        for(;;) {
                p = rendezvous((void*)pid, (void*)pid);
                if(p == (void*)~0)                      /* Interrupted */
                        continue;
 
-               DEBUG(DFD, "\tslave: %p %F b %d p %p\n", pid, &p->work, p->busy, p->pid);
+               DEBUG(DFD, "\tslave: %p %F p %p\n", pid, &p->work, p->pid);
                if(p->flushtag != NOTAG)
                        goto flushme;
 
@@ -570,8 +574,8 @@ flushme:
                        p->work.tag = p->flushtag;
                        reply(&p->work, &rhdr, 0);
                }
-               p->busy = 0;
                m->busy = 0;
+               putsbuf(p);
        }
 }
 
@@ -703,8 +707,10 @@ slaveread(Fsrpc *p)
        if(p->flushtag != NOTAG)
                return;
        data = malloc(n);
-       if(data == nil)
-               fatal(Enomem);
+       if(data == 0) {
+               reply(work, &rhdr, Enomem);
+               return;
+       }
 
        /* can't just call pread, since directories must update the offset */
        if(patternfile != nil && (f->f->qid.type&QTDIR))

diff --git a/sys/src/cmd/exportfs/pattern.c b/sys/src/cmd/exportfs/pattern.c
index 91abfd755175cf2e3933803b45d12af9c67595f4..4dfed5bca67855914709f67cef963c9cc3ea8a0d 100644 (file)
--- a/sys/src/cmd/exportfs/pattern.c
+++ b/sys/src/cmd/exportfs/pattern.c
@@ -24,15 +24,11 @@ exclusions(void)
                fatal("cannot open patternfile");
        ni = 0;
        nmaxi = 100;
-       include = malloc(nmaxi*sizeof(*include));
-       if(include == nil)
-               fatal("out of memory");
+       include = emallocz(nmaxi*sizeof(*include));
        include[0] = nil;
        ne = 0;
        nmaxe = 100;
-       exclude = malloc(nmaxe*sizeof(*exclude));
-       if(exclude == nil)
-               fatal("out of memory");
+       exclude = emallocz(nmaxe*sizeof(*exclude));
        exclude[0] = nil;
        while(line = Brdline(f, '\n')){
                line[Blinelen(f) - 1] = 0;