Improve the portability of .alock.

 * In one version COALESCE was trying to cast int4[] to int8[].
   Avoid this by COALESCE'ing the results of array_upper().
 * Define the alock primitives as methods. The try_* interfaces came later,
   so referencing all the statements at once yielded a portability problem
   despite the server's capability of acquiring waiting locks.

Add a test actually exercising concurrency. It's not particularly good as it
would be successful if ALock() did nothing at all, but, nonetheless, exercise
the code path.

Author: James William Pye

postgresql/alock.py

@@ -7,6 +7,12 @@
 from abc import abstractmethod, abstractproperty
 from .python.element import Element
 
+__all__ = [
+	'ALock',
+	'ExclusiveLock',
+	'ShareLock',
+]
+
 class ALock(Element):
 	"""
 	Advisory Lock class for managing the acquisition and release of a sequence
@@ -33,10 +39,21 @@ def mode(self):
 		"""
 
 	@abstractmethod
-	def __select_statements__(self):
+	def _try(self, id_pairs, ids):
 		"""
-		ALock subclasses need to implement this in order to specify
-		the actual statements that try, acquire, and release the locks.
+		Try and acquire.
+		"""
+
+	@abstractmethod
+	def _acquire(self, id_pairs, ids):
+		"""
+		Acquire and wait if necessary.
+		"""
+
+	@abstractmethod
+	def _release(self, id_pairs, ids):
+		"""
+		Release the locks.
 		"""
 
 	@staticmethod
@@ -60,12 +77,12 @@ def acquire(self, blocking = True, len = len):
 			# _count is zero, so the locks need to be acquired.
 			wait = bool(blocking)
 			if wait:
-				self._acquire_stmt(self._id_pairs, self._ids)
+				self._acquire(self._id_pairs, self._ids)
 			else:
 				# grab the success of each lock id. if some were
 				# unsuccessful, then the ones that were successful need to be
 				# released.
-				r = self._try_stmt(self._id_pairs, self._ids)
+				r = self._try(self._id_pairs, self._ids)
 				# accumulate the identifiers that *did* lock
 				release_seq = [
 					id for didlock, id in zip(r, self.identifiers) if didlock[0]
@@ -78,7 +95,7 @@ def acquire(self, blocking = True, len = len):
 					# on the same seq that it should be able to acquire all of
 					# them once the contended lock is released.
 					release_seq.reverse()
-					self._release_stmt(*self._split_lock_identifiers(release_seq))
+					self._release(*self._split_lock_identifiers(release_seq))
 					# unable to acquire all.
 					return False
 		self._count = self._count + 1
@@ -97,7 +114,7 @@ def release(self):
 		if not self.database.closed and self._count > 0:
 			# if the database has been closed, or the count will
 			# remain non-zero, there is no need to release.
-			self._release_stmt(reversed(self._id_pairs), reversed(self._ids))
+			self._release(reversed(self._id_pairs), reversed(self._ids))
 			# decrement the count nonetheless.
 		self._count = self._count - 1
 
@@ -124,25 +141,25 @@ def __init__(self, database, *identifiers):
 		self.connection = self.database = database
 		self.identifiers = identifiers
 		self._id_pairs, self._ids = self._split_lock_identifiers(identifiers)
-		self._try_stmt, self._acquire_stmt, self._release_stmt = \
-			self.__select_statements__()
 
 class ShareLock(ALock):
 	mode = 'share'
-	def __select_statements__(self):
-		sys = self.database.sys
-		return (
-			sys.try_advisory_shared,
-			sys.acquire_advisory_shared,
-			sys.release_advisory_shared
-		)
+	def _try(self, *args):
+		return self.database.sys.try_advisory_shared(*args)
+
+	def _acquire(self, *args):
+		return self.database.sys.acquire_advisory_shared(*args)
+
+	def _release(self, *args):
+		return self.database.sys.release_advisory_shared(*args)
 
 class ExclusiveLock(ALock):
 	mode = 'exclusive'
-	def __select_statements__(self):
-		sys = self.database.sys
-		return (
-			sys.try_advisory_exclusive,
-			sys.acquire_advisory_exclusive,
-			sys.release_advisory_exclusive
-		)
+	def _try(self, *args):
+		return self.database.sys.try_advisory_exclusive(*args)
+
+	def _acquire(self, *args):
+		return self.database.sys.acquire_advisory_exclusive(*args)
+
+	def _release(self, *args):
+		return self.database.sys.release_advisory_exclusive(*args)

postgresql/lib/libsys.sql

@@ -208,7 +208,7 @@ SELECT
 		pg_catalog.pg_advisory_unlock_shared($2[i])
 	END AS released
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)
 
 [acquire_advisory_shared]
 SELECT COUNT((
@@ -220,7 +220,7 @@ SELECT COUNT((
 	END
 ) IS NULL) AS acquired
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)
 
 [try_advisory_shared]
 SELECT
@@ -231,7 +231,7 @@ SELECT
 		pg_catalog.pg_try_advisory_lock_shared($2[i])
 	END AS acquired
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)
 
 [release_advisory_exclusive]
 SELECT
@@ -242,7 +242,7 @@ SELECT
 		pg_catalog.pg_advisory_unlock($2[i])
 	END AS released
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)
 
 [acquire_advisory_exclusive]
 SELECT COUNT((
@@ -254,7 +254,7 @@ SELECT COUNT((
 	END
 ) IS NULL) AS acquired -- Guaranteed to be acquired once complete.
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)
 
 [try_advisory_exclusive]
 SELECT
@@ -265,4 +265,4 @@ SELECT
 		pg_catalog.pg_try_advisory_lock($2[i])
 	END AS acquired
 FROM
-	pg_catalog.generate_series(1, array_upper(COALESCE($1::int4[],$2::int8[]), 1)) AS g(i)
+	pg_catalog.generate_series(1, COALESCE(array_upper($2::int8[], 1), array_upper($1::int4[], 1))) AS g(i)

postgresql/temporal.py

@@ -61,7 +61,10 @@ def destroy(self):
 		if os.getpid() == self._init_pid_:
 			with self:
 				# Kill all the open connections.
-				db.sys.terminate_backends()
+				try:
+					db.sys.terminate_backends()
+				except Exception:
+					pass
 			cluster = self.cluster
 			self.cluster = None
 			self._init_pid_ = None

postgresql/test/test_alock.py

@@ -7,13 +7,44 @@
 from ..temporal import pg_tmp
 from .. import alock
 
+n_alocks = "select count(*) FROM pg_locks WHERE locktype = 'advisory'"
+
 class test_alock(unittest.TestCase):
+	@pg_tmp
+	def testALockWait(self):
+		# sadly, this is primarily used to exercise the code paths..
+		ad = prepare(n_alocks).first
+		self.failUnlessEqual(ad(), 0)
+		state = [False, False, False]
+		alt = new()
+		def concurrent_lock():
+			pass
+			with alock.ExclusiveLock(alt, (1,1)):
+				with alock.ExclusiveLock(alt, (0,0)):
+					# start it
+					state[0] = True
+					while not state[1]:
+						pass
+						time.sleep(0.01)
+				while not state[2]:
+					time.sleep(0.01)
+		t = threading.Thread(target = concurrent_lock)
+		t.start()
+		while not state[0]:
+			time.sleep(0.01)
+		self.failUnlessEqual(ad(), 2)
+		state[1] = True
+		with alock.ExclusiveLock(db, (0,0)):
+			self.failUnlessEqual(ad(), 2)
+			state[2] = True
+			with alock.ExclusiveLock(db, (1,1)):
+				self.failUnlessEqual(ad(), 2)
+		t.join(timeout = 1)
+
 	@pg_tmp
 	def testALockNoWait(self):
 		alt = new()
-		ad = db.prepare(
-			"select count(*) FROM pg_locks WHERE locktype = 'advisory'"
-		).first
+		ad = prepare(n_alocks).first
 		self.failUnlessEqual(ad(), 0)
 		with alock.ExclusiveLock(db, (0,0)):
 			l=alock.ExclusiveLock(alt, (0,0))
@@ -24,9 +55,7 @@ def testALockNoWait(self):
 
 	@pg_tmp
 	def testALock(self):
-		ad = db.prepare(
-			"select count(*) FROM pg_locks WHERE locktype = 'advisory'"
-		).first
+		ad = prepare(n_alocks).first
 		self.failUnlessEqual(ad(), 0)
 		# test a variety..
 		lockids = [
@@ -64,9 +93,7 @@ def testALock(self):
 	@pg_tmp
 	def testPartialALock(self):
 		# Validates that release is properly cleaning up
-		ad = db.prepare(
-			"select count(*) FROM pg_locks WHERE locktype = 'advisory'"
-		).first
+		ad = prepare(n_alocks).first
 		self.failUnlessEqual(ad(), 0)
 		held = (0,-1234)
 		wanted = [0, 324, -1232948, 7, held, 1, (2,4), (834,1)]
@@ -89,9 +116,7 @@ def testALockParameterErrors(self):
 
 	@pg_tmp
 	def testALockOnClosed(self):
-		ad = db.prepare(
-			"select count(*) FROM pg_locks WHERE locktype = 'advisory'"
-		).first
+		ad = prepare(n_alocks).first
 		self.failUnlessEqual(ad(), 0)
 		held = (0,-1234)
 		alt = new()