From 905ec70637191636e1e997e9a42e867a3521f768 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Date: Fri, 15 Oct 2021 13:15:13 +0900
Subject: [PATCH v2 3/5] Remove O(n) behavior of SIInsertDataEntries

SIInsertDataEntries ran a loop over a certain range of procState array
at every insertion.  Get rid of the behavior so that we won't be
annoyed by performance drag that may happen when procState array gets
sparse. This commit removes hasMessage from ProcState. Instead the
function uses maxMsgNum to check for new messages.
---
 src/backend/storage/ipc/sinvaladt.c | 63 +++++++++--------------------
 1 file changed, 20 insertions(+), 43 deletions(-)

diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c
index a90e9920e7..ee3a3accfd 100644
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -143,7 +143,6 @@ typedef struct ProcState
 	int			nextMsgNum;		/* next message number to read */
 	bool		resetState;		/* backend needs to reset its state */
 	bool		signaled;		/* backend has been sent catchup signal */
-	bool		hasMessages;	/* backend has unread messages */
 
 	/*
 	 * Backend only sends invalidations, never receives them. This only makes
@@ -244,7 +243,6 @@ CreateSharedInvalidationState(void)
 		shmInvalBuffer->procState[i].nextMsgNum = 0;	/* meaningless */
 		shmInvalBuffer->procState[i].resetState = false;
 		shmInvalBuffer->procState[i].signaled = false;
-		shmInvalBuffer->procState[i].hasMessages = false;
 		shmInvalBuffer->procState[i].nextLXID = InvalidLocalTransactionId;
 	}
 }
@@ -309,7 +307,6 @@ SharedInvalBackendInit(bool sendOnly)
 	stateP->nextMsgNum = segP->maxMsgNum;
 	stateP->resetState = false;
 	stateP->signaled = false;
-	stateP->hasMessages = false;
 	stateP->sendOnly = sendOnly;
 
 	LWLockRelease(SInvalWriteLock);
@@ -417,7 +414,6 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n)
 		int			nthistime = Min(n, WRITE_QUANTUM);
 		int			numMsgs;
 		int			max;
-		int			i;
 
 		n -= nthistime;
 
@@ -450,24 +446,13 @@ SIInsertDataEntries(const SharedInvalidationMessage *data, int n)
 			max++;
 		}
 
-		/* Update current value of maxMsgNum using spinlock */
-		SpinLockAcquire(&segP->msgnumLock);
+		/*
+		 * Update current value of maxMsgNum without taking locks. Make sure
+		 * the inserted messages habve been flushed to main memory before the
+		 * update of the shared variable,
+		 */
+		pg_memory_barrier();
 		segP->maxMsgNum = max;
-		SpinLockRelease(&segP->msgnumLock);
-
-		/*
-		 * Now that the maxMsgNum change is globally visible, we give everyone
-		 * a swift kick to make sure they read the newly added messages.
-		 * Releasing SInvalWriteLock will enforce a full memory barrier, so
-		 * these (unlocked) changes will be committed to memory before we exit
-		 * the function.
-		 */
-		for (i = 0; i < segP->lastBackend; i++)
-		{
-			ProcState  *stateP = &segP->procState[i];
-
-			stateP->hasMessages = true;
-		}
 
 		LWLockRelease(SInvalWriteLock);
 	}
@@ -523,27 +508,24 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
 	 * invalidations, any such occurrence is not much different than if the
 	 * invalidation had arrived slightly later in the first place.
 	 */
-	if (!stateP->hasMessages)
+	max = segP->maxMsgNum;
+
+	if (stateP->nextMsgNum >= max)
 		return 0;
 
 	LWLockAcquire(SInvalReadLock, LW_SHARED);
 
-	/*
-	 * We must reset hasMessages before determining how many messages we're
-	 * going to read.  That way, if new messages arrive after we have
-	 * determined how many we're reading, the flag will get reset and we'll
-	 * notice those messages part-way through.
-	 *
-	 * Note that, if we don't end up reading all of the messages, we had
-	 * better be certain to reset this flag before exiting!
-	 */
-	stateP->hasMessages = false;
-
-	/* Fetch current value of maxMsgNum using spinlock */
-	SpinLockAcquire(&segP->msgnumLock);
-	max = segP->maxMsgNum;
-	SpinLockRelease(&segP->msgnumLock);
-
+	if (stateP->nextMsgNum < max)
+	{
+		/*
+		 * nextMsgNum has been rewinded before we acquired the lock. Recheck
+		 * with maxMsgNum again.
+		 */
+		max = segP->maxMsgNum;
+		if (stateP->nextMsgNum >= max)
+			return 0;
+	}
+		
 	if (stateP->resetState)
 	{
 		/*
@@ -576,14 +558,9 @@ SIGetDataEntries(SharedInvalidationMessage *data, int datasize)
 	/*
 	 * If we have caught up completely, reset our "signaled" flag so that
 	 * we'll get another signal if we fall behind again.
-	 *
-	 * If we haven't caught up completely, reset the hasMessages flag so that
-	 * we see the remaining messages next time.
 	 */
 	if (stateP->nextMsgNum >= max)
 		stateP->signaled = false;
-	else
-		stateP->hasMessages = true;
 
 	LWLockRelease(SInvalReadLock);
 	return n;
-- 
2.27.0