From ce90bd9a1e2c8b5783ebeea83594da7d3a1c63de Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki.linnakangas@iki.fi>
Date: Wed, 15 Jul 2015 18:33:28 +0300
Subject: [PATCH 1/1] Fix LWLock "variable" support, broken by the lwlock
 scalability patch

---
 src/backend/access/transam/xlog.c |  15 ++---
 src/backend/storage/lmgr/lwlock.c | 135 ++++++++++++++++++++++----------------
 src/include/storage/lwlock.h      |   1 -
 3 files changed, 83 insertions(+), 68 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1dd31b3..2f34346 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -1408,9 +1408,7 @@ WALInsertLockAcquire(void)
 	 * The insertingAt value is initially set to 0, as we don't know our
 	 * insert location yet.
 	 */
-	immed = LWLockAcquireWithVar(&WALInsertLocks[MyLockNo].l.lock,
-								 &WALInsertLocks[MyLockNo].l.insertingAt,
-								 0);
+	immed = LWLockAcquire(&WALInsertLocks[MyLockNo].l.lock, LW_EXCLUSIVE);
 	if (!immed)
 	{
 		/*
@@ -1442,13 +1440,12 @@ WALInsertLockAcquireExclusive(void)
 	 */
 	for (i = 0; i < NUM_XLOGINSERT_LOCKS - 1; i++)
 	{
-		LWLockAcquireWithVar(&WALInsertLocks[i].l.lock,
-							 &WALInsertLocks[i].l.insertingAt,
-							 PG_UINT64_MAX);
+		LWLockAcquire(&WALInsertLocks[i].l.lock, LW_EXCLUSIVE);
+		LWLockUpdateVar(&WALInsertLocks[i].l.lock,
+						&WALInsertLocks[i].l.insertingAt,
+						PG_UINT64_MAX);
 	}
-	LWLockAcquireWithVar(&WALInsertLocks[i].l.lock,
-						 &WALInsertLocks[i].l.insertingAt,
-						 0);
+	LWLockAcquire(&WALInsertLocks[i].l.lock, LW_EXCLUSIVE);
 
 	holdingAllLocks = true;
 }
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index 46cab49..92a1aef 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -11,12 +11,12 @@
  * LWLocks to protect its shared state.
  *
  * In addition to exclusive and shared modes, lightweight locks can be used
- * to wait until a variable changes value.  The variable is initially set
- * when the lock is acquired with LWLockAcquireWithVar, and can be updated
- * without releasing the lock by calling LWLockUpdateVar.  LWLockWaitForVar
- * waits for the variable to be updated, or until the lock is free.  The
- * meaning of the variable is up to the caller, the lightweight lock code
- * just assigns and compares it.
+ * to wait until a variable changes value.  The variable is initially put
+ * in a "not-set" state when the lock is acquired with LWLockAcquire, and
+ * can be updated without releasing the lock by calling LWLockUpdateVar.
+ * LWLockWaitForVar waits for the variable to be updated, or until the lock
+ * is free.  The meaning of the variable is up to the caller, the lightweight
+ * lock code just assigns and compares it.
  *
  * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -99,6 +99,7 @@ extern slock_t *ShmemLock;
 
 #define LW_FLAG_HAS_WAITERS			((uint32) 1 << 30)
 #define LW_FLAG_RELEASE_OK			((uint32) 1 << 29)
+#define LW_FLAG_VAR_SET				((uint32) 1 << 28)
 
 #define LW_VAL_EXCLUSIVE			((uint32) 1 << 24)
 #define LW_VAL_SHARED				1
@@ -150,9 +151,6 @@ static LWLockHandle held_lwlocks[MAX_SIMUL_LWLOCKS];
 static int	lock_addin_request = 0;
 static bool lock_addin_request_allowed = true;
 
-static inline bool LWLockAcquireCommon(LWLock *l, LWLockMode mode,
-					uint64 *valptr, uint64 val);
-
 #ifdef LWLOCK_STATS
 typedef struct lwlock_stats_key
 {
@@ -601,7 +599,14 @@ LWLockAttemptLock(LWLock *lock, LWLockMode mode)
 		{
 			lock_free = (expected_state & LW_LOCK_MASK) == 0;
 			if (lock_free)
-				desired_state += LW_VAL_EXCLUSIVE;
+			{
+				/*
+				 * Make the lock exclusively-locked for us, and make the
+				 * associated variable, if any, un-set.
+				 */
+				desired_state =
+					(desired_state + LW_VAL_EXCLUSIVE) & ~LW_FLAG_VAR_SET;
+			}
 		}
 		else
 		{
@@ -745,7 +750,7 @@ LWLockWakeup(LWLock *lock)
  * NB: Mode can be LW_WAIT_UNTIL_FREE here!
  */
 static void
-LWLockQueueSelf(LWLock *lock, LWLockMode mode)
+LWLockQueueSelf(LWLock *lock, LWLockMode mode, bool spinlock_held)
 {
 #ifdef LWLOCK_STATS
 	lwlock_stats *lwstats;
@@ -764,11 +769,14 @@ LWLockQueueSelf(LWLock *lock, LWLockMode mode)
 	if (MyProc->lwWaiting)
 		elog(PANIC, "queueing for lock while waiting on another one");
 
+	if (!spinlock_held)
+	{
 #ifdef LWLOCK_STATS
-	lwstats->spin_delay_count += SpinLockAcquire(&lock->mutex);
+		lwstats->spin_delay_count += SpinLockAcquire(&lock->mutex);
 #else
-	SpinLockAcquire(&lock->mutex);
+		SpinLockAcquire(&lock->mutex);
 #endif
+	}
 
 	/* setting the flag is protected by the spinlock */
 	pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_HAS_WAITERS);
@@ -783,7 +791,8 @@ LWLockQueueSelf(LWLock *lock, LWLockMode mode)
 		dlist_push_tail(&lock->waiters, &MyProc->lwWaitLink);
 
 	/* Can release the mutex now */
-	SpinLockRelease(&lock->mutex);
+	if (!spinlock_held)
+		SpinLockRelease(&lock->mutex);
 
 #ifdef LOCK_DEBUG
 	pg_atomic_fetch_add_u32(&lock->nwaiters, 1);
@@ -900,25 +909,7 @@ LWLockDequeueSelf(LWLock *lock)
  * Side effect: cancel/die interrupts are held off until lock release.
  */
 bool
-LWLockAcquire(LWLock *l, LWLockMode mode)
-{
-	return LWLockAcquireCommon(l, mode, NULL, 0);
-}
-
-/*
- * LWLockAcquireWithVar - like LWLockAcquire, but also sets *valptr = val
- *
- * The lock is always acquired in exclusive mode with this function.
- */
-bool
-LWLockAcquireWithVar(LWLock *l, uint64 *valptr, uint64 val)
-{
-	return LWLockAcquireCommon(l, LW_EXCLUSIVE, valptr, val);
-}
-
-/* internal function to implement LWLockAcquire and LWLockAcquireWithVar */
-static inline bool
-LWLockAcquireCommon(LWLock *lock, LWLockMode mode, uint64 *valptr, uint64 val)
+LWLockAcquire(LWLock *lock, LWLockMode mode)
 {
 	PGPROC	   *proc = MyProc;
 	bool		result = true;
@@ -1003,7 +994,7 @@ LWLockAcquireCommon(LWLock *lock, LWLockMode mode, uint64 *valptr, uint64 val)
 		 */
 
 		/* add to the queue */
-		LWLockQueueSelf(lock, mode);
+		LWLockQueueSelf(lock, mode, false);
 
 		/* we're now guaranteed to be woken up if necessary */
 		mustwait = LWLockAttemptLock(lock, mode);
@@ -1065,10 +1056,6 @@ LWLockAcquireCommon(LWLock *lock, LWLockMode mode, uint64 *valptr, uint64 val)
 		result = false;
 	}
 
-	/* If there's a variable associated with this lock, initialize it */
-	if (valptr)
-		*valptr = val;
-
 	TRACE_POSTGRESQL_LWLOCK_ACQUIRE(T_NAME(lock), T_ID(lock), mode);
 
 	/* Add lock to list of locks held by this backend */
@@ -1181,7 +1168,7 @@ LWLockAcquireOrWait(LWLock *lock, LWLockMode mode)
 
 	if (mustwait)
 	{
-		LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE);
+		LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE, false);
 
 		mustwait = LWLockAttemptLock(lock, mode);
 
@@ -1315,24 +1302,53 @@ LWLockWaitForVar(LWLock *lock, uint64 *valptr, uint64 oldval, uint64 *newval)
 	{
 		bool		mustwait;
 		uint64		value;
+		uint32		state;
 
 		mustwait = (pg_atomic_read_u32(&lock->state) & LW_VAL_EXCLUSIVE) != 0;
 
-		if (mustwait)
-		{
-			/*
-			 * Perform comparison using spinlock as we can't rely on atomic 64
-			 * bit reads/stores.
-			 */
+		if (!mustwait)
+			break;				/* the lock was free */
+
+		/*
+		 * Check if the variable is set and whether its current value matches
+		 * the old value the caller provided. We have to hold the spinlock
+		 * while we check that, until we have queued ourselves to the lock,
+		 * so that the lock-holder cannot change the value without waking us
+		 * up.
+		 */
 #ifdef LWLOCK_STATS
-			lwstats->spin_delay_count += SpinLockAcquire(&lock->mutex);
+		lwstats->spin_delay_count += SpinLockAcquire(&lock->mutex);
 #else
-			SpinLockAcquire(&lock->mutex);
+		SpinLockAcquire(&lock->mutex);
 #endif
 
+		/*
+		 * XXX: We can significantly optimize this on platforms with 64bit
+		 * atomics.
+		 */
+		state = pg_atomic_read_u32(&lock->state);
+		if ((state & LW_VAL_EXCLUSIVE) == 0)
+		{
 			/*
-			 * XXX: We can significantly optimize this on platforms with 64bit
-			 * atomics.
+			 * The lock is free (or at least not exclusively-locked). We can
+			 * return to caller immediately.
+			 */
+			result = true;
+			mustwait = false;
+		}
+		else if ((state & LW_FLAG_VAR_SET) == 0)
+		{
+			/*
+			 * The lock is held, but the locker has not set the variable
+			 * yet. We have to wait.
+			 */
+			mustwait = true;
+		}
+		else
+		{
+			/*
+			 * The lock is held, and the locker has set the variable. Compare
+			 * the variable's current value.
 			 */
 			value = *valptr;
 			if (value != oldval)
@@ -1341,23 +1357,21 @@ LWLockWaitForVar(LWLock *lock, uint64 *valptr, uint64 oldval, uint64 *newval)
 				mustwait = false;
 				*newval = value;
 			}
-			else
-				mustwait = true;
-			SpinLockRelease(&lock->mutex);
 		}
-		else
-			mustwait = false;
 
 		if (!mustwait)
-			break;				/* the lock was free or value didn't match */
+		{
+			SpinLockRelease(&lock->mutex);
+			break;			/* the lock was free or the value didn't match */
+		}
 
-		/*
+		/* XXX: is this racy still?
 		 * Add myself to wait queue. Note that this is racy, somebody else
 		 * could wakeup before we're finished queuing. NB: We're using nearly
 		 * the same twice-in-a-row lock acquisition protocol as
 		 * LWLockAcquire(). Check its comments for details.
 		 */
-		LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE);
+		LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE, true);
 
 		/*
 		 * Set RELEASE_OK flag, to make sure we get woken up as soon as the
@@ -1367,10 +1381,13 @@ LWLockWaitForVar(LWLock *lock, uint64 *valptr, uint64 oldval, uint64 *newval)
 
 		/*
 		 * We're now guaranteed to be woken up if necessary. Recheck the
-		 * lock's state.
+		 * lock's state. No-one could change the protected variable's value
+		 * while we're holding the spinlock, but the lock may have been freed.
 		 */
 		mustwait = (pg_atomic_read_u32(&lock->state) & LW_VAL_EXCLUSIVE) != 0;
 
+		SpinLockRelease(&lock->mutex);
+
 		/* Ok, lock is free after we queued ourselves. Undo queueing. */
 		if (!mustwait)
 		{
@@ -1495,6 +1512,8 @@ LWLockUpdateVar(LWLock *lock, uint64 *valptr, uint64 val)
 		dlist_push_tail(&wakeup, &waiter->lwWaitLink);
 	}
 
+	pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_VAR_SET);
+
 	/* We are done updating shared state of the lock itself. */
 	SpinLockRelease(&lock->mutex);
 
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index cff3b99..a0ee608 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -185,7 +185,6 @@ extern void LWLockRelease(LWLock *lock);
 extern void LWLockReleaseAll(void);
 extern bool LWLockHeldByMe(LWLock *lock);
 
-extern bool LWLockAcquireWithVar(LWLock *lock, uint64 *valptr, uint64 val);
 extern bool LWLockWaitForVar(LWLock *lock, uint64 *valptr, uint64 oldval, uint64 *newval);
 extern void LWLockUpdateVar(LWLock *lock, uint64 *valptr, uint64 value);
 
-- 
2.1.4

