.

wcfs/internal/wcfs_virtmem.cpp

@@ -51,6 +51,12 @@ using namespace golang;
 template<typename Key, typename Value>
 using dict = std::unordered_map<Key, Value>;
 
+// has returns whether container d (e.g. dict) has element k.
+template<typename Container, typename Key>
+bool has(const Container &d, Key k) {
+    return d.find(k) != d.end();
+}
+
 template<typename Key>
 using set  = std::unordered_set<Key>;
 
@@ -241,8 +247,7 @@ void Conn::_pin1(SrvReq *req) {
     // XXX return error?
 
     wconn->_filemu.lock();
-    auto _ = wconn->_filetab.find(req->foid);
-    if (_ == wconn->_filetab.end()) {
+    if (has(wconn->_filetab, req->foid)) {
         wconn->_filemu.unlock();
         // XXX err = we are not watching the file - why wcfs sent us this update?
     }
@@ -337,6 +342,7 @@ error _Mapping::_remmapblk(int64_t blk, Tid at) {
 // ---- WatchLink ----
 
 // Pkt internally represents data of one message sent/received over WatchLink.
+// XXX used only for recv?
 struct Pkt {
     // stream over which the data was received; used internally by send
     StreamID stream;
@@ -393,7 +399,8 @@ error WatchLink::_send(StreamID stream, const string &msg) {
     pkt.rawdata = // XXX copy msg   XXX + "%d <stream> ...\n" ?
     return wlink->_write(&pkt)
 #endif
-    pkt = fmt::sprintf("%ul %s\n", stream, msg.c_str());
+    string pkt = fmt::sprintf("%ul %s\n", stream, msg.c_str());
+    return wlink->_write(pkt);
 }
 
 //error WatchLink::_write(const Pkt *pkt) {
@@ -417,6 +424,43 @@ error WatchLink::_write(const string &pkt) {
 #endif
 }
 
+// sendReq sends client -> server request and returns server reply.
+def WatchLink::sendReq(ctx, req) {   // -> reply | None when EOF
+    WatchLink *wlink = this;
+    rxq = wlink._sendReq(ctx, req)
+
+    _, _rx = select(
+        ctx.done().recv,    // 0
+        rxq.recv,           // 1
+    )
+    if _ == 0:
+        raise ctx.err()
+    return _rx
+}
+
+def WatchLink::_sendReq(ctx, req) { // -> rxq
+    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->_txmu.unlock();
+
+    rxq = chan()
+    wlink->_rxmu.lock();
+        if (has(wlink->rxtab, stream)) {
+            wlink->_rxmu.unlock();
+            panic("BUG: to-be-sent stream is present in rxtab");
+        }
+        wlink->_rxtab[stream] = rxq;
+    wlink->_rxmu.unlock();
+
+    err = wlink->_send(stream, req);
+    // XXX err
+
+    return rxq;
+}
+
 // ---- WCFS raw file access ----
 
 // _path returns path for object on wcfs.