Clean up lockmanager data structures some more, in preparation for planned

rewrite of deadlock checking.  Lock holder objects are now reachable from
the associated LOCK as well as from the owning PROC.  This makes it
practical to find all the processes holding a lock, as well as all those
waiting on the lock.  Also, clean up some of the grottier aspects of the
SHMQueue API, and cause the waitProcs list to be stored in the intuitive
direction instead of the nonintuitive one.  (Bet you didn't know that
the code followed the 'prev' link to get to the next waiting process,
instead of the 'next' link.  It doesn't do that anymore.)

1 files changed, 111 insertions, 109 deletions

diff --git a/src/backend/storage/ipc/shmqueue.c b/src/backend/storage/ipc/shmqueue.c
index 2cdccd0a550..ae6950ca6f0 100644
--- a/src/backend/storage/ipc/shmqueue.c
+++ b/src/backend/storage/ipc/shmqueue.c
@@ -8,32 +8,34 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/shmqueue.c,v 1.13 2000/01/26 05:56:58 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/shmqueue.c,v 1.14 2001/01/22 22:30:06 tgl Exp $
  *
  * NOTES
  *
  * Package for managing doubly-linked lists in shared memory.
  * The only tricky thing is that SHM_QUEUE will usually be a field
- * in a larger record.	SHMQueueGetFirst has to return a pointer
+ * in a larger record.	SHMQueueNext has to return a pointer
  * to the record itself instead of a pointer to the SHMQueue field
- * of the record.  It takes an extra pointer and does some extra
+ * of the record.  It takes an extra parameter and does some extra
  * pointer arithmetic to do this correctly.
  *
  * NOTE: These are set up so they can be turned into macros some day.
  *
  *-------------------------------------------------------------------------
  */
-
 #include "postgres.h"
+
 #include "storage/shmem.h"
 
 /*#define SHMQUEUE_DEBUG*/
 #ifdef SHMQUEUE_DEBUG
-#define SHMQUEUE_DEBUG_DEL		/* deletions */
-#define SHMQUEUE_DEBUG_HD		/* head inserts */
-#define SHMQUEUE_DEBUG_TL		/* tail inserts */
+
 #define SHMQUEUE_DEBUG_ELOG NOTICE
-#endif	 /* SHMQUEUE_DEBUG */
+
+static void dumpQ(SHM_QUEUE *q, char *s);
+
+#endif
+
 
 /*
  * ShmemQueueInit -- make the head of a new queue point
@@ -84,76 +86,23 @@ SHMQueueDelete(SHM_QUEUE *queue)
 	Assert(SHM_PTR_VALID(nextElem));
 	Assert(SHM_PTR_VALID(prevElem));
 
-#ifdef SHMQUEUE_DEBUG_DEL
+#ifdef SHMQUEUE_DEBUG
 	dumpQ(queue, "in SHMQueueDelete: begin");
-#endif	 /* SHMQUEUE_DEBUG_DEL */
+#endif
 
 	prevElem->next = (queue)->next;
 	nextElem->prev = (queue)->prev;
 
-#ifdef SHMQUEUE_DEBUG_DEL
-	dumpQ((SHM_QUEUE *) MAKE_PTR(queue->prev), "in SHMQueueDelete: end");
-#endif	 /* SHMQUEUE_DEBUG_DEL */
-}
-
-#ifdef SHMQUEUE_DEBUG
-void
-dumpQ(SHM_QUEUE *q, char *s)
-{
-	char		elem[NAMEDATALEN];
-	char		buf[1024];
-	SHM_QUEUE  *start = q;
-	int			count = 0;
-
-	sprintf(buf, "q prevs: %x", MAKE_OFFSET(q));
-	q = (SHM_QUEUE *) MAKE_PTR(q->prev);
-	while (q != start)
-	{
-		sprintf(elem, "--->%x", MAKE_OFFSET(q));
-		strcat(buf, elem);
-		q = (SHM_QUEUE *) MAKE_PTR(q->prev);
-		if (q->prev == MAKE_OFFSET(q))
-			break;
-		if (count++ > 40)
-		{
-			strcat(buf, "BAD PREV QUEUE!!");
-			break;
-		}
-	}
-	sprintf(elem, "--->%x", MAKE_OFFSET(q));
-	strcat(buf, elem);
-	elog(SHMQUEUE_DEBUG_ELOG, "%s: %s", s, buf);
-
-	sprintf(buf, "q nexts: %x", MAKE_OFFSET(q));
-	count = 0;
-	q = (SHM_QUEUE *) MAKE_PTR(q->next);
-	while (q != start)
-	{
-		sprintf(elem, "--->%x", MAKE_OFFSET(q));
-		strcat(buf, elem);
-		q = (SHM_QUEUE *) MAKE_PTR(q->next);
-		if (q->next == MAKE_OFFSET(q))
-			break;
-		if (count++ > 10)
-		{
-			strcat(buf, "BAD NEXT QUEUE!!");
-			break;
-		}
-	}
-	sprintf(elem, "--->%x", MAKE_OFFSET(q));
-	strcat(buf, elem);
-	elog(SHMQUEUE_DEBUG_ELOG, "%s: %s", s, buf);
+	(queue)->prev = (queue)->next = INVALID_OFFSET;
 }
 
-#endif	 /* SHMQUEUE_DEBUG */
-
 /*
- * SHMQueueInsertHD -- put elem in queue between the queue head
- *		and its "prev" element.
+ * SHMQueueInsertBefore -- put elem in queue before the given queue
+ *		element.  Inserting "before" the queue head puts the elem
+ *		at the tail of the queue.
  */
-#ifdef NOT_USED
 void
-SHMQueueInsertHD(SHM_QUEUE *queue, SHM_QUEUE *elem)
+SHMQueueInsertBefore(SHM_QUEUE *queue, SHM_QUEUE *elem)
 {
 	SHM_QUEUE  *prevPtr = (SHM_QUEUE *) MAKE_PTR((queue)->prev);
 	SHMEM_OFFSET elemOffset = MAKE_OFFSET(elem);
@@ -161,24 +110,28 @@ SHMQueueInsertHD(SHM_QUEUE *queue, SHM_QUEUE *elem)
 	Assert(SHM_PTR_VALID(queue));
 	Assert(SHM_PTR_VALID(elem));
 
-#ifdef SHMQUEUE_DEBUG_HD
-	dumpQ(queue, "in SHMQueueInsertHD: begin");
-#endif	 /* SHMQUEUE_DEBUG_HD */
+#ifdef SHMQUEUE_DEBUG
+	dumpQ(queue, "in SHMQueueInsertBefore: begin");
+#endif
 
 	(elem)->next = prevPtr->next;
 	(elem)->prev = queue->prev;
 	(queue)->prev = elemOffset;
 	prevPtr->next = elemOffset;
 
-#ifdef SHMQUEUE_DEBUG_HD
-	dumpQ(queue, "in SHMQueueInsertHD: end");
-#endif	 /* SHMQUEUE_DEBUG_HD */
-}
-
+#ifdef SHMQUEUE_DEBUG
+	dumpQ(queue, "in SHMQueueInsertBefore: end");
 #endif
+}
 
+/*
+ * SHMQueueInsertAfter -- put elem in queue after the given queue
+ *		element.  Inserting "after" the queue head puts the elem
+ *		at the head of the queue.
+ */
+#ifdef NOT_USED
 void
-SHMQueueInsertTL(SHM_QUEUE *queue, SHM_QUEUE *elem)
+SHMQueueInsertAfter(SHM_QUEUE *queue, SHM_QUEUE *elem)
 {
 	SHM_QUEUE  *nextPtr = (SHM_QUEUE *) MAKE_PTR((queue)->next);
 	SHMEM_OFFSET elemOffset = MAKE_OFFSET(elem);
@@ -186,58 +139,55 @@ SHMQueueInsertTL(SHM_QUEUE *queue, SHM_QUEUE *elem)
 	Assert(SHM_PTR_VALID(queue));
 	Assert(SHM_PTR_VALID(elem));
 
-#ifdef SHMQUEUE_DEBUG_TL
-	dumpQ(queue, "in SHMQueueInsertTL: begin");
-#endif	 /* SHMQUEUE_DEBUG_TL */
+#ifdef SHMQUEUE_DEBUG
+	dumpQ(queue, "in SHMQueueInsertAfter: begin");
+#endif
 
 	(elem)->prev = nextPtr->prev;
 	(elem)->next = queue->next;
 	(queue)->next = elemOffset;
 	nextPtr->prev = elemOffset;
 
-#ifdef SHMQUEUE_DEBUG_TL
-	dumpQ(queue, "in SHMQueueInsertTL: end");
-#endif	 /* SHMQUEUE_DEBUG_TL */
+#ifdef SHMQUEUE_DEBUG
+	dumpQ(queue, "in SHMQueueInsertAfter: end");
+#endif
 }
+#endif /* NOT_USED */
 
-/*
- * SHMQueueFirst -- Get the first element from a queue
+/*--------------------
+ * SHMQueueNext -- Get the next element from a queue
  *
- * First element is queue->next.  If SHMQueue is part of
+ * To start the iteration, pass the queue head as both queue and curElem.
+ * Returns NULL if no more elements.
+ *
+ * Next element is at curElem->next.  If SHMQueue is part of
  * a larger structure, we want to return a pointer to the
  * whole structure rather than a pointer to its SHMQueue field.
  * I.E. struct {
  *		int				stuff;
  *		SHMQueue		elem;
  * } ELEMType;
- * when this element is in a queue (queue->next) is struct.elem.
- * nextQueue allows us to calculate the offset of the SHMQueue
- * field in the structure.
- *
- * call to SHMQueueFirst should take these parameters:
+ * When this element is in a queue, (prevElem->next) is struct.elem.
+ * We subtract linkOffset to get the correct start address of the structure.
  *
- *	 &(queueHead),&firstElem,&(firstElem->next)
+ * calls to SHMQueueNext should take these parameters:
  *
- * Note that firstElem may well be uninitialized.  if firstElem
- * is initially K, &(firstElem->next) will be K+ the offset to
- * next.
+ *	 &(queueHead), &(queueHead), offsetof(ELEMType, elem)
+ * or
+ *	 &(queueHead), &(curElem->elem), offsetof(ELEMType, elem)
+ *--------------------
  */
-void
-SHMQueueFirst(SHM_QUEUE *queue, Pointer *nextPtrPtr, SHM_QUEUE *nextQueue)
+Pointer
+SHMQueueNext(SHM_QUEUE *queue, SHM_QUEUE *curElem, Size linkOffset)
 {
-	SHM_QUEUE  *elemPtr = (SHM_QUEUE *) MAKE_PTR((queue)->next);
+	SHM_QUEUE  *elemPtr = (SHM_QUEUE *) MAKE_PTR((curElem)->next);
 
-	Assert(SHM_PTR_VALID(queue));
-	*nextPtrPtr = (Pointer) (((unsigned long) *nextPtrPtr) +
-				((unsigned long) elemPtr) - ((unsigned long) nextQueue));
-
-	/*
-	 * nextPtrPtr a ptr to a structure linked in the queue nextQueue is
-	 * the SHMQueue field of the structure nextPtrPtr - nextQueue is 0
-	 * minus the offset of the queue field n the record elemPtr +
-	 * (*nextPtrPtr - nexQueue) is the start of the structure containing
-	 * elemPtr.
-	 */
+	Assert(SHM_PTR_VALID(curElem));
+
+	if (elemPtr == queue)		/* back to the queue head? */
+		return NULL;
+
+	return (Pointer) (((char *) elemPtr) - linkOffset);
 }
 
 /*
@@ -255,3 +205,55 @@ SHMQueueEmpty(SHM_QUEUE *queue)
 	}
 	return FALSE;
 }
+
+#ifdef SHMQUEUE_DEBUG
+
+static void
+dumpQ(SHM_QUEUE *q, char *s)
+{
+	char		elem[NAMEDATALEN];
+	char		buf[1024];
+	SHM_QUEUE  *start = q;
+	int			count = 0;
+
+	sprintf(buf, "q prevs: %lx", MAKE_OFFSET(q));
+	q = (SHM_QUEUE *) MAKE_PTR(q->prev);
+	while (q != start)
+	{
+		sprintf(elem, "--->%lx", MAKE_OFFSET(q));
+		strcat(buf, elem);
+		q = (SHM_QUEUE *) MAKE_PTR(q->prev);
+		if (q->prev == MAKE_OFFSET(q))
+			break;
+		if (count++ > 40)
+		{
+			strcat(buf, "BAD PREV QUEUE!!");
+			break;
+		}
+	}
+	sprintf(elem, "--->%lx", MAKE_OFFSET(q));
+	strcat(buf, elem);
+	elog(SHMQUEUE_DEBUG_ELOG, "%s: %s", s, buf);
+
+	sprintf(buf, "q nexts: %lx", MAKE_OFFSET(q));
+	count = 0;
+	q = (SHM_QUEUE *) MAKE_PTR(q->next);
+	while (q != start)
+	{
+		sprintf(elem, "--->%lx", MAKE_OFFSET(q));
+		strcat(buf, elem);
+		q = (SHM_QUEUE *) MAKE_PTR(q->next);
+		if (q->next == MAKE_OFFSET(q))
+			break;
+		if (count++ > 10)
+		{
+			strcat(buf, "BAD NEXT QUEUE!!");
+			break;
+		}
+	}
+	sprintf(elem, "--->%lx", MAKE_OFFSET(q));
+	strcat(buf, elem);
+	elog(SHMQUEUE_DEBUG_ELOG, "%s: %s", s, buf);
+}
+
+#endif	 /* SHMQUEUE_DEBUG */