rio: huge simplification of flush handling

xfidflush() now uses a simpler way to flush another xfid.

when we can send to Xfid.c channel, we know that the
xfid is completed. so all xfidflush() needs todo is to
do an alt sending to both Xfid.c and Xfid.flushc.

once the send to Xfid.c succeeds, we know the xfid is gone
and we can respond to the flush request.

theres no need to keep track of flush state anymore. so
Xfid.active and Xfid.flushing can be removed.

note, that it is ok to respond normally to a currently
flushed request.

diff --git a/sys/src/cmd/rio/dat.h b/sys/src/cmd/rio/dat.h
index dd7d560d530508bc2c0316329012b9de5a98adf8..454fcd9aceff1251fec53e63f011779eb187c8b4 100644 (file)
--- a/sys/src/cmd/rio/dat.h
+++ b/sys/src/cmd/rio/dat.h
@@ -258,8 +258,6 @@ struct Xfid
                Fid             *f;
                uchar   *buf;
                Filsys  *fs;
-               QLock   active;
-               int             flushing;       /* another Xfid is trying to flush us */
                int             flushtag;       /* our tag, so flush can find us */
                Channel *flushc;        /* channel(int) to notify us we're being flushed */
 };

diff --git a/sys/src/cmd/rio/fsys.c b/sys/src/cmd/rio/fsys.c
index 41db21b96ceaec9bb66fde2db01529235ea86019..5e680ea85920152d5c1206caff833264120b812e 100644 (file)
--- a/sys/src/cmd/rio/fsys.c
+++ b/sys/src/cmd/rio/fsys.c
@@ -270,8 +270,6 @@ filsysrespond(Filsys *fs, Xfid *x, Fcall *t, char *err)
        free(x->buf);
        x->buf = nil;
        x->flushtag = -1;
-       if(x->flushing)
-               recv(x->flushc, nil);   /* wakeup flushing xfid */
        return x;
 }
 

diff --git a/sys/src/cmd/rio/xfid.c b/sys/src/cmd/rio/xfid.c
index be7c65d9c240fcab9d610fe193f2c88383508114..7cf7eadb6cd3fdd82ec2dcb07c278e5d6d513dbe 100644 (file)
--- a/sys/src/cmd/rio/xfid.c
+++ b/sys/src/cmd/rio/xfid.c
@@ -70,8 +70,6 @@ xfidallocthread(void*)
                                fprint(2, "%p incref %ld\n", x, x->ref);
                                error("incref");
                        }
-                       if(x->flushing)
-                               error("flushing in allocate");
                        if(x->flushtag != -1)
                                error("flushtag in allocate");
                        incref(x);
@@ -82,8 +80,6 @@ xfidallocthread(void*)
                                fprint(2, "%p decref %ld\n", x, x->ref);
                                error("decref");
                        }
-                       if(x->flushing)
-                               error("flushing in free");
                        if(x->flushtag != -1)
                                error("flushtag in free");
                        x->free = xfidfree;
@@ -114,8 +110,10 @@ xfidctl(void *arg)
        threadsetname(buf);
        for(;;){
                f = recvp(x->c);
-               x->flushtag = x->tag;
-               (*f)(x);
+               if(f){
+                       x->flushtag = x->tag;
+                       (*f)(x);
+               }
                if(decref(x) == 0)
                        sendp(cxfidfree, x);
        }
@@ -146,28 +144,30 @@ xfidflush(Xfid *x)
        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);
+               enum { Done, Flush, End };
+               Alt alts[End+1];
+               void *f;
+               int z;
+
+               z = 0;
+               f = nil;
+
+               alts[Done].c = xf->c;
+               alts[Done].v = &f;
+               alts[Done].op = CHANSND;
+               alts[Flush].c = xf->flushc;
+               alts[Flush].v = &z;
+               alts[Flush].op = CHANSND;
+               alts[End].op = CHANEND;
+
+               while(alt(alts) != Done)
+                       ;
        }
-
-       qlock(&x->active);
-       if(x->flushing){
-               qunlock(&x->active);
-               recv(x->flushc, nil);   /* wakeup flushing xfid */
+       if(nbrecv(x->flushc, nil)){
                filsyscancel(x);
                return;
        }
        filsysrespond(x->fs, x, &t, nil);
-       qunlock(&x->active);
 }
 
 void
@@ -439,16 +439,9 @@ xfidwrite(Xfid *x)
                case CWdata:
                        break;
                case CWgone:
-                       qlock(&x->active);
-                       if(!x->flushing){
-                               filsysrespond(x->fs, x, &fc, Edeleted);
-                               qunlock(&x->active);
-                               free(r);
-                               return;
-                       }
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       /* no break */
+                       filsysrespond(x->fs, x, &fc, Edeleted);
+                       free(r);
+                       return;
                case CWflush:
                        free(r);
                        filsyscancel(x);
@@ -456,23 +449,11 @@ xfidwrite(Xfid *x)
                }
 
                /* received data */
-               qlock(&x->active);
-               if(x->flushing){
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       free(r);
-                       pair.s = runemalloc(1);
-                       pair.ns = 0;
-                       send(cwm.cw, &pair);            /* wake up window with empty data */
-                       filsyscancel(x);
-                       return;
-               }
                pair.s = r;
                pair.ns = nr;
                send(cwm.cw, &pair);
                fc.count = x->count;
                filsysrespond(x->fs, x, &fc, nil);
-               qunlock(&x->active);
                return;
 
        case Qconsctl:
@@ -678,15 +659,8 @@ xfidread(Xfid *x)
                case CRdata:
                        break;
                case CRgone:
-                       qlock(&x->active);
-                       if(!x->flushing){
-                               filsysrespond(x->fs, x, &fc, Edeleted);
-                               qunlock(&x->active);
-                               return;
-                       }
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       /* no break */
+                       filsysrespond(x->fs, x, &fc, Edeleted);
+                       return;
                case CRflush:
                        filsyscancel(x);
                        return;
@@ -699,20 +673,10 @@ xfidread(Xfid *x)
                pair.s = t;
                pair.ns = cnt;
                send(c1, &pair);
-               qlock(&x->active);
-               if(x->flushing){
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       recv(c2, nil);                  /* wake up window and toss data */
-                       free(t);
-                       filsyscancel(x);
-                       return;
-               }
                recv(c2, &pair);
                fc.data = pair.s;
                fc.count = pair.ns;
                filsysrespond(x->fs, x, &fc, nil);
-               qunlock(&x->active);
                free(t);
                break;
 
@@ -743,26 +707,9 @@ xfidread(Xfid *x)
                case MRdata:
                        break;
                case MRgone:
-                       qlock(&x->active);
-                       if(!x->flushing){
-                               filsysrespond(x->fs, x, &fc, Edeleted);
-                               qunlock(&x->active);
-                               return;
-                       }
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       /* no break */
-               case MRflush:
-                       filsyscancel(x);
+                       filsysrespond(x->fs, x, &fc, Edeleted);
                        return;
-               }
-
-               /* received data */
-               qlock(&x->active);
-               if(x->flushing){
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);           /* wake up flushing xfid */
-                       recv(mrm.cm, nil);                      /* wake up window and toss data */
+               case MRflush:
                        filsyscancel(x);
                        return;
                }
@@ -775,7 +722,6 @@ xfidread(Xfid *x)
                fc.data = buf;
                fc.count = min(n, cnt);
                filsysrespond(x->fs, x, &fc, nil);
-               qunlock(&x->active);
                break;
 
        case Qkbd:
@@ -794,15 +740,8 @@ xfidread(Xfid *x)
                case MRdata:
                        break;
                case MRgone:
-                       qlock(&x->active);
-                       if(!x->flushing){
-                               filsysrespond(x->fs, x, &fc, Edeleted);
-                               qunlock(&x->active);
-                               return;
-                       }
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       /* no break */
+                       filsysrespond(x->fs, x, &fc, Edeleted);
+                       return;
                case MRflush:
                        filsyscancel(x);
                        return;
@@ -810,18 +749,9 @@ xfidread(Xfid *x)
 
                /* received data */
                t = recvp(krm.ck);
-               qlock(&x->active);
-               if(x->flushing){
-                       qunlock(&x->active);
-                       free(t);                /* toss data */
-                       recv(x->flushc, nil);           /* wake up flushing xfid */
-                       filsyscancel(x);
-                       return;
-               }
                fc.data = t;
                fc.count = strlen(t)+1;
                filsysrespond(x->fs, x, &fc, nil);
-               qunlock(&x->active);
                free(t);
                break;
 
@@ -934,15 +864,8 @@ xfidread(Xfid *x)
                case WCRdata:
                        break;
                case WCRgone:
-                       qlock(&x->active);
-                       if(!x->flushing){
-                               filsysrespond(x->fs, x, &fc, Edeleted);
-                               qunlock(&x->active);
-                               return;
-                       }
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       /* no break */
+                       filsysrespond(x->fs, x, &fc, Edeleted);
+                       return;
                case WCRflush:
                        filsyscancel(x);
                        return;
@@ -955,22 +878,12 @@ xfidread(Xfid *x)
                pair.s = t;
                pair.ns = cnt+1;
                send(c1, &pair);
-               qlock(&x->active);
-               if(x->flushing){
-                       qunlock(&x->active);
-                       recv(x->flushc, nil);   /* wake up flushing xfid */
-                       recv(c2, nil);                  /* wake up window and toss data */
-                       free(t);
-                       filsyscancel(x);
-                       return;
-               }
                recv(c2, &pair);
                fc.data = pair.s;
                if(pair.ns > cnt)
                        pair.ns = cnt;
                fc.count = pair.ns;
                filsysrespond(x->fs, x, &fc, nil);
-               qunlock(&x->active);
                free(t);
                break;