rio: properly handle follow up flushes (fixes unexpected reply tag)

when multiple flushes are send, they need to be replied
in order. we ensure this by having the flush xfid taking
over the flushtag (synchronized with a new fs->csyncflush
channel) so the next flush will flush the previous flush.

4 files changed, 53 insertions, 20 deletions

diff --git a/sys/src/cmd/rio/dat.h b/sys/src/cmd/rio/dat.h
index f4f7f4efa..dd7d560d5 100644
--- a/sys/src/cmd/rio/dat.h
+++ b/sys/src/cmd/rio/dat.h
@@ -285,6 +285,7 @@ struct Filsys
 		int		pid;
 		char		*user;
 		Channel	*cxfidalloc;	/* chan(Xfid*) */
+		Channel	*csyncflush;	/* chan(int) */
 		Fid		*fids[Nhash];
 };
 
@@ -357,3 +358,4 @@ int		menuing;		/* menu action is pending; waiting for window to be indicated */
 int		snarfversion;	/* updated each time it is written */
 int		messagesize;		/* negotiated in 9P version setup */
 int		shiftdown;
+int		debug;
diff --git a/sys/src/cmd/rio/fsys.c b/sys/src/cmd/rio/fsys.c
index f4fea7723..5e680ea85 100644
--- a/sys/src/cmd/rio/fsys.c
+++ b/sys/src/cmd/rio/fsys.c
@@ -18,10 +18,6 @@ char	Eoffset[] = "illegal offset";
 
 int	messagesize = 8192+IOHDRSZ;	/* good start */
 
-enum{
-	DEBUG = 0
-};
-
 Dirtab dirtab[]=
 {
 	{ ".",			QTDIR,	Qdir,			0500|DMDIR },
@@ -142,6 +138,9 @@ filsysinit(Channel *cxfidalloc)
 		close(fd);
 	}
 	fs->user = estrdup(buf);
+	fs->csyncflush = chancreate(sizeof(int), 0);
+	if(fs->csyncflush == nil)
+		error("chancreate syncflush");
 	fs->cxfidalloc = cxfidalloc;
 	pid = getpid();
 
@@ -210,7 +209,7 @@ filsysproc(void *arg)
 		x->buf = buf;
 		if(convM2S(buf, n, x) != n)
 			error("convert error in convM2S");
-		if(DEBUG)
+		if(debug)
 			fprint(2, "rio:<-%F\n", &x->Fcall);
 		if(fcall[x->type] == nil)
 			x = filsysrespond(fs, x, &t, Ebadfcall);
@@ -266,7 +265,7 @@ filsysrespond(Filsys *fs, Xfid *x, Fcall *t, char *err)
 		error("convert error in convS2M");
 	if(write(fs->sfd, x->buf, n) != n)
 		error("write error in respond");
-	if(DEBUG)
+	if(debug)
 		fprint(2, "rio:->%F\n", t);
 	free(x->buf);
 	x->buf = nil;
@@ -307,14 +306,21 @@ filsysauth(Filsys *fs, Xfid *x, Fid*)
 {
 	Fcall t;
 
-		return filsysrespond(fs, x, &t, "rio: authentication not required");
+	return filsysrespond(fs, x, &t, "rio: authentication not required");
 }
 
 static
 Xfid*
-filsysflush(Filsys*, Xfid *x, Fid*)
+filsysflush(Filsys *fs, Xfid *x, Fid*)
 {
 	sendp(x->c, xfidflush);
+
+	/*
+	 * flushes need to be replied in order. xfidflush() will
+	 * awaken us when the flush has been queued.
+	 */
+	recv(fs->csyncflush, nil);
+
 	return nil;
 }
 
diff --git a/sys/src/cmd/rio/rio.c b/sys/src/cmd/rio/rio.c
index c65af8891..e6537283f 100644
--- a/sys/src/cmd/rio/rio.c
+++ b/sys/src/cmd/rio/rio.c
@@ -156,6 +156,11 @@ threadmain(int argc, char *argv[])
 	case 's':
 		scrolling = TRUE;
 		break;
+	case 'D':
+		debug++;
+		break;
+	default:
+		usage();
 	}ARGEND
 
 	if(getwd(buf, sizeof buf) == nil)
diff --git a/sys/src/cmd/rio/xfid.c b/sys/src/cmd/rio/xfid.c
index b505b2f06..662d686cf 100644
--- a/sys/src/cmd/rio/xfid.c
+++ b/sys/src/cmd/rio/xfid.c
@@ -124,24 +124,44 @@ xfidflush(Xfid *x)
 
 	for(xf=xfid; xf; xf=xf->next)
 		if(xf->flushtag == x->oldtag){
-			xf->flushtag = -1;
-			xf->flushing = TRUE;
 			incref(xf);	/* to hold data structures up at tail of synchronization */
 			if(xf->ref == 1)
 				error("ref 1 in flush");
-			if(canqlock(&xf->active)){
-				qunlock(&xf->active);
-				sendul(xf->flushc, 0);
-			}else{
-				qlock(&xf->active);	/* wait for him to finish */
-				qunlock(&xf->active);
-			}
-			xf->flushing = FALSE;
-			if(decref(xf) == 0)
-				sendp(cxfidfree, xf);
+			/* take over flushtag so follow up flushes wait for us */
+			x->flushtag = x->oldtag;
+			xf->flushtag = -1;
 			break;
 		}
+
+	/*
+	 * wakeup filsysflush() in the filsysproc so the next
+	 * flush can come in.
+	 */
+	sendul(x->fs->csyncflush, 0);
+
+	if(xf){
+		qlock(&xf->active);
+		if(xf->buf){	/* not responded yet? */
+			xf->flushing = TRUE;
+			qunlock(&xf->active);
+			sendul(xf->flushc, 0);
+			xf->flushing = FALSE;
+		}else{
+			qunlock(&xf->active);
+		}
+		if(decref(xf) == 0)
+			sendp(cxfidfree, xf);
+	}
+
+	qlock(&x->active);
+	if(x->flushing){
+		qunlock(&x->active);
+		recv(x->flushc, nil);	/* wakeup flushing xfid */
+		filsyscancel(x);
+		return;
+	}
 	filsysrespond(x->fs, x, &t, nil);
+	qunlock(&x->active);
 }
 
 void