Merge lp:~allenap/postgresfixture/more-locking into lp:~lazr-developers/postgresfixture/trunk

[Gavin Panella]

Lock, using lockf, around critical sections during cluster set-up and tear-down.

[Raphaël Badin (rvb)]

Looks good.

rvba -> allenap: could you tell me why you did s/locks/shares/ in the postgresfixtur branch?
allenap -> rvba: That change, from lock to shares, is because of the addition of the "critical section" lock (the lockf based one). The shares lock is better named now because it reflects that each acquired lock on it represents a share in the resource.

review: Approve

=== modified file 'postgresfixture/cluster.py'
--- postgresfixture/cluster.py	2012-05-14 13:37:38 +0000
+++ postgresfixture/cluster.py	2012-05-22 22:41:20 +0000
@@ -14,7 +14,16 @@
     "Cluster",
     ]
 
-from contextlib import closing
+from contextlib import (
+    closing,
+    contextmanager,
+    )
+from fcntl import (
+    LOCK_EX,
+    LOCK_UN,
+    lockf,
+    )
+from functools import wraps
 from os import (
     devnull,
     environ,
@@ -46,12 +55,40 @@
     return path.pathsep.join(exe_path)
 
 
+def locked(method):
+    """Execute the decorated method with its instance's lock held."""
+    @wraps(method)
+    def wrapper(self, *args, **kwargs):
+        with self.lock:
+            return method(self, *args, **kwargs)
+    return wrapper
+
+
 class Cluster:
     """Represents a PostgreSQL cluster, running or not."""
 
     def __init__(self, datadir):
         self.datadir = path.abspath(datadir)
-
+        self.lockfile = path.join(
+            path.dirname(self.datadir),
+            ".%s.lock" % path.basename(self.datadir))
+
+    @property
+    @contextmanager
+    def lock(self):
+        """Context that takes out a lock for critical sections.
+
+        The lock is meant to be short-lived, to avoid race conditions. As
+        such, acquiring this lock will block.
+        """
+        with open(self.lockfile, "ab") as fd:
+            lockf(fd, LOCK_EX)
+            try:
+                yield
+            finally:
+                lockf(fd, LOCK_UN)
+
+    @locked
     def execute(self, *command, **options):
         """Execute a command with an environment suitable for this cluster."""
         env = options.pop("env", environ).copy()
@@ -92,6 +129,7 @@
             else:
                 return True
 
+    @locked
     def create(self):
         """Create this cluster, if it does not exist."""
         if not self.exists:
@@ -99,6 +137,7 @@
                 makedirs(self.datadir)
             self.execute("pg_ctl", "init", "-s", "-o", "-E utf8 -A trust")
 
+    @locked
     def start(self):
         """Start this cluster, if it's not already started."""
         if not self.running:
@@ -149,6 +188,7 @@
             with closing(conn.cursor()) as cur:
                 cur.execute("DROP DATABASE %s" % name)
 
+    @locked
     def stop(self):
         """Stop this cluster, if started."""
         if self.running:
@@ -157,6 +197,7 @@
             #  -m <mode> -- shutdown mode.
             self.execute("pg_ctl", "stop", "-s", "-w", "-m", "fast")
 
+    @locked
     def destroy(self):
         """Destroy this cluster, if it exists.
 

=== modified file 'postgresfixture/clusterfixture.py'
--- postgresfixture/clusterfixture.py	2012-05-14 13:37:38 +0000
+++ postgresfixture/clusterfixture.py	2012-05-22 22:41:20 +0000
@@ -99,8 +99,8 @@
         """
         super(ClusterFixture, self).__init__(datadir)
         self.preserve = preserve
-        self.lock = ProcessSemaphore(
-            path.join(self.datadir, "locks"))
+        self.shares = ProcessSemaphore(
+            path.join(self.datadir, "shares"))
 
     def setUp(self):
         super(ClusterFixture, self).setUp()
@@ -112,8 +112,8 @@
             self.create()
         self.addCleanup(self.stop)
         self.start()
-        self.addCleanup(self.lock.release)
-        self.lock.acquire()
+        self.addCleanup(self.shares.release)
+        self.shares.acquire()
 
     def createdb(self, name):
         """Create the named database if it does not exist already.
@@ -133,10 +133,10 @@
 
     def stop(self):
         """Stop the cluster, but only if there are no other users."""
-        if not self.lock.locked:
+        if not self.shares.locked:
             super(ClusterFixture, self).stop()
 
     def destroy(self):
         """Destroy the cluster, but only if there are no other users."""
-        if not self.lock.locked:
+        if not self.shares.locked:
             super(ClusterFixture, self).destroy()

=== modified file 'postgresfixture/main.py'
--- postgresfixture/main.py	2012-05-22 14:53:09 +0000
+++ postgresfixture/main.py	2012-05-22 22:41:20 +0000
@@ -67,9 +67,9 @@
     action_stop(cluster, arguments)
     cluster.destroy()
     if cluster.exists:
-        if cluster.lock.locked:
+        if cluster.shares.locked:
             message = "%s: cluster is %s" % (
-                cluster.datadir, locked_by_description(cluster.lock))
+                cluster.datadir, locked_by_description(cluster.shares))
         else:
             message = "%s: cluster could not be removed." % cluster.datadir
         error(message)
@@ -121,9 +121,9 @@
     """Stop the cluster."""
     cluster.stop()
     if cluster.running:
-        if cluster.lock.locked:
+        if cluster.shares.locked:
             message = "%s: cluster is %s" % (
-                cluster.datadir, locked_by_description(cluster.lock))
+                cluster.datadir, locked_by_description(cluster.shares))
         else:
             message = "%s: cluster is still running." % cluster.datadir
         error(message)

=== modified file 'postgresfixture/tests/test_cluster.py'
--- postgresfixture/tests/test_cluster.py	2012-05-14 15:01:01 +0000
+++ postgresfixture/tests/test_cluster.py	2012-05-22 22:41:20 +0000
@@ -17,7 +17,13 @@
     getpid,
     path,
     )
-from subprocess import CalledProcessError
+from subprocess import (
+    CalledProcessError,
+    PIPE,
+    Popen,
+    )
+import sys
+from textwrap import dedent
 
 import postgresfixture.cluster
 from postgresfixture.cluster import (
@@ -27,6 +33,7 @@
     )
 from postgresfixture.main import repr_pid
 from postgresfixture.testing import TestCase
+from testtools.content import text_content
 from testtools.matchers import (
     DirExists,
     FileExists,
@@ -62,9 +69,46 @@
     def test_init(self):
         # The datadir passed into the Cluster constructor is resolved to an
         # absolute path.
-        datadir = path.join(self.make_dir(), "locks")
+        tmpdir = self.make_dir()
+        datadir = path.join(tmpdir, "somewhere")
         cluster = self.make(path.relpath(datadir))
         self.assertEqual(datadir, cluster.datadir)
+        # The lock file is in the parent directory of the data directory.
+        self.assertEqual(
+            path.join(tmpdir, ".somewhere.lock"),
+            cluster.lockfile)
+
+    def test_lock(self):
+        # To test the lock - based on lockf - we take the lock locally then
+        # check if it appears locked from a separate process.
+        cluster = self.make(self.make_dir())
+        script = dedent("""\
+            from errno import EAGAIN
+            from fcntl import LOCK_EX, LOCK_NB, lockf
+            with open(%r, "ab") as fd:
+                try:
+                    lockf(fd, LOCK_EX | LOCK_NB)
+                except IOError, error:
+                    if error.errno != EAGAIN:
+                        raise
+                else:
+                    raise AssertionError("Not locked")
+            """) % cluster.lockfile
+        with cluster.lock:
+            process = Popen(
+                sys.executable, stdin=PIPE,
+                stdout=PIPE, stderr=PIPE)
+            stdout, stderr = process.communicate(script)
+            self.addDetail("stdout", text_content(stdout))
+            self.addDetail("stderr", text_content(stderr))
+            self.assertEqual(0, process.returncode)
+
+    def test_lock_is_reentrant(self):
+        # The lock can be acquired several times.
+        cluster = self.make(self.make_dir())
+        with cluster.lock:
+            with cluster.lock:
+                pass  # We get here okay.
 
     def patch_check_call(self, returncode=0):
         calls = []
@@ -112,19 +156,19 @@
 
     def test_running(self):
         calls = self.patch_check_call(returncode=0)
-        cluster = self.make("/some/where")
+        cluster = self.make(self.make_dir())
         self.assertTrue(cluster.running)
         [(command, options)] = calls
         self.assertEqual(("pg_ctl", "status"), command)
 
     def test_running_not(self):
         self.patch_check_call(returncode=1)
-        cluster = self.make("/some/where")
+        cluster = self.make(self.make_dir())
         self.assertFalse(cluster.running)
 
     def test_running_error(self):
         self.patch_check_call(returncode=2)
-        cluster = self.make("/some/where")
+        cluster = self.make(self.make_dir())
         self.assertRaises(
             CalledProcessError, getattr, cluster, "running")
 

=== modified file 'postgresfixture/tests/test_clusterfixture.py'
--- postgresfixture/tests/test_clusterfixture.py	2012-05-21 10:34:58 +0000
+++ postgresfixture/tests/test_clusterfixture.py	2012-05-22 22:41:20 +0000
@@ -68,10 +68,10 @@
     def test_init_fixture(self):
         fixture = self.make("/some/where")
         self.assertEqual(False, fixture.preserve)
-        self.assertIsInstance(fixture.lock, ProcessSemaphore)
+        self.assertIsInstance(fixture.shares, ProcessSemaphore)
         self.assertEqual(
-            path.join(fixture.datadir, "locks"),
-            fixture.lock.lockdir)
+            path.join(fixture.datadir, "shares"),
+            fixture.shares.lockdir)
 
     def test_createdb_no_preserve(self):
         fixture = self.make(self.make_dir(), preserve=False)
@@ -108,26 +108,26 @@
         fixture.dropdb("diekrupps")
         # The test is that we arrive here without error.
 
-    def test_stop_locked(self):
-        # The cluster is not stopped if a lock is held.
+    def test_stop_share_locked(self):
+        # The cluster is not stopped if a shared lock is held.
         fixture = self.make(self.make_dir())
         self.addCleanup(fixture.stop)
         fixture.start()
-        fixture.lock.acquire()
+        fixture.shares.acquire()
         fixture.stop()
         self.assertTrue(fixture.running)
-        fixture.lock.release()
+        fixture.shares.release()
         fixture.stop()
         self.assertFalse(fixture.running)
 
-    def test_destroy_locked(self):
-        # The cluster is not destroyed if a lock is held.
+    def test_destroy_share_locked(self):
+        # The cluster is not destroyed if a shared lock is held.
         fixture = self.make(self.make_dir())
         fixture.create()
-        fixture.lock.acquire()
+        fixture.shares.acquire()
         fixture.destroy()
         self.assertTrue(fixture.exists)
-        fixture.lock.release()
+        fixture.shares.release()
         fixture.destroy()
         self.assertFalse(fixture.exists)
 

=== modified file 'postgresfixture/tests/test_main.py'
--- postgresfixture/tests/test_main.py	2012-05-22 14:53:09 +0000
+++ postgresfixture/tests/test_main.py	2012-05-22 22:41:20 +0000
@@ -137,12 +137,12 @@
         self.assertFalse(cluster.running)
         self.assertTrue(cluster.exists)
 
-    def test_stop_when_locked(self):
+    def test_stop_when_share_locked(self):
         cluster = ClusterFixture(self.make_dir())
         self.addCleanup(cluster.stop)
         cluster.start()
-        self.addCleanup(cluster.lock.release)
-        cluster.lock.acquire()
+        self.addCleanup(cluster.shares.release)
+        cluster.shares.acquire()
         self.patch(sys, "stderr", StringIO())
         error = self.assertRaises(
             SystemExit, main.action_stop, cluster,
@@ -161,10 +161,10 @@
         self.assertFalse(cluster.running)
         self.assertFalse(cluster.exists)
 
-    def test_destroy_when_locked(self):
+    def test_destroy_when_share_locked(self):
         cluster = ClusterFixture(self.make_dir())
         cluster.create()
-        cluster.lock.acquire()
+        cluster.shares.acquire()
         self.patch(sys, "stderr", StringIO())
         error = self.assertRaises(
             SystemExit, main.action_destroy, cluster,