.

wcfs/wcfs_test.py

@@ -479,30 +479,32 @@ class tFile:
     def assertCache(t, incorev):
         assert t.cached() == incorev
 
-    # assertBlk asserts that file block #blk has data as expected.
+    # assertBlk asserts that file[blk] has data as expected.
     #
     # Expected data may be given with size < t.blksize. In such case the data
     # is implicitly appended with trailing zeros. Data can be both bytes and unicode.
     #
     # pinokByWLink: {} tWatchLink -> (zf, {} blk -> at).
     # pinokByWLink can be None - in that case it is computed automatically.
-    def assertBlk(t, blk, data, pinokByWLink=None):
-        if not isinstance(data, bytes):
-            data = data.encode('utf-8')
-        assert len(data) <= t.blksize
-        dataOk, blkrev = t.tdb._blkData(t.zf, blk, t.at)
-        assert dataOk == data, "computed vs explicit data"
-        data += b'\0'*(t.blksize - len(data))   # tailing zeros
+    def assertBlk(t, blk, dataok, pinokByWLink=None):
+        if not isinstance(dataok, bytes):
+            dataok = dataok.encode('utf-8')
+        assert len(dataok) <= t.blksize
+        blkdata, blkrev = t.tdb._blkData(t.zf, blk, t.at)
+        assert blkdata == dataok, "computed vs explicit data"
+        dataok += b'\0'*(t.blksize - len(dataok))   # tailing zeros
         assert blk < t._sizeinblk()
 
-        # if access goes to @head/file watches must be notified. if to @rev/file - silent.
+        cached = f.cached()[blk]
+
+        # watches must be notified if access goes to @head/file; not if to @rev/file.   XXX text
         wpin = {}   # tWatchLink -> (zf, pinok)
         for wlink in t.tdb._wlinks:
             pinok = {}
             if t.at is None: # @head/...
                 wat = wlink._watching.get(zf)
-                if wat is not None and wat < blkrev:
-                    # XXX and block not cached and watch not already pinned on the watch
+                if wat is not None and wat < blkrev and cached == 0:
+                    # XXX and watch not already pinned on the watch
                     pinok = {blk: t.tdb._blkRev(t.zf, blk, wat)}
             wpin[wlink] = (t.zf, pinok)
 
@@ -510,31 +512,67 @@ class tFile:
             assert wpin == pinokByWLink, "computed vs explicit pinokByWLink"
         pinokByWLink = wpin
 
-
+        blkview = f.blk(blk)
+        assert f.cached()[blk] == cached
+
+        def _(ctx, ev):
+            assert f.cached()[blk] == cached
+            ev.append('read pre')
+
+            # access data with released GIL so that the thread that reads data from
+            # head/watch can receive pin message. Be careful to handle cancelation,
+            # so that on error in another worker we don't stuck and the error can
+            # be propagated to wait and reported.
+            have_read = chan(1)
+            def _():
+                b = read0_nogil(blkview)
+                have_read.send(b)
+            go(_)
+            _, _rx = select(
+                ctx.done().recv,    # 0
+                have_read.recv,     # 1
+            )
+            if _ == 0:
+                raise ctx.err()
+            b = _rx
+
+            ev.append('read ' + chr(b))
+        ev = doCheckingPin(_, pinokByWLink)
+
+        # XXX hack - wlinks are notified and emit events simultaneously - we
+        # check only that events begin and end with read-related and that pins
+        # are inside. Better do explicit check in tracetest style.
+        assert ev[0]  == 'read pre'
+        assert ev[-1] == 'read ' + dataok[0]
+        ev = ev[1:-1]
+        # XXX ev == 'pin rx', 'pin ack pre' ....
+
+        assert f.cached()[blk] > 0
 
         # XXX assert individually for every block's page? (easier debugging?)
-        assert t.blk(blk) == data, ("#blk: %d" % blk)
+        assert blkview == dataok, ("#blk: %d" % blk)
 
         # we just accessed the block - it has to be in OS cache
         assert t.cached()[blk] == 1
 
+
     # assertData asserts that file has data blocks as specified.
     #
     # Expected blocks may be given with size < zf.blksize. In such case they
     # are implicitly appended with trailing zeros.
     #
     # It also checks file size and optionally mtime.
-    def assertData(t, datav, mtime=None):
+    def assertData(t, dataokv, mtime=None):
         st = os.fstat(t.f.fileno())
-        assert st.st_size == len(datav)*t.blksize
+        assert st.st_size == len(dataokv)*t.blksize
         if mtime is not None:
             assert st.st_mtime == tidtime(mtime)
 
-        for blk, data in enumerate(datav):
-            t.assertBlk(blk, data)
+        for blk, dataok in enumerate(dataokv):
+            t.assertBlk(blk, dataok)
 
         # all blocks must be in cache after we touched them all
-        t.assertCache([1]*len(datav))
+        t.assertCache([1]*len(dataokv))
 
 
 # tWatchLink provides testing environment for /head/watch link opened on wcfs.
@@ -1094,14 +1132,15 @@ def test_wcfs():
     at4 = t.commit()
     f.assertCache([1,1,0,1,0,0])                # FIXME a must be invalidated - see δbtree ^^^
 
-    # XXX -> f.assertBlk(2, '4c', {wl: {2: at3}})
-    blk = 2
-    pinok = {2: at3}     # XXX at3 -> at <= wl.at for zf
-
+    f.assertBlk(2, '4c', {wl: {2: at3}})
     # XXX 5, {5: ø}
     # XXX 0, {0, at3} after δbtree works
 
-    blk_data = f.blk(blk)
+"""
+    blk = 2
+    pinok = {2: at3}     # XXX at3 -> at <= wl.at for zf
+
+    blkview = f.blk(blk)
     assert f.cached()[blk] == 0
 
     def _(ctx, ev):
@@ -1114,7 +1153,7 @@ def test_wcfs():
         # successfully propagated to wait and reported.
         have_read = chan(1)
         def _():
-            b = read0_nogil(blk_data)
+            b = read0_nogil(blkview)
             have_read.send(b)
         go(_)
         _, _rx = select(
@@ -1134,6 +1173,7 @@ def test_wcfs():
     assert f.cached()[blk] > 0
     f.assertBlk(blk, '4c')
     assert f.cached()[blk] == 1
+"""
 
     wl.close()