Significant cleanups in SysV IPC handling (shared mem and semaphores).

IPC key assignment will now work correctly even when multiple postmasters
are using same logical port number (which is possible given -k switch).
There is only one shared-mem segment per postmaster now, not 3.
Rip out broken code for non-TAS case in bufmgr and xlog, substitute a
complete S_LOCK emulation using semaphores in spin.c.  TAS and non-TAS
logic is now exactly the same.
When deadlock is detected, "Deadlock detected" is now the elog(ERROR)
message, rather than a NOTICE that comes out before an unhelpful ERROR.

1 files changed, 236 insertions, 167 deletions

diff --git a/src/backend/storage/ipc/spin.c b/src/backend/storage/ipc/spin.c
index 674ee06a9a3..a93ae69e032 100644
--- a/src/backend/storage/ipc/spin.c
+++ b/src/backend/storage/ipc/spin.c
@@ -3,31 +3,24 @@
  * spin.c
  *	  routines for managing spin locks
  *
+ * POSTGRES has two kinds of locks: semaphores (which put the
+ * process to sleep) and spinlocks (which are supposed to be
+ * short term locks).  Spinlocks are implemented via test-and-set (TAS)
+ * instructions if possible, else via semaphores.  The semaphore method
+ * is too slow to be useful :-(
+ *
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/Attic/spin.c,v 1.25 2000/05/31 00:28:29 petere Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/Attic/spin.c,v 1.26 2000/11/28 23:27:56 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
-/*
- * POSTGRES has two kinds of locks: semaphores (which put the
- * process to sleep) and spinlocks (which are supposed to be
- * short term locks).  Currently both are implemented as SysV
- * semaphores, but presumably this can change if we move to
- * a machine with a test-and-set (TAS) instruction.  Its probably
- * a good idea to think about (and allocate) short term and long
- * term semaphores separately anyway.
- *
- * NOTE: These routines are not supposed to be widely used in Postgres.
- *		 They are preserved solely for the purpose of porting Mark Sullivan's
- *		 buffer manager to Postgres.
- */
-#include <errno.h>
 #include "postgres.h"
 
+#include <errno.h>
 #ifndef HAS_TEST_AND_SET
 #include <sys/sem.h>
 #endif
@@ -35,39 +28,33 @@
 #include "storage/proc.h"
 #include "storage/s_lock.h"
 
-
-/* globals used in this file */
-IpcSemaphoreId SpinLockId;
-
-#ifdef HAS_TEST_AND_SET
-/* real spin lock implementations */
-
-void
-CreateSpinlocks(IPCKey key)
-{
-	/* the spin lock shared memory must have been created by now */
-	return;
-}
-
-void
-InitSpinLocks(void)
-{
-	extern SPINLOCK ShmemLock;
-	extern SPINLOCK ShmemIndexLock;
-	extern SPINLOCK BufMgrLock;
-	extern SPINLOCK LockMgrLock;
-	extern SPINLOCK ProcStructLock;
-	extern SPINLOCK SInvalLock;
-	extern SPINLOCK OidGenLockId;
-	extern SPINLOCK XidGenLockId;
-	extern SPINLOCK ControlFileLockId;
+/* Probably should move these to an appropriate header file */
+extern SPINLOCK ShmemLock;
+extern SPINLOCK ShmemIndexLock;
+extern SPINLOCK BufMgrLock;
+extern SPINLOCK LockMgrLock;
+extern SPINLOCK ProcStructLock;
+extern SPINLOCK SInvalLock;
+extern SPINLOCK OidGenLockId;
+extern SPINLOCK XidGenLockId;
+extern SPINLOCK ControlFileLockId;
 
 #ifdef STABLE_MEMORY_STORAGE
-	extern SPINLOCK MMCacheLock;
+extern SPINLOCK MMCacheLock;
 
 #endif
 
-	/* These six spinlocks have fixed location is shmem */
+
+/*
+ * Initialize identifiers for permanent spinlocks during startup
+ *
+ * The same identifiers are used for both TAS and semaphore implementations,
+ * although in one case they are indexes into a shmem array and in the other
+ * they are semaphore numbers.
+ */
+static void
+InitSpinLockIDs(void)
+{
 	ShmemLock = (SPINLOCK) SHMEMLOCKID;
 	ShmemIndexLock = (SPINLOCK) SHMEMINDEXLOCKID;
 	BufMgrLock = (SPINLOCK) BUFMGRLOCKID;
@@ -81,11 +68,18 @@ InitSpinLocks(void)
 #ifdef STABLE_MEMORY_STORAGE
 	MMCacheLock = (SPINLOCK) MMCACHELOCKID;
 #endif
-
-	return;
 }
 
 
+#ifdef HAS_TEST_AND_SET
+
+/* real spin lock implementation */
+
+typedef struct slock
+{
+	slock_t		shlock;
+} SLock;
+
 #ifdef LOCK_DEBUG
 bool Trace_spinlocks = false;
 
@@ -93,193 +87,268 @@ inline static void
 PRINT_SLDEBUG(const char * where, SPINLOCK lockid, const SLock * lock)
 {
     if (Trace_spinlocks)
-        elog(DEBUG,
-             "%s: id=%d (locklock=%d, flag=%d, nshlocks=%d, shlock=%d, exlock=%d)",
-             where, lockid,
-             lock->locklock, lock->flag, lock->nshlocks, lock->shlock, lock->exlock);
+        elog(DEBUG, "%s: id=%d", where, lockid);
 }
 #else  /* not LOCK_DEBUG */
 #define PRINT_SLDEBUG(a,b,c)
 #endif /* not LOCK_DEBUG */
 
 
-/* from ipc.c */
-extern SLock *SLockArray;
+static SLock *SLockArray = NULL;
+
+#define SLOCKMEMORYSIZE		((int) MAX_SPINS * sizeof(SLock))
+
+/*
+ * SLockShmemSize --- return shared-memory space needed
+ */
+int
+SLockShmemSize(void)
+{
+	return MAXALIGN(SLOCKMEMORYSIZE);
+}
+
+/*
+ * CreateSpinlocks --- create and initialize spinlocks during startup
+ */
+void
+CreateSpinlocks(PGShmemHeader *seghdr)
+{
+	int			id;
+
+	/*
+	 * We must allocate the space "by hand" because shmem.c isn't up yet
+	 */
+	SLockArray = (SLock *) (((char *) seghdr) + seghdr->freeoffset);
+	seghdr->freeoffset += MAXALIGN(SLOCKMEMORYSIZE);
+	Assert(seghdr->freeoffset <= seghdr->totalsize);
+
+	/*
+	 * Initialize all spinlocks to "unlocked" state
+	 */
+	for (id = 0; id < (int) MAX_SPINS; id++)
+	{
+		SLock	   *slckP = &(SLockArray[id]);
+
+		S_INIT_LOCK(&(slckP->shlock));
+	}
+
+	/*
+	 * Assign indexes for fixed spinlocks
+	 */
+	InitSpinLockIDs();
+}
 
 void
 SpinAcquire(SPINLOCK lockid)
 {
-	SLock	   *slckP;
+	SLock	   *slckP = &(SLockArray[lockid]);
 
-	/* This used to be in ipc.c, but move here to reduce function calls */
-	slckP = &(SLockArray[lockid]);
 	PRINT_SLDEBUG("SpinAcquire", lockid, slckP);
-ex_try_again:
-	S_LOCK(&(slckP->locklock));
-	switch (slckP->flag)
-	{
-		case NOLOCK:
-			slckP->flag = EXCLUSIVELOCK;
-			S_LOCK(&(slckP->exlock));
-			S_LOCK(&(slckP->shlock));
-			S_UNLOCK(&(slckP->locklock));
-            PRINT_SLDEBUG("OUT", lockid, slckP);
-			break;
-		case SHAREDLOCK:
-		case EXCLUSIVELOCK:
-			S_UNLOCK(&(slckP->locklock));
-			S_LOCK(&(slckP->exlock));
-			S_UNLOCK(&(slckP->exlock));
-			goto ex_try_again;
-	}
+	S_LOCK(&(slckP->shlock));
 	PROC_INCR_SLOCK(lockid);
-    PRINT_SLDEBUG("SpinAcquire/success", lockid, slckP);
+    PRINT_SLDEBUG("SpinAcquire/done", lockid, slckP);
 }
 
 void
 SpinRelease(SPINLOCK lockid)
 {
-	SLock	   *slckP;
-
-	/* This used to be in ipc.c, but move here to reduce function calls */
-	slckP = &(SLockArray[lockid]);
+	SLock	   *slckP = &(SLockArray[lockid]);
 
 	/*
 	 * Check that we are actually holding the lock we are releasing. This
 	 * can be done only after MyProc has been initialized.
 	 */
     Assert(!MyProc || MyProc->sLocks[lockid] > 0);
-	Assert(slckP->flag != NOLOCK);
-
 
 	PROC_DECR_SLOCK(lockid);
     PRINT_SLDEBUG("SpinRelease", lockid, slckP);
-	S_LOCK(&(slckP->locklock));
-	/* -------------
-	 *	give favor to read processes
-	 * -------------
+	S_UNLOCK(&(slckP->shlock));
+    PRINT_SLDEBUG("SpinRelease/done", lockid, slckP);
+}
+
+#else /* !HAS_TEST_AND_SET */
+
+/*
+ * No TAS, so spinlocks are implemented using SysV semaphores.
+ *
+ * We support two slightly different APIs here: SpinAcquire/SpinRelease
+ * work with SPINLOCK integer indexes for the permanent spinlocks, which
+ * are all assumed to live in the first spinlock semaphore set.  There
+ * is also an emulation of the s_lock.h TAS-spinlock macros; for that case,
+ * typedef slock_t stores the semId and sem number of the sema to use.
+ * The semas needed are created by CreateSpinlocks and doled out by
+ * s_init_lock_sema.
+ *
+ * Since many systems have a rather small SEMMSL limit on semas per set,
+ * we allocate the semaphores required in sets of SPINLOCKS_PER_SET semas.
+ * This value is deliberately made equal to PROC_NSEMS_PER_SET so that all
+ * sema sets allocated by Postgres will be the same size; that eases the
+ * semaphore-recycling logic in IpcSemaphoreCreate().
+ *
+ * Note that the SpinLockIds array is not in shared memory; it is filled
+ * by the postmaster and then inherited through fork() by backends.  This
+ * is OK because its contents do not change after system startup.
+ */
+
+#define SPINLOCKS_PER_SET  PROC_NSEMS_PER_SET
+
+static IpcSemaphoreId *SpinLockIds = NULL;
+
+static int numSpinSets = 0;		/* number of sema sets used */
+static int numSpinLocks = 0;	/* total number of semas allocated */
+static int nextSpinLock = 0;	/* next free spinlock index */
+
+static void SpinFreeAllSemaphores(void);
+
+/*
+ * SLockShmemSize --- return shared-memory space needed
+ */
+int
+SLockShmemSize(void)
+{
+	return 0;
+}
+
+/*
+ * CreateSpinlocks --- create and initialize spinlocks during startup
+ */
+void
+CreateSpinlocks(PGShmemHeader *seghdr)
+{
+	int		i;
+
+	if (SpinLockIds == NULL)
+	{
+		/*
+		 * Compute number of spinlocks needed.  If this logic gets any more
+		 * complicated, it should be distributed into the affected modules,
+		 * similar to the way shmem space estimation is handled.
+		 *
+		 * For now, though, we just need the fixed spinlocks (MAX_SPINS),
+		 * two spinlocks per shared disk buffer, and four spinlocks for XLOG.
+		 */
+		numSpinLocks = (int) MAX_SPINS + 2 * NBuffers + 4;
+
+		/* might as well round up to a multiple of SPINLOCKS_PER_SET */
+		numSpinSets = (numSpinLocks - 1) / SPINLOCKS_PER_SET + 1;
+		numSpinLocks = numSpinSets * SPINLOCKS_PER_SET;
+
+		SpinLockIds = (IpcSemaphoreId *)
+			malloc(numSpinSets * sizeof(IpcSemaphoreId));
+		Assert(SpinLockIds != NULL);
+	}
+
+	for (i = 0; i < numSpinSets; i++)
+		SpinLockIds[i] = -1;
+
+	/*
+	 * Arrange to delete semas on exit --- set this up now so that we
+	 * will clean up if allocation fails.  We use our own freeproc,
+	 * rather than IpcSemaphoreCreate's removeOnExit option, because
+	 * we don't want to fill up the on_shmem_exit list with a separate
+	 * entry for each semaphore set.
 	 */
-	slckP->flag = NOLOCK;
-	if (slckP->nshlocks > 0)
+	on_shmem_exit(SpinFreeAllSemaphores, 0);
+
+	/* Create sema sets and set all semas to count 1 */
+	for (i = 0; i < numSpinSets; i++)
 	{
-		while (slckP->nshlocks > 0)
-		{
-			S_UNLOCK(&(slckP->shlock));
-			S_LOCK(&(slckP->comlock));
-		}
-		S_UNLOCK(&(slckP->shlock));
+		SpinLockIds[i] = IpcSemaphoreCreate(SPINLOCKS_PER_SET,
+											IPCProtection,
+											1,
+											false);
 	}
-	else
-		S_UNLOCK(&(slckP->shlock));
-	S_UNLOCK(&(slckP->exlock));
-	S_UNLOCK(&(slckP->locklock));
-    PRINT_SLDEBUG("SpinRelease/released", lockid, slckP);
+
+	/*
+	 * Assign indexes for fixed spinlocks
+	 */
+	Assert(MAX_SPINS <= SPINLOCKS_PER_SET);
+	InitSpinLockIDs();
+
+	/* Init counter for allocating dynamic spinlocks */
+	nextSpinLock = MAX_SPINS;
 }
 
-#else /* !HAS_TEST_AND_SET */
-/* Spinlocks are implemented using SysV semaphores */
+/*
+ * SpinFreeAllSemaphores -
+ *	  called at shmem_exit time, ie when exiting the postmaster or
+ *	  destroying shared state for a failed set of backends.
+ *	  Free up all the semaphores allocated for spinlocks.
+ */
+static void
+SpinFreeAllSemaphores(void)
+{
+	int			i;
 
-static bool AttachSpinLocks(IPCKey key);
-static bool SpinIsLocked(SPINLOCK lock);
+	for (i = 0; i < numSpinSets; i++)
+	{
+		if (SpinLockIds[i] >= 0)
+			IpcSemaphoreKill(SpinLockIds[i]);
+	}
+}
 
 /*
- * SpinAcquire -- try to grab a spinlock
+ * SpinAcquire -- grab a fixed spinlock
  *
  * FAILS if the semaphore is corrupted.
  */
 void
 SpinAcquire(SPINLOCK lock)
 {
-	IpcSemaphoreLock(SpinLockId, lock, IpcExclusiveLock);
+	IpcSemaphoreLock(SpinLockIds[0], lock);
 	PROC_INCR_SLOCK(lock);
 }
 
 /*
- * SpinRelease -- release a spin lock
+ * SpinRelease -- release a fixed spin lock
  *
  * FAILS if the semaphore is corrupted
  */
 void
 SpinRelease(SPINLOCK lock)
 {
-	Assert(SpinIsLocked(lock));
-	PROC_DECR_SLOCK(lock);
-	IpcSemaphoreUnlock(SpinLockId, lock, IpcExclusiveLock);
-}
-
-static bool
-SpinIsLocked(SPINLOCK lock)
-{
+#ifdef USE_ASSERT_CHECKING
+	/* Check it's locked */
 	int			semval;
 
-	semval = IpcSemaphoreGetValue(SpinLockId, lock);
-	return semval < IpcSemaphoreDefaultStartValue;
+	semval = IpcSemaphoreGetValue(SpinLockIds[0], lock);
+	Assert(semval < 1);
+#endif
+	PROC_DECR_SLOCK(lock);
+	IpcSemaphoreUnlock(SpinLockIds[0], lock);
 }
 
 /*
- * CreateSpinlocks -- Create a sysV semaphore array for
- *		the spinlocks
- *
+ * s_lock.h hardware-spinlock emulation
  */
+
 void
-CreateSpinlocks(IPCKey key)
+s_init_lock_sema(volatile slock_t *lock)
 {
-
-	SpinLockId = IpcSemaphoreCreate(key, MAX_SPINS, IPCProtection,
-									IpcSemaphoreDefaultStartValue, 1);
-
-	if (SpinLockId <= 0)
-		elog(STOP, "CreateSpinlocks: cannot create spin locks");
-
-	return;
+	if (nextSpinLock >= numSpinLocks)
+		elog(FATAL, "s_init_lock_sema: not enough semaphores");
+	lock->semId = SpinLockIds[nextSpinLock / SPINLOCKS_PER_SET];
+	lock->sem = nextSpinLock % SPINLOCKS_PER_SET;
+	nextSpinLock++;
 }
 
-/*
- * InitSpinLocks -- Spinlock bootstrapping
- *
- * We need several spinlocks for bootstrapping:
- * ShmemIndexLock (for the shmem index table) and
- * ShmemLock (for the shmem allocator), BufMgrLock (for buffer
- * pool exclusive access), LockMgrLock (for the lock table), and
- * ProcStructLock (a spin lock for the shared process structure).
- * If there's a Sony WORM drive attached, we also have a spinlock
- * (SJCacheLock) for it.  Same story for the main memory storage mgr.
- *
- */
 void
-InitSpinLocks(void)
+s_unlock_sema(volatile slock_t *lock)
 {
-	extern SPINLOCK ShmemLock;
-	extern SPINLOCK ShmemIndexLock;
-	extern SPINLOCK BufMgrLock;
-	extern SPINLOCK LockMgrLock;
-	extern SPINLOCK ProcStructLock;
-	extern SPINLOCK SInvalLock;
-	extern SPINLOCK OidGenLockId;
-	extern SPINLOCK XidGenLockId;
-	extern SPINLOCK ControlFileLockId;
-
-#ifdef STABLE_MEMORY_STORAGE
-	extern SPINLOCK MMCacheLock;
-
-#endif
-
-	/* These five (or six) spinlocks have fixed location is shmem */
-	ShmemLock = (SPINLOCK) SHMEMLOCKID;
-	ShmemIndexLock = (SPINLOCK) SHMEMINDEXLOCKID;
-	BufMgrLock = (SPINLOCK) BUFMGRLOCKID;
-	LockMgrLock = (SPINLOCK) LOCKMGRLOCKID;
-	ProcStructLock = (SPINLOCK) PROCSTRUCTLOCKID;
-	SInvalLock = (SPINLOCK) SINVALLOCKID;
-	OidGenLockId = (SPINLOCK) OIDGENLOCKID;
-	XidGenLockId = (SPINLOCK) XIDGENLOCKID;
-	ControlFileLockId = (SPINLOCK) CNTLFILELOCKID;
+	IpcSemaphoreUnlock(lock->semId, lock->sem);
+}
 
-#ifdef STABLE_MEMORY_STORAGE
-	MMCacheLock = (SPINLOCK) MMCACHELOCKID;
-#endif
+bool
+s_lock_free_sema(volatile slock_t *lock)
+{
+	return IpcSemaphoreGetValue(lock->semId, lock->sem) > 0;
+}
 
-	return;
+int
+tas_sema(volatile slock_t *lock)
+{
+	/* Note that TAS macros return 0 if *success* */
+	return ! IpcSemaphoreTryLock(lock->semId, lock->sem);
 }
 
 #endif /* !HAS_TEST_AND_SET */