.

wcfs/__init__.py

@@ -87,9 +87,13 @@ class Conn(object):
     # ._wc          WCFS
     # .at           Tid
     # ._wlink       WatchLink watch/receive pins for created mappings
-
+    #
     # ._filemu      threading.Lock
     # ._filetab     {} foid -> _File
+    #
+    # ._pinWG
+    # ._pinCancel
+
     pass
 
 # _File represent isolated file view under Conn.
@@ -127,7 +131,9 @@ def connect(wc, at): # -> Conn
     wconn._filemu   = threading.Lock()
     wconn._filetab  = {}
 
-    # XXX wg.go(wconn._pinner, xxxctx)
+    pinCtx, wconn._pinCancel = context.with_cancel(context.background())
+    wconn._pinWG = sync.WorkGroup(pinCtx)
+    wconn._pinWG.go(wconn._pinner)
 
     return wconn
 
@@ -135,8 +141,9 @@ def connect(wc, at): # -> Conn
 # XXX what happens to file mmappings?
 @func(Conn)
 def close(wconn):
-    # XXX stop/join pinner
     wconn._wlink.close()
+    wconn._pinCancel()
+    wconn._pinWG.wait()     # XXX canceled - ok
 
     # close all files - both that have no mappings and that still have opened mappings.
     # XXX after file is closed mappings continue to survive, but we can no
@@ -157,8 +164,23 @@ def close(wconn):
 # python code (thus with GIL taken).
 @func(Conn)
 def _pinner(wconn, ctx):
+    # if pinner fails, wcfs will kill us.
+    # log pinner exception so the error is not hidden.
+    def _():
+        exc = sys.exc_info()[1]
+        if exc is None:
+            return
+        log.critical('pinner failed:', exc_info=1)
+
+    defer(_)
+
+    def trace(msg):
+        print('pinner: %s' % msg)
+    trace('start')
     while 1:
+        trace('recvReq ...')
         req = wconn._wlink.recvReq(ctx)
+        trace('-> %r' % req)
         if req is None:
             return  # XXX ok? (EOF - when wcfs closes wlink)
 

wcfs/wcfs_test.py

@@ -1712,6 +1712,7 @@ class tMapping(object):
         dataok += b'\0'*(f.blksize - len(dataok))   # trailing zeros
 
         blkview = memoryview(t.mmap.mem[blk_inmmap*f.blksize:][:f.blksize])
+        print('aaa')
 
         # XXX first access without GIL, so that e.g. if there is timeout on
         # wcfs.py side, _abort_ontimeout could run and kill WCFS.
@@ -1720,6 +1721,7 @@ class tMapping(object):
         _ = read_nogil(blkview[:1])
         assert _ == dataok[0]
         assert blkview.tobytes() == dataok
+        print('bbb')
 
     # XXX assertData
 
@@ -1748,7 +1750,9 @@ def test_wcfspy_virtmem():
     #assertCache(m1, [0,0,0])
     assert f.pinned == {}
 
+    print('000')
     tm1.assertBlk(2, 'c1')
+    print('\n\n111')
     assert f.pinned == {2:at1}
     tm1.assertBlk(3, 'd1')
     assert f.pinned == {2:at1}