.

wcfs/__init__.py

@@ -49,23 +49,20 @@ from zodbtools.util import ashex
 
 
 # WCFS represents connection to wcfs server.
+#
+# It has to be created with join.
 # Only 1 connection is maintained for one file server.
 class WCFS(object):
     # .mountpoint   path to wcfs mountpoint
     # ._fwcfs       /.wcfs/zurl opened to keep the server from going away (at least cleanly)
+    # ._njoin       this connection was returned for so many joins
 
     # XXX for-testing only?
     # ._proc        wcfs process if it was opened by this WCFS | None
+    pass
 
-    def __init__(self, mountpoint, fwcfs, proc):
-        self.mountpoint = mountpoint
-        self._fwcfs     = fwcfs
-        self._proc      = proc
-
-    def close(self):
-        # XXX unmount wcfs as well?
-        self._fwcfs.close()
 
+"""
     # open creates wcfs file handle, which can be mmaped to give data of ZBigFile.
     #
     # XXX more text
@@ -106,6 +103,7 @@ class FileH(object):
 
     def mmap(self, pgoffset, pglen):
         return _VMA(self.fd, pgoffset, pglen, pagesize, mmap.PROT_READ)
+"""
 
 
 
@@ -113,39 +111,24 @@ class FileH(object):
 
 # ---- join/run wcfs ----
 
-# serve starts and runs wcfs server for ZODB @ zurl.
-#
-# it mounts wcfs at a location that is with 1-1 correspondence with zurl.
-# it then waits for wcfs to exit (either due to unmount or an error).
-#
-# it is an error if wcfs was already started.
-#
-# XXX optv
-# if exec_ is True, wcfs is not spawned, but executed into.
-#
-# serve(zurl, exec_=False).
-def serve(zurl, optv, exec_=False):
-    mntpt = _mntpt_4zurl(zurl)
+_wcmu = threading.Lock()
+_wcregistry = {} # mntpt -> WCFS
 
-    # try opening .wcfs - it is an error if we can do it.
-    # XXX -> option to wcfs itself?
-    try:
-        f = open(mntpt + "/.wcfs/zurl")
-    except IOError as e:
-        if e.errno != ENOENT:
-            raise
-    else:
-        f.close()
-        raise RuntimeError("wcfs: start %s: already started" % zurl)
-
-    # seems to be ok to start
-    # XXX race window if something starts after ^^^ check
-    argv = [_wcfs_exe()] + list(optv) + [zurl, mntpt]
-    if not exec_:
-        subprocess.check_call(argv, close_fds=True)
-    else:
-        os.execv(argv[0], argv)
+@func(WCFS)
+def __init__(wc, mountpoint, fwcfs, proc):
+    wc.mountpoint = mountpoint
+    wc._fwcfs     = fwcfs
+    wc._njoin     = 1
+    wc._proc      = proc
 
+@func(WCFS)
+def close(wc):
+    with _wcmu:
+        wc._njoin -= 1
+        if wc._njoin == 0:
+            # XXX unmount wcfs as well?
+            wc._fwcfs.close()
+            del _wcregistry[wc.mountpoint]
 
 # _default_autostart returns default autostart setting for join.
 #
@@ -162,16 +145,16 @@ def _default_autostart():
 #
 # If wcfs for that zurl was already started, join connects to it.
 # Otherwise it starts wcfs for zurl if autostart is True.
-#
-# If shared is True - a shared connection is returned - one that will be also   XXX -> always this way
-# returned for following join(shared=True) requests with the same zurl.
-def join(zurl, autostart=_default_autostart(), shared=False):   # -> Conn
-    # XXX implement shared
-
+def join(zurl, autostart=_default_autostart()):   # -> WCFS
     mntpt = _mntpt_4zurl(zurl)
+    with _wcmu:
+        # check if we already have connection to wcfs server from this process
+        wc = _wcregistry.get(mntpt)
+        if wc is not None:
+            wc._njoin += 1
+            return wc
 
-    # try opening .wcfs - if we succeed - it is already mounted.
-    # XXX -> wcfs itself?
+        # no. try opening .wcfs - if we succeed - wcfs is already mounted.
         try:
             f = open(mntpt + "/.wcfs/zurl")
         except IOError as e:
@@ -179,7 +162,9 @@ def join(zurl, autostart=_default_autostart(), shared=False):   # -> Conn
                 raise
         else:
             # already have it
-        return WCFS(mntpt, f, None)
+            wc = WCFS(mntpt, f, None)
+            _wcregistry[mntpt] = wc
+            return wc
 
         if not autostart:
             raise RuntimeError("wcfs: join %s: server not started" % zurl)
@@ -192,6 +177,7 @@ def join(zurl, autostart=_default_autostart(), shared=False):   # -> Conn
 # _start starts wcfs server for ZODB @ zurl.
 #
 # optv can be optionally given to pass flags to wcfs.
+# called under _wcmu
 def _start(zurl, *optv):    # -> WCFS
     mntpt = _mntpt_4zurl(zurl)
     log.info("wcfs: starting for %s ...", zurl)
@@ -255,9 +241,14 @@ def _start(zurl, *optv):    # -> WCFS
     wg.go(_)
 
     wg.wait()
+
+    assert mntpt not in _wcregistry
+    _wcregistry[mntpt] = wc
     return wc
 
 
+# ---- misc ----
+
 # _wcfs_exe returns path to wcfs executable.
 def _wcfs_exe():
     return '%s/wcfs' % dirname(__file__)
@@ -301,6 +292,40 @@ def _mkdir_p(path):
             raise
 
 
+# serve starts and runs wcfs server for ZODB @ zurl.
+#
+# it mounts wcfs at a location that is with 1-1 correspondence with zurl.
+# it then waits for wcfs to exit (either due to unmount or an error).
+#
+# it is an error if wcfs was already started.
+#
+# XXX optv
+# if exec_ is True, wcfs is not spawned, but executed into.
+#
+# serve(zurl, exec_=False).
+def serve(zurl, optv, exec_=False):
+    mntpt = _mntpt_4zurl(zurl)
+
+    # try opening .wcfs - it is an error if we can do it.
+    # XXX -> option to wcfs itself?
+    try:
+        f = open(mntpt + "/.wcfs/zurl")
+    except IOError as e:
+        if e.errno != ENOENT:
+            raise
+    else:
+        f.close()
+        raise RuntimeError("wcfs: start %s: already started" % zurl)
+
+    # seems to be ok to start
+    # XXX race window if something starts after ^^^ check
+    argv = [_wcfs_exe()] + list(optv) + [zurl, mntpt]
+    if not exec_:
+        subprocess.check_call(argv, close_fds=True)
+    else:
+        os.execv(argv[0], argv)
+
+
 
 
 # if called as main just -> serve()

wcfs/wcfs_test.py

@@ -114,16 +114,23 @@ def test_join():
     with raises(RuntimeError, match="wcfs: join .*: server not started"):
         wcfs.join(zurl, autostart=False)
 
+    assert wcfs._wcregistry == {}
+    def _():
+        assert wcfs._wcregistry == {}
+    defer(_)
+
     wc = wcfs._start(zurl)
     defer(wc.close)
     assert wc.mountpoint == testmntpt
+    assert wc._njoin == 1
     assert readfile(wc.mountpoint + "/.wcfs/zurl") == zurl
     assert os.path.isdir(wc.mountpoint + "/head")
     assert os.path.isdir(wc.mountpoint + "/head/bigfile")
 
     wc2 = wcfs.join(zurl, autostart=False)
     defer(wc2.close)
-    assert wc2.mountpoint == wc.mountpoint
+    assert wc2 is wc
+    assert wc._njoin == 2
 
 # test that join(autostart=y) works.
 @func
@@ -132,9 +139,15 @@ def test_join_autostart():
     with raises(RuntimeError, match="wcfs: join .*: server not started"):
         wcfs.join(zurl, autostart=False)
 
+    assert wcfs._wcregistry == {}
+    def _():
+        assert wcfs._wcregistry == {}
+    defer(_)
+
     wc = wcfs.join(zurl, autostart=True)
     defer(wc.close)
     assert wc.mountpoint == testmntpt
+    assert wc._njoin == 1
     assert readfile(wc.mountpoint + "/.wcfs/zurl") == zurl
     assert os.path.isdir(wc.mountpoint + "/head")
     assert os.path.isdir(wc.mountpoint + "/head/bigfile")