.

wcfs/__init__.py

@@ -565,7 +565,7 @@ class WatchLink(object):
             stream = wlink._req_next
             wlink._req_next = (wlink._req_next + 2) & ((1<<64)-1)
 
-        rxq = chan()
+        rxq = chan()    # -> XXX cap=1 so that we don't need to drain if _send fails
         with wlink._rxmu:
             assert stream not in wlink._rxtab   # XXX !test assert - recheck
             wlink._rxtab[stream] = rxq

wcfs/internal/wcfs_virtmem.cpp

@@ -406,13 +406,12 @@ error WatchLink::_send(StreamID stream, const string &msg) {
 //error WatchLink::_write(const Pkt *pkt) {
 error WatchLink::_write(const string &pkt) {
     WatchLink *wlink = this;
+
     wlink->_txmu.lock();
     defer([&]() {
         wlink->_txmu.unlock();
     })
 
-
-
     //printf('C: watch  : tx: %r' % pkt)
     int n;
     error err;
@@ -425,28 +424,38 @@ error WatchLink::_write(const string &pkt) {
 }
 
 // sendReq sends client -> server request and returns server reply.
-def WatchLink::sendReq(ctx, req) {   // -> reply | None when EOF
+// XXX -> reply | None when EOF
+tuple<string, error> WatchLink::sendReq(IContext *ctx, const string &req) {
     WatchLink *wlink = this;
-    rxq = wlink._sendReq(ctx, req)
-
-    _, _rx = select(
-        ctx.done().recv,    // 0
-        rxq.recv,           // 1
-    )
-    if _ == 0:
-        raise ctx.err()
-    return _rx
+    // XXX err ctx
+
+    rxPkt       rx;
+    chan<rxPkt> rxq;
+    error       err;
+    tie(rxq, err) = wlink._sendReq(ctx, req);
+    // XXX err
+
+    _ = select(
+        ctx.done().recvs(), // 0
+        rxq.recvs(&rx),     // 1
+    );
+    if (_ == 0)
+        return make_tuple("", ctx.err());
+
+    // XXX check for EOF
+    string reply = rx.XXXtostring();
+    return make_tuple(reply, nil);
 }
 
-def WatchLink::_sendReq(ctx, req) { // -> rxq
+tuple</*rxq*/chan<rxPkt>, error> WatchLink::_sendReq(IContext *ctx, const string &req) {
     WatchLink *wlink = this;
 
     wlink->_txmu.lock(); // XXX -> atomic (currently uses arbitrary lock)
         StreamID stream = wlink->_req_next;
-        wlink->_req_next = (wlink->_req_next + 2) & ((1<<64)-1);
+        wlink->_req_next = (wlink->_req_next + 2) & ((1ULL<<64)-1);
     wlink->_txmu.unlock();
 
-    rxq = chan()
+    auto rxq = makechan<rxPkt>(1);
     wlink->_rxmu.lock();
         if (has(wlink->rxtab, stream)) {
             wlink->_rxmu.unlock();
@@ -456,9 +465,17 @@ def WatchLink::_sendReq(ctx, req) { // -> rxq
     wlink->_rxmu.unlock();
 
     err = wlink->_send(stream, req);
-    // XXX err
+    if (err != nil) {
+        // remove rxq from rxtab
+        wlink->_rxmu.lock();
+        wlink->rxtab.erase(stream)
+        wlink->_rxmu.unlock();
+
+        // no need to drain rxq - it was created with cap=1
+        rxq = nil;
+    }
 
-    return rxq;
+    return make_tuple(rxq, err);
 }
 
 // ---- WCFS raw file access ----