Slim the sync-rep queue lock (4-patch series for upstream)

src/backend/replication/syncrep.c

@@ -84,6 +84,7 @@
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
 #include "utils/guc_hooks.h"
+#include "utils/memutils.h"
 #include "utils/ps_status.h"
 #include "utils/wait_event.h"
 
@@ -100,7 +101,9 @@ static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
 
 static void SyncRepQueueInsert(int mode);
 static void SyncRepCancelWait(void);
-static int	SyncRepWakeQueue(bool all, int mode);
+static PGPROC **SyncRepWakeList(void);
+static int	SyncRepWakeQueue(bool all, int mode,
+							 PGPROC **wakelist, int *nwake);
 
 static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePtr,
 								 XLogRecPtr *flushPtr,
@@ -190,6 +193,17 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
 	Assert(dlist_node_is_detached(&MyProc->syncRepLinks));
 	Assert(WalSndCtl != NULL);
 
+	/*
+	 * A published watermark that already covers this LSN says a valid quorum
+	 * acknowledged it, which is the same answer the check below the lock
+	 * would give.  The watermark only ever moves forward, so a stale read can
+	 * only send us to take the lock for nothing, never past a wait we owe.
+	 * How often this exit fires depends on the wait mode: it needs the
+	 * acknowledgement to have arrived before the committer got here.
+	 */
+	if (lsn <= (XLogRecPtr) pg_atomic_read_u64(&WalSndCtl->lsn_published[mode]))
+		return;
+
 	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
 	Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING);
 
@@ -491,6 +505,8 @@ SyncRepReleaseWaiters(void)
 	int			numwrite = 0;
 	int			numflush = 0;
 	int			numapply = 0;
+	PGPROC	  **wakelist;
+	int			nwake = 0;
 
 	/*
 	 * If this WALSender is serving a standby that is not on the list of
@@ -509,18 +525,15 @@ SyncRepReleaseWaiters(void)
 	}
 
 	/*
-	 * We're a potential sync standby. Release waiters if there are enough
-	 * sync standbys and we are considered as sync.
-	 */
-	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
-
-	/*
-	 * Check whether we are a sync standby or not, and calculate the synced
-	 * positions among all sync standbys.  (Note: although this step does not
-	 * of itself require holding SyncRepLock, it seems like a good idea to do
-	 * it after acquiring the lock.  This ensures that the WAL pointers we use
-	 * to release waiters are newer than any previous execution of this
-	 * routine used.)
+	 * We're a potential sync standby.  Check whether we are a sync standby
+	 * and calculate the synced positions among all sync standbys before
+	 * taking the lock: the walk over the walsender slots takes their
+	 * spinlocks, allocates, and possibly sorts, and doing all of it under
+	 * SyncRepLock delays every committer.
+	 *
+	 * Positions gone stale by the time the lock is held cost nothing.  The
+	 * guards further down only ever move lsn[] forward, so a reading older
+	 * than a concurrent walsender's simply releases nobody.
 	 */
 	got_recptr = SyncRepGetSyncRecPtr(&writePtr, &flushPtr, &applyPtr, &am_sync);
 
@@ -548,33 +561,59 @@ SyncRepReleaseWaiters(void)
 	 */
 	if (!got_recptr || !am_sync)
 	{
-		LWLockRelease(SyncRepLock);
 		announce_next_takeover = !am_sync;
 		return;
 	}
 
+	wakelist = SyncRepWakeList();
+
+	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
+
 	/*
 	 * Set the lsn first so that when we wake backends they will release up to
 	 * this location.
 	 */
 	if (WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] < writePtr)
 	{
 		WalSndCtl->lsn[SYNC_REP_WAIT_WRITE] = writePtr;
-		numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
+		pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_WRITE],
+							(uint64) writePtr);
+		numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE,
+									wakelist, &nwake);
 	}
 	if (WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr)
 	{
 		WalSndCtl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr;
-		numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
+		pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_FLUSH],
+							(uint64) flushPtr);
+		numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH,
+									wakelist, &nwake);
 	}
 	if (WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr)
 	{
 		WalSndCtl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr;
-		numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY);
+		pg_atomic_write_u64(&WalSndCtl->lsn_published[SYNC_REP_WAIT_APPLY],
+							(uint64) applyPtr);
+		numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY,
+									wakelist, &nwake);
 	}
 
 	LWLockRelease(SyncRepLock);
 
+	/*
+	 * Wake the released backends now that the lock is down.  Each latch is a
+	 * kill() for a sleeping proc, and running one per released commit inside
+	 * the exclusive section makes every committer wait for those syscalls.
+	 * The procs below are off the queue with their state already complete, so
+	 * nothing here needs the lock's protection.  Nothing here can error out
+	 * either: only this process dying outright could leave a released proc
+	 * completed but unlatched, and the in-lock SetLatch had that same window.
+	 * A proc that noticed its state on its own and moved on, even into a new
+	 * wait, gets a spurious latch set, which every latch sleeper tolerates.
+	 */
+	for (int i = 0; i < nwake; i++)
+		SetLatch(&(wakelist[i]->procLatch));
+
 	elog(DEBUG3, "released %d procs up to write %X/%08X, %d procs up to flush %X/%08X, %d procs up to apply %X/%08X",
 		 numwrite, LSN_FORMAT_ARGS(writePtr),
 		 numflush, LSN_FORMAT_ARGS(flushPtr),
@@ -902,16 +941,38 @@ SyncRepGetStandbyPriority(void)
 	return (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ? priority : 1;
 }
 
+/*
+ * Return the list a release collects the procs to latch into, allocating it
+ * the first time this process releases anybody.  It is sized for every
+ * backend to be waiting at once; a proc waits in at most one queue, so one
+ * list of that size is enough for a pass over all three.
+ */
+static PGPROC **
+SyncRepWakeList(void)
+{
+	static PGPROC **wakelist = NULL;
+
+	if (wakelist == NULL)
+		wakelist = (PGPROC **)
+			MemoryContextAlloc(TopMemoryContext,
+							   MaxBackends * sizeof(PGPROC *));
+	return wakelist;
+}
+
 /*
  * Walk the specified queue from head.  Set the state of any backends that
- * need to be woken, remove them from the queue, and then wake them.
- * Pass all = true to wake whole queue; otherwise, just wake up to
+ * need to be woken and remove them from the queue; the procs to wake are
+ * appended to wakelist for the caller to latch once the lock is down.
+ * Pass all = true to release the whole queue; otherwise, just release up to
  * the walsender's LSN.
  *
- * The caller must hold SyncRepLock in exclusive mode.
+ * The caller must hold SyncRepLock in exclusive mode, and must set the
+ * latches itself after releasing it.  Unlink, barrier and state stay together
+ * in here: a waiter reads syncRepState without the lock and must never find
+ * itself completed while still on the queue.
  */
 static int
-SyncRepWakeQueue(bool all, int mode)
+SyncRepWakeQueue(bool all, int mode, PGPROC **wakelist, int *nwake)
 {
 	int			numprocs = 0;
 	dlist_mutable_iter iter;
@@ -948,10 +1009,9 @@ SyncRepWakeQueue(bool all, int mode)
 		 */
 		proc->syncRepState = SYNC_REP_WAIT_COMPLETE;
 
-		/*
-		 * Wake only when we have set state and removed from queue.
-		 */
-		SetLatch(&(proc->procLatch));
+		/* the list is sized to every process that can ever queue here */
+		Assert(*nwake < MaxBackends);
+		wakelist[(*nwake)++] = proc;
 
 		numprocs++;
 	}
@@ -974,6 +1034,9 @@ SyncRepUpdateSyncStandbysDefined(void)
 	if (sync_standbys_defined !=
 		((WalSndCtl->sync_standbys_status & SYNC_STANDBY_DEFINED) != 0))
 	{
+		PGPROC	  **wakelist = SyncRepWakeList();
+		int			nwake = 0;
+
 		LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
 
 		/*
@@ -986,7 +1049,7 @@ SyncRepUpdateSyncStandbysDefined(void)
 			int			i;
 
 			for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
-				SyncRepWakeQueue(true, i);
+				SyncRepWakeQueue(true, i, wakelist, &nwake);
 		}
 
 		/*
@@ -1000,6 +1063,10 @@ SyncRepUpdateSyncStandbysDefined(void)
 			(sync_standbys_defined ? SYNC_STANDBY_DEFINED : 0);
 
 		LWLockRelease(SyncRepLock);
+
+		/* wake the released backends now that the lock is down */
+		for (int i = 0; i < nwake; i++)
+			SetLatch(&(wakelist[i]->procLatch));
 	}
 	else if ((WalSndCtl->sync_standbys_status & SYNC_STANDBY_INIT) == 0)
 	{

src/backend/replication/walsender.c

@@ -94,6 +94,7 @@
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/guc.h"
+#include "utils/injection_point.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/pg_lsn.h"
@@ -206,6 +207,13 @@ static TimestampTz last_reply_timestamp = 0;
 /* Have we sent a heartbeat message asking for reply, since last reply? */
 static bool waiting_for_ping_response = false;
 
+/*
+ * Set when a standby reply has updated this walsender's positions and the
+ * waiters those positions release have not been released yet.  Raised per
+ * reply, acted on once per drain of the socket.
+ */
+static bool syncrep_release_pending = false;
+
 /* Timestamp when walsender received the shutdown request */
 static TimestampTz shutdown_request_timestamp = 0;
 
@@ -300,6 +308,7 @@ static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd);
 static void DropReplicationSlot(DropReplicationSlotCmd *cmd);
 static void StartReplication(StartReplicationCmd *cmd);
 static void StartLogicalReplication(StartReplicationCmd *cmd);
+static void SyncRepFlushPendingRelease(void);
 static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
 static void ProcessStandbyHSFeedbackMessage(void);
@@ -381,6 +390,15 @@ WalSndErrorCleanup(void)
 	pgstat_report_wait_end();
 	pgaio_error_cleanup();
 
+	/*
+	 * A release deferred by the reply drain survives an error thrown while a
+	 * later message in the same drain was being parsed.  The positions the
+	 * drained reply put in shared memory are valid whatever came after it,
+	 * and the committers it acknowledged have no other process to wake them.
+	 * The locks are released above, so the queue lock is free to take.
+	 */
+	SyncRepFlushPendingRelease();
+
 	if (xlogreader != NULL && xlogreader->seg.ws_file >= 0)
 		wal_segment_close(xlogreader);
 
@@ -2353,6 +2371,22 @@ exec_replication_command(const char *cmd_string)
 	return true;
 }
 
+/*
+ * Run the release a drained reply deferred.  A reply already processed has
+ * put its positions in shared memory, and the committers it acknowledged
+ * have nothing but this process to wake them, so every exit out of the reply
+ * drain runs through here before leaving.  The standby's goodbye is the
+ * common one.
+ */
+static void
+SyncRepFlushPendingRelease(void)
+{
+	if (!syncrep_release_pending)
+		return;
+	syncrep_release_pending = false;
+	SyncRepReleaseWaiters();
+}
+
 /*
  * Process any incoming messages while streaming. Also checks if the remote
  * end has closed the connection.
@@ -2379,6 +2413,7 @@ ProcessRepliesIfAny(void)
 		if (r < 0)
 		{
 			/* unexpected error or EOF */
+			SyncRepFlushPendingRelease();
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
 					 errmsg("unexpected EOF on standby connection")));
@@ -2402,6 +2437,7 @@ ProcessRepliesIfAny(void)
 				maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
 				break;
 			default:
+				SyncRepFlushPendingRelease();
 				ereport(FATAL,
 						(errcode(ERRCODE_PROTOCOL_VIOLATION),
 						 errmsg("invalid standby message type \"%c\"",
@@ -2414,6 +2450,7 @@ ProcessRepliesIfAny(void)
 		resetStringInfo(&reply_message);
 		if (pq_getmessage(&reply_message, maxmsglen))
 		{
+			SyncRepFlushPendingRelease();
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
 					 errmsg("unexpected EOF on standby connection")));
@@ -2429,6 +2466,7 @@ ProcessRepliesIfAny(void)
 				 */
 			case PqMsg_CopyData:
 				ProcessStandbyMessage();
+				INJECTION_POINT("walsender-reply-drained", NULL);
 				received = true;
 				break;
 
@@ -2450,9 +2488,12 @@ ProcessRepliesIfAny(void)
 
 				/*
 				 * PqMsg_Terminate means that the standby is closing down the
-				 * socket.
+				 * socket.  The last reply it sent is drained already, and
+				 * what that reply acknowledged must not leave with this
+				 * process.
 				 */
 			case PqMsg_Terminate:
+				SyncRepFlushPendingRelease();
 				proc_exit(0);
 
 			default:
@@ -2468,6 +2509,13 @@ ProcessRepliesIfAny(void)
 		last_reply_timestamp = last_processing;
 		waiting_for_ping_response = false;
 	}
+
+	/*
+	 * One release covers every reply drained above: the positions in shared
+	 * memory are already the newest ones, and each release takes SyncRepLock
+	 * exclusively.
+	 */
+	SyncRepFlushPendingRelease();
 }
 
 /*
@@ -2498,6 +2546,7 @@ ProcessStandbyMessage(void)
 			break;
 
 		default:
+			SyncRepFlushPendingRelease();
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
 					 errmsg("unexpected message type \"%c\"", msgtype)));
@@ -2633,8 +2682,14 @@ ProcessStandbyReplyMessage(void)
 		SpinLockRelease(&walsnd->mutex);
 	}
 
+	/*
+	 * The release is left for ProcessRepliesIfAny() to run once per drain of
+	 * the socket: several replies routinely sit in the buffer together, and
+	 * every release takes SyncRepLock exclusively to compute positions this
+	 * message has just made stale anyway.
+	 */
 	if (!am_cascading_walsender)
-		SyncRepReleaseWaiters();
+		syncrep_release_pending = true;
 
 	/*
 	 * Advance our local xmin horizon when the client confirmed a flush.
@@ -4019,7 +4074,10 @@ static void
 WalSndShmemInit(void *arg)
 {
 	for (int i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+	{
 		dlist_init(&(WalSndCtl->SyncRepQueue[i]));
+		pg_atomic_init_u64(&(WalSndCtl->lsn_published[i]), 0);
+	}
 
 	for (int i = 0; i < max_wal_senders; i++)
 	{