.

wcfs/__init__.py

@@ -474,6 +474,7 @@ class WatchLink(object):
         wlink._wrx.close()
 
         # disable all established watches
+        # XXX test only -> tWatchLiink
         for w in wlink._watching.values():
             w.at     = z64
             w.pinned = {}

wcfs/internal/wcfs_virtmem.cpp

@@ -192,7 +192,7 @@ class WatchLink {
 
     StreamID        _req_next;  // stream ID for next client-originated request XXX -> atomic
     sync::Mutex     _txmu;      // serializes writes
-    bool            _txclosed;  // XXX -> _txcloseOnce
+    sync::Once      _txclose1;
 
 #if 0
     func() _serveCancel
@@ -201,10 +201,12 @@ class WatchLink {
 
 public:
     friend tuple<WatchLink*, error> WCFS::_openwatch();
+    error close();
     SrvReq *recvReq(IContext *ctx);
     tuple<string, error> sendReq(IContext *ctx, const string &req);
 
 private:
+    void _closeTX();
     error _send(StreamID stream, const string &msg);
     error _write(const string &pkt);
     tuple<chan<rxPkt>, error> _sendReq(IContext *ctx, const string &req);
@@ -392,7 +394,6 @@ tuple<WatchLink*, error> WCFS::_openwatch() {
     wlink->_rx_eof    = makechan<structZ>();
     wlink->_acceptq   = makechan<rxPkt>();
     wlink->_req_next  = 1;
-    wlink->_txclosed  = false;
 
 #if 0
     serveCtx, wlink._serveCancel = context.with_cancel(context.background())
@@ -403,7 +404,40 @@ tuple<WatchLink*, error> WCFS::_openwatch() {
     return make_tuple(wlink, nil);
 }
 
-// XXX close
+void WatchLink::_closeTX() {
+    WatchLink &wlink = *this;
+    wlink._txclose1.do_([&]() {
+        // ask wcfs to close its tx & rx sides; close(wcfs.tx) wakes up
+        // _serveRX on client (= on us). The connection can be already closed
+        // by wcfs - so ignore errors when sending bye.
+        (void)wlink._send(1, "bye");
+        wlink._wtx.close(); // XXX ._f.closeTX() ?  XXX err
+    });
+}
+
+// close closes the link.
+error WatchLink::close() {
+    WatchLink& wlink = *this;
+
+    wlink._closeTX();
+    wlink._serveCancel();
+    // XXX we can get stuck here if wcfs does not behave as we want.
+    // XXX in particular if there is a silly - e.g. syntax or type error in
+    //     test code - we currently get stuck here.
+    //
+    // XXX -> better pthread_kill(SIGINT) instead of relying on wcfs proper behaviour?
+    // XXX -> we now have `kill -QUIT` to wcfs.go on test timeout - remove ^^^ comments?
+    error err = wlink._serveWG.wait();
+    // canceled is expected and ok
+    if (err == context.canceled)
+        err = nil;
+
+    error err2 = wlink._wrx.close(); // XXX err;
+    if (err == nil)
+        err = err2;
+
+    return err;
+}
 
 // _send sends raw message via specified stream.
 //