Modify sinval so that InvalidateSharedInvalid() does not hold

the SInval spinlock while it is calling the passed invalFunction or
resetFunction.  This is necessary to avoid deadlock with lmgr change;
InvalidateSharedInvalid can be called recursively now.  It should be
a good performance improvement anyway --- holding a spinlock for more
than a very short interval is a no-no.

3 files changed, 114 insertions, 93 deletions

diff --git a/src/backend/storage/ipc/sinval.c b/src/backend/storage/ipc/sinval.c
index 21585cc2406..e993cef74aa 100644
--- a/src/backend/storage/ipc/sinval.c
+++ b/src/backend/storage/ipc/sinval.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.16 1999/07/15 22:39:49 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.17 1999/09/04 18:36:45 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -21,9 +21,9 @@
 #include "storage/sinval.h"
 #include "storage/sinvaladt.h"
 
-extern SISeg *shmInvalBuffer;	/* the shared buffer segment, set by */
-
- /* SISegmentAttach()			   */
+extern SISeg *shmInvalBuffer;	/* the shared buffer segment, set by
+								 * SISegmentAttach()
+								 */
 extern BackendId MyBackendId;
 extern BackendTag MyBackendTag;
 
@@ -127,21 +127,20 @@ RegisterSharedInvalid(int cacheId,		/* XXX */
 		ItemPointerSetInvalid(&newInvalid.pointerData);
 
 	SpinAcquire(SInvalLock);
-	if (!SISetDataEntry(shmInvalBuffer, &newInvalid))
+	while (!SISetDataEntry(shmInvalBuffer, &newInvalid))
 	{
 		/* buffer full */
 		/* release a message, mark process cache states to be invalid */
 		SISetProcStateInvalid(shmInvalBuffer);
 
-		if (!SIDelDataEntry(shmInvalBuffer))
+		if (!SIDelDataEntries(shmInvalBuffer, 1))
 		{
 			/* inconsistent buffer state -- shd never happen */
 			SpinRelease(SInvalLock);
 			elog(FATAL, "RegisterSharedInvalid: inconsistent buffer state");
 		}
 
-		/* write again */
-		SISetDataEntry(shmInvalBuffer, &newInvalid);
+		/* loop around to try write again */
 	}
 	SpinRelease(SInvalLock);
 }
@@ -157,13 +156,41 @@ RegisterSharedInvalid(int cacheId,		/* XXX */
 /*	should be called by a backend											*/
 /****************************************************************************/
 void
-			InvalidateSharedInvalid(void (*invalFunction) (),
-									void (*resetFunction) ())
+InvalidateSharedInvalid(void (*invalFunction) (),
+						void (*resetFunction) ())
 {
-	SpinAcquire(SInvalLock);
-	SIReadEntryData(shmInvalBuffer, MyBackendId,
-					invalFunction, resetFunction);
+	SharedInvalidData	data;
+	int					getResult;
+	bool				gotMessage = false;
 
-	SIDelExpiredDataEntries(shmInvalBuffer);
-	SpinRelease(SInvalLock);
+	for (;;)
+	{
+		SpinAcquire(SInvalLock);
+		getResult = SIGetDataEntry(shmInvalBuffer, MyBackendId, &data);
+		SpinRelease(SInvalLock);
+		if (getResult == 0)
+			break;				/* nothing more to do */
+		if (getResult < 0)
+		{
+			/* got a reset message */
+			elog(NOTICE, "InvalidateSharedInvalid: cache state reset");
+			resetFunction();
+		}
+		else
+		{
+			/* got a normal data message */
+			invalFunction(data.cacheId,
+						  data.hashIndex,
+						  &data.pointerData);
+		}
+		gotMessage = true;
+	}
+
+	/* If we got any messages, try to release dead messages */
+	if (gotMessage)
+	{
+		SpinAcquire(SInvalLock);
+		SIDelExpiredDataEntries(shmInvalBuffer);
+		SpinRelease(SInvalLock);
+	}
 }
diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c
index cffc4aac229..2e64d027f31 100644
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.23 1999/07/17 20:17:44 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.24 1999/09/04 18:36:45 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -450,20 +450,11 @@ SIGetLastDataEntry(SISeg *segP)
 /************************************************************************/
 /* SIGetNextDataEntry(segP, offset)  returns next data entry			*/
 /************************************************************************/
-static SISegEntry *
-SIGetNextDataEntry(SISeg *segP, Offset offset)
-{
-	SISegEntry *eP;
-
-	if (offset == InvalidOffset)
-		return NULL;
-
-	eP = (SISegEntry *) ((Pointer) segP +
-						 SIGetStartEntrySection(segP) +
-						 offset);
-	return eP;
-}
-
+#define SIGetNextDataEntry(segP,offset) \
+	(((offset) == InvalidOffset) ? (SISegEntry *) NULL : \
+	 (SISegEntry *) ((Pointer) (segP) + \
+					 (segP)->startEntrySection + \
+					 (Offset) (offset)))
 
 /************************************************************************/
 /* SIGetNthDataEntry(segP, n)	returns the n-th data entry in chain	*/
@@ -566,31 +557,38 @@ SIDecProcLimit(SISeg *segP, int num)
 
 
 /************************************************************************/
-/* SIDelDataEntry(segP)		- free the FIRST entry						*/
+/* SIDelDataEntries(segP, n)		- free the FIRST n entries			*/
 /************************************************************************/
 bool
-SIDelDataEntry(SISeg *segP)
+SIDelDataEntries(SISeg *segP, int n)
 {
-	SISegEntry *e1P;
+	int			i;
+
+	if (n <= 0)
+		return false;
 
-	if (!SIDecNumEntries(segP, 1))
+	if (!SIDecNumEntries(segP, n))
 	{
-		/* no entries in buffer */
+		/* not that many entries in buffer */
 		return false;
 	}
 
-	e1P = SIGetFirstDataEntry(segP);
-	SISetStartEntryChain(segP, e1P->next);
-	if (SIGetStartEntryChain(segP) == InvalidOffset)
+	for (i = 1; i <= n; i++)
 	{
-		/* it was the last entry */
-		SISetEndEntryChain(segP, InvalidOffset);
+		SISegEntry *e1P = SIGetFirstDataEntry(segP);
+		SISetStartEntryChain(segP, e1P->next);
+		if (SIGetStartEntryChain(segP) == InvalidOffset)
+		{
+			/* it was the last entry */
+			SISetEndEntryChain(segP, InvalidOffset);
+		}
+		/* free the entry */
+		e1P->isfree = true;
+		e1P->next = SIGetStartFreeSpace(segP);
+		SISetStartFreeSpace(segP, SIEntryOffset(segP, e1P));
 	}
-	/* free the entry */
-	e1P->isfree = true;
-	e1P->next = SIGetStartFreeSpace(segP);
-	SISetStartFreeSpace(segP, SIEntryOffset(segP, e1P));
-	SIDecProcLimit(segP, 1);
+
+	SIDecProcLimit(segP, n);
 	return true;
 }
 
@@ -621,51 +619,51 @@ SISetProcStateInvalid(SISeg *segP)
 }
 
 /************************************************************************/
-/* SIReadEntryData(segP, backendId, function)							*/
-/*						- marks messages to be read by id				*/
-/*						  and executes function							*/
+/* SIGetDataEntry(segP, backendId, data)								*/
+/*		get next SI message for specified backend, if there is one		*/
+/*																		*/
+/*		Possible return values:											*/
+/*			0: no SI message available									*/
+/*			1: next SI message has been extracted into *data			*/
+/*				(there may be more messages available after this one!)	*/
+/*		   -1: SI reset message extracted								*/
 /************************************************************************/
-void
-SIReadEntryData(SISeg *segP,
-				int backendId,
-				void (*invalFunction) (),
-				void (*resetFunction) ())
+int
+SIGetDataEntry(SISeg *segP, int backendId,
+			   SharedInvalidData *data)
 {
-	int			i = 0;
-	SISegEntry *data;
+	SISegEntry *msg;
 
 	Assert(segP->procState[backendId - 1].tag == MyBackendTag);
 
-	if (!segP->procState[backendId - 1].resetState)
+	if (segP->procState[backendId - 1].resetState)
 	{
-		/* invalidate data, but only those, you have not seen yet !! */
-		/* therefore skip read messages */
-		data = SIGetNthDataEntry(segP,
-						   SIGetProcStateLimit(segP, backendId - 1) + 1);
-		while (data != NULL)
-		{
-			i++;
-			segP->procState[backendId - 1].limit++;		/* one more message read */
-			invalFunction(data->entryData.cacheId,
-						  data->entryData.hashIndex,
-						  &data->entryData.pointerData);
-			data = SIGetNextDataEntry(segP, data->next);
-		}
-		/* SIDelExpiredDataEntries(segP); */
-	}
-	else
-	{
-		/* backend must not read messages, its own state has to be reset	 */
-		elog(NOTICE, "SIReadEntryData: cache state reset");
-		resetFunction();		/* XXXX call it here, parameters? */
-
 		/* new valid state--mark all messages "read" */
 		segP->procState[backendId - 1].resetState = false;
 		segP->procState[backendId - 1].limit = SIGetNumEntries(segP);
+		return -1;
 	}
-	/* check whether we can remove dead messages							*/
-	if (i > MAXNUMMESSAGES)
-		elog(FATAL, "SIReadEntryData: Invalid segment state");
+
+	/* Get next message for this backend, if any */
+
+	/* This is fairly inefficient if there are many messages,
+	 * but normally there should not be...
+	 */
+	msg = SIGetNthDataEntry(segP,
+							SIGetProcStateLimit(segP, backendId - 1) + 1);
+
+	if (msg == NULL)
+		return 0;				/* nothing to read */
+
+	*data = msg->entryData;		/* return contents of message */
+
+	segP->procState[backendId - 1].limit++;		/* one more message read */
+
+	/* There may be other backends that haven't read the message,
+	 * so we cannot delete it here.
+	 * SIDelExpiredDataEntries() should be called to remove dead messages.
+	 */
+	return 1;					/* got a message */
 }
 
 /************************************************************************/
@@ -688,15 +686,12 @@ SIDelExpiredDataEntries(SISeg *segP)
 				min = h;
 		}
 	}
-	if (min != 9999999)
+	if (min < 9999999 && min > 0)
 	{
 		/* we can remove min messages */
-		for (i = 1; i <= min; i++)
-		{
-			/* this  adjusts also the state limits! */
-			if (!SIDelDataEntry(segP))
-				elog(FATAL, "SIDelExpiredDataEntries: Invalid segment state");
-		}
+		/* this adjusts also the state limits! */
+		if (!SIDelDataEntries(segP, min))
+			elog(FATAL, "SIDelExpiredDataEntries: Invalid segment state");
 	}
 }
 
@@ -784,8 +779,7 @@ SISegmentAttach(IpcMemoryId shmid)
 	if (shmInvalBuffer == IpcMemAttachFailed)
 	{
 		/* XXX use validity function */
-		elog(NOTICE, "SISegmentAttach: Could not attach segment");
-		elog(FATAL, "SISegmentAttach: %m");
+		elog(FATAL, "SISegmentAttach: Could not attach segment: %m");
 	}
 }
 
diff --git a/src/include/storage/sinvaladt.h b/src/include/storage/sinvaladt.h
index 4885b7380c8..e008e52d30f 100644
--- a/src/include/storage/sinvaladt.h
+++ b/src/include/storage/sinvaladt.h
@@ -6,7 +6,7 @@
  *
  * Copyright (c) 1994, Regents of the University of California
  *
- * $Id: sinvaladt.h,v 1.16 1999/07/16 17:07:38 momjian Exp $
+ * $Id: sinvaladt.h,v 1.17 1999/09/04 18:36:44 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -128,9 +128,9 @@ extern int	SISegmentInit(bool killExistingSegment, IPCKey key,
 
 extern bool SISetDataEntry(SISeg *segP, SharedInvalidData *data);
 extern void SISetProcStateInvalid(SISeg *segP);
-extern bool SIDelDataEntry(SISeg *segP);
-extern void SIReadEntryData(SISeg *segP, int backendId,
-				void (*invalFunction) (), void (*resetFunction) ());
+extern int	SIGetDataEntry(SISeg *segP, int backendId,
+						   SharedInvalidData *data);
+extern bool SIDelDataEntries(SISeg *segP, int n);
 extern void SIDelExpiredDataEntries(SISeg *segP);
 
 #endif	 /* SINVALADT_H */