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.

4 files changed, 116 insertions, 111 deletions

diff --git a/sys/src/cmd/exportfs/exportfs.c b/sys/src/cmd/exportfs/exportfs.c
index 821e01718..c16220dee 100644
--- 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 0f7ba3e71..77709d0d9 100644
--- 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 058e84193..d7a0ad290 100644
--- 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 91abfd755..4dfed5bca 100644
--- 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;