.

wcfs/wcfs_test.py

@@ -225,15 +225,20 @@ class tDB:
 
     # open opens wcfs file corresponding to zf@at and starts to track it.
     # see returned tFile for details.
-    def open(t, zf, at=None):
+    def open(t, zf, at=None):   # -> tFile
         return tFile(t, zf, at=at)
 
+    # openwatch opens /head/watch on wcfs.
+    # see returned tWatch for details.
+    def openwatch(t):   # -> tWatch
+        return tWatch(t)
+
 
 # tFile is testing environment for one bigfile on wcfs.
 class tFile:
     # maximum number of pages we mmap for 1 file.
     # this should be not big not to exceed mlock limit.
-    _max_tracked = 10
+    _max_tracked = 8
 
     def __init__(t, tdb, zf, at=None):
         assert isinstance(zf, ZBigFile)
@@ -359,7 +364,7 @@ class tWatch:
 
     def __init__(t, tdb):
         t.tdb = tdb
-        t.w   = tdb._open("/head/watch", mode='rwb')
+        t.w   = tdb._open("head/watch", mode='rwb')
 
         t._acceptq  = chan() # (stream, msg)    server originated messages go here
         t._rxmu     = threading.Lock()
@@ -374,11 +379,16 @@ class tWatch:
         tdb._tracked.add(t)
 
     def close(t):
-        # XXX close * from rxtab
         t.tdb._tracked.remove(t)
         t.w.close()
         t._serveDone.recv()
-
+        # wakeup everyone waiting for rx
+        t._acceptq.close()
+        with t._rxmu:
+            rxtab = t._rxtab
+            t._rxtab = None     # don't allow new rxtab registers
+        for rxq in rxtab.values():
+            rxq.close()
 
     # _serveRecv receives messages from .w and dispatches them according to streamID.
     @func
@@ -406,14 +416,13 @@ class tWatch:
                     t._accepted.add(stream)
                 t._acceptq.send((stream, msg))
 
-
     # _send sends raw message via specified stream.
     #
     # multiple _send can be called in parallel - _send serializes writes.
     def _send(t, stream, msg):
         assert '\n' not in msg
         with t._txmu:
-            t.w.write("%d %s\n" % (stream, msg))
+            t.w.write(b"%d %s\n" % (stream, msg))
 
     # sendReq sends client -> server request and returns server reply.
     #
@@ -440,6 +449,29 @@ class tWatch:
         stream, msg = rx
         return tSrvReq(t, stream, msg)
 
+
+    # expectPin ... XXX
+    #
+    # expectv is [] of (zf, blk, at)
+    # returns [] of received pin requests.
+    def expectPin(t, expectv):
+        expected = set()    # of expected pin messages
+        for zf, blk, at in expectv:
+            msg = b"pin %s #%d @%s" % (h(zf._p_oid), blk, h(at))
+            assert msg not in expected
+            expected.add(msg)
+
+        reqv = []   # of received requests
+        while len(expected) > 0:
+            req = t.recvReq()
+            assert req.msg in expected
+            expected.delete(req.msg)
+            reqv.append(req)
+
+        return reqv
+
+
+
 # tSrvReq represents 1 server-initiated wcfs request received over /head/watch.
 class tSrvReq:
     def __init__(req, twatch, stream, msg):
@@ -547,15 +579,17 @@ def test_wcfs():
 
     # >>> XXX commit data to not yet accessed f part - nothing happens
 
-    w = tdb.openwatch()
+    # XXX invalidation protocol ...
+
+    w = t.openwatch()
 
-    # XXX vvv -> expectPin?
     done = chan()
+    @func
     def _():
-        r = w.recvReq(); assert r == ("pin %s #%d @%s" % zf, ...)
-        r.ack()
-        ...
-        done.close()
+        defer(done.close)
+        pinv = w.expectPin([(zf, blk, at), (zf, blk, at)])
+        for p in pinv:
+            p.ack()
     go(_)
     assert w.sentReq("watch %s @%s", zf, at1) == "ok"
     done.recv()