Clean up WAL/buffer interactions as per my recent proposal. Get rid of the

misleadingly-named WriteBuffer routine, and instead require routines that
change buffer pages to call MarkBufferDirty (which does exactly what it says).
We also require that they do so before calling XLogInsert; this takes care of
the synchronization requirement documented in SyncOneBuffer.  Note that
because bufmgr takes the buffer content lock (in shared mode) while writing
out any buffer, it doesn't matter whether MarkBufferDirty is executed before
the buffer content change is complete, so long as the content change is
completed before releasing exclusive lock on the buffer.  So it's OK to set
the dirtybit before we fill in the LSN.
This eliminates the former kluge of needing to set the dirtybit in LockBuffer.
Aside from making the code more transparent, we can also add some new
debugging assertions, in particular that the caller of MarkBufferDirty must
hold the buffer content lock, not merely a pin.

3 files changed, 71 insertions, 82 deletions

diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index d997db37efb..93ec60481f8 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/gist/gist.c,v 1.130 2006/03/30 23:03:09 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/gist/gist.c,v 1.131 2006/03/31 23:32:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -111,6 +111,9 @@ gistbuild(PG_FUNCTION_ARGS)
 	START_CRIT_SECTION();
 
 	GISTInitBuffer(buffer, F_LEAF);
+
+	MarkBufferDirty(buffer);
+
 	if (!index->rd_istemp)
 	{
 		XLogRecPtr	recptr;
@@ -127,8 +130,8 @@ gistbuild(PG_FUNCTION_ARGS)
 	}
 	else
 		PageSetLSN(page, XLogRecPtrForTemp);
-	LockBuffer(buffer, GIST_UNLOCK);
-	WriteBuffer(buffer);
+
+	UnlockReleaseBuffer(buffer);
 
 	END_CRIT_SECTION();
 
@@ -345,6 +348,15 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 		itvec = gistjoinvector(itvec, &tlen, state->itup, state->ituplen);
 		newitup = gistSplit(state->r, state->stack->buffer, itvec, &tlen, &dist, giststate);
 
+		/*
+		 * must mark buffers dirty before XLogInsert, even though we'll
+		 * still be changing their opaque fields below
+		 */
+		for (ptr = dist; ptr; ptr = ptr->next)
+		{
+			MarkBufferDirty(ptr->buffer);
+		}
+
 		if (!state->r->rd_istemp)
 		{
 			XLogRecPtr	recptr;
@@ -354,21 +366,17 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 								   is_leaf, &(state->key), dist);
 
 			recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);
-			ptr = dist;
-			while (ptr)
+			for (ptr = dist; ptr; ptr = ptr->next)
 			{
 				PageSetLSN(BufferGetPage(ptr->buffer), recptr);
 				PageSetTLI(BufferGetPage(ptr->buffer), ThisTimeLineID);
-				ptr = ptr->next;
 			}
 		}
 		else
 		{
-			ptr = dist;
-			while (ptr)
+			for (ptr = dist; ptr; ptr = ptr->next)
 			{
 				PageSetLSN(BufferGetPage(ptr->buffer), XLogRecPtrForTemp);
-				ptr = ptr->next;
 			}
 		}
 
@@ -379,17 +387,14 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 		{
 			gistnewroot(state->r, state->stack->buffer, state->itup, state->ituplen, &(state->key));
 			state->needInsertComplete = false;
-			ptr = dist;
-			while (ptr)
+			for (ptr = dist; ptr; ptr = ptr->next)
 			{
 				Page		page = (Page) BufferGetPage(ptr->buffer);
 
 				GistPageGetOpaque(page)->rightlink = (ptr->next) ?
 					ptr->next->block.blkno : InvalidBlockNumber;
 				GistPageGetOpaque(page)->nsn = PageGetLSN(page);
-				LockBuffer(ptr->buffer, GIST_UNLOCK);
-				WriteBuffer(ptr->buffer);
-				ptr = ptr->next;
+				UnlockReleaseBuffer(ptr->buffer);
 			}
 		}
 		else
@@ -430,11 +435,9 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 			opaque->rightlink = ourpage->next->block.blkno;
 
 			/*
-			 * fills and write all new pages. They isn't linked into tree yet
+			 * fill and release all new pages. They isn't linked into tree yet
 			 */
-
-			ptr = ourpage->next;
-			while (ptr)
+			for (ptr = ourpage->next; ptr; ptr = ptr->next)
 			{
 				page = (Page) BufferGetPage(ptr->buffer);
 				GistPageGetOpaque(page)->rightlink = (ptr->next) ?
@@ -443,12 +446,8 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 				GistPageGetOpaque(page)->nsn = (ptr->next) ?
 					opaque->nsn : oldnsn;
 
-				LockBuffer(ptr->buffer, GIST_UNLOCK);
-				WriteBuffer(ptr->buffer);
-				ptr = ptr->next;
+				UnlockReleaseBuffer(ptr->buffer);
 			}
-
-			WriteNoReleaseBuffer(state->stack->buffer);
 		}
 
 		END_CRIT_SECTION();
@@ -460,6 +459,8 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 
 		gistfillbuffer(state->r, state->stack->page, state->itup, state->ituplen, InvalidOffsetNumber);
 
+		MarkBufferDirty(state->stack->buffer);
+
 		oldlsn = PageGetLSN(state->stack->page);
 		if (!state->r->rd_istemp)
 		{
@@ -489,7 +490,6 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 
 		if (state->stack->blkno == GIST_ROOT_BLKNO)
 			state->needInsertComplete = false;
-		WriteNoReleaseBuffer(state->stack->buffer);
 
 		END_CRIT_SECTION();
 
@@ -561,8 +561,7 @@ gistfindleaf(GISTInsertState *state, GISTSTATE *giststate)
 			 * caused split non-root page is detected, go up to parent to
 			 * choose best child
 			 */
-			LockBuffer(state->stack->buffer, GIST_UNLOCK);
-			ReleaseBuffer(state->stack->buffer);
+			UnlockReleaseBuffer(state->stack->buffer);
 			state->stack = state->stack->parent;
 			continue;
 		}
@@ -630,8 +629,7 @@ gistfindleaf(GISTInsertState *state, GISTSTATE *giststate)
 				 */
 
 				/* forget buffer */
-				LockBuffer(state->stack->buffer, GIST_UNLOCK);
-				ReleaseBuffer(state->stack->buffer);
+				UnlockReleaseBuffer(state->stack->buffer);
 
 				state->stack = state->stack->parent;
 				continue;
@@ -681,8 +679,7 @@ gistFindPath(Relation r, BlockNumber child)
 		if (GistPageIsLeaf(page))
 		{
 			/* we can safety go away, follows only leaf pages */
-			LockBuffer(buffer, GIST_UNLOCK);
-			ReleaseBuffer(buffer);
+			UnlockReleaseBuffer(buffer);
 			return NULL;
 		}
 
@@ -735,8 +732,7 @@ gistFindPath(Relation r, BlockNumber child)
 					ptr = ptr->parent;
 				}
 				top->childoffnum = i;
-				LockBuffer(buffer, GIST_UNLOCK);
-				ReleaseBuffer(buffer);
+				UnlockReleaseBuffer(buffer);
 				return top;
 			}
 			else
@@ -753,8 +749,7 @@ gistFindPath(Relation r, BlockNumber child)
 			}
 		}
 
-		LockBuffer(buffer, GIST_UNLOCK);
-		ReleaseBuffer(buffer);
+		UnlockReleaseBuffer(buffer);
 		top = top->next;
 	}
 
@@ -801,8 +796,7 @@ gistFindCorrectParent(Relation r, GISTInsertStack *child)
 			}
 
 			parent->blkno = GistPageGetOpaque(parent->page)->rightlink;
-			LockBuffer(parent->buffer, GIST_UNLOCK);
-			ReleaseBuffer(parent->buffer);
+			UnlockReleaseBuffer(parent->buffer);
 			if (parent->blkno == InvalidBlockNumber)
 
 				/*
@@ -881,8 +875,7 @@ gistmakedeal(GISTInsertState *state, GISTSTATE *giststate)
 		is_splitted = gistplacetopage(state, giststate);
 
 		/* parent locked above, so release child buffer */
-		LockBuffer(state->stack->buffer, GIST_UNLOCK);
-		ReleaseBuffer(state->stack->buffer);
+		UnlockReleaseBuffer(state->stack->buffer);
 
 		/* pop parent page from stack */
 		state->stack = state->stack->parent;
@@ -1182,6 +1175,9 @@ gistSplit(Relation r,
 	return newtup;
 }
 
+/*
+ * buffer must be pinned and locked by caller
+ */
 void
 gistnewroot(Relation r, Buffer buffer, IndexTuple *itup, int len, ItemPointer key)
 {
@@ -1192,9 +1188,11 @@ gistnewroot(Relation r, Buffer buffer, IndexTuple *itup, int len, ItemPointer ke
 
 	START_CRIT_SECTION();
 
-	GISTInitBuffer(buffer, 0);	/* XXX not F_LEAF? */
+	GISTInitBuffer(buffer, 0);
 	gistfillbuffer(r, page, itup, len, FirstOffsetNumber);
 
+	MarkBufferDirty(buffer);
+
 	if (!r->rd_istemp)
 	{
 		XLogRecPtr	recptr;
@@ -1211,8 +1209,6 @@ gistnewroot(Relation r, Buffer buffer, IndexTuple *itup, int len, ItemPointer ke
 	else
 		PageSetLSN(page, XLogRecPtrForTemp);
 
-	WriteNoReleaseBuffer(buffer);
-
 	END_CRIT_SECTION();
 }
 
diff --git a/src/backend/access/gist/gistvacuum.c b/src/backend/access/gist/gistvacuum.c
index e7925c2c151..285f1301321 100644
--- a/src/backend/access/gist/gistvacuum.c
+++ b/src/backend/access/gist/gistvacuum.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/gist/gistvacuum.c,v 1.17 2006/03/30 23:03:10 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/gist/gistvacuum.c,v 1.18 2006/03/31 23:32:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -71,11 +71,7 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 	vacuum_delay_point();
 
 	buffer = ReadBuffer(gv->index, blkno);
-
-	/*
-	 * This is only used during VACUUM FULL, so we need not bother to lock
-	 * individual index pages
-	 */
+	LockBuffer(buffer, GIST_EXCLUSIVE);
 	gistcheckpage(gv->index, buffer);
 	page = (Page) BufferGetPage(buffer);
 	maxoff = PageGetMaxOffsetNumber(page);
@@ -183,6 +179,11 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 				}
 				res.itup = vec;
 
+				for (ptr = dist; ptr; ptr = ptr->next)
+				{
+					MarkBufferDirty(ptr->buffer);
+				}
+
 				if (!gv->index->rd_istemp)
 				{
 					XLogRecPtr	recptr;
@@ -198,12 +199,10 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 					xlinfo = rdata->data;
 
 					recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);
-					ptr = dist;
-					while (ptr)
+					for (ptr = dist; ptr; ptr = ptr->next)
 					{
 						PageSetLSN(BufferGetPage(ptr->buffer), recptr);
 						PageSetTLI(BufferGetPage(ptr->buffer), ThisTimeLineID);
-						ptr = ptr->next;
 					}
 
 					pfree(xlinfo);
@@ -211,21 +210,18 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 				}
 				else
 				{
-					ptr = dist;
-					while (ptr)
+					for (ptr = dist; ptr; ptr = ptr->next)
 					{
 						PageSetLSN(BufferGetPage(ptr->buffer), XLogRecPtrForTemp);
-						ptr = ptr->next;
 					}
 				}
 
-				ptr = dist;
-				while (ptr)
+				for (ptr = dist; ptr; ptr = ptr->next)
 				{
+					/* we must keep the buffer lock on the head page */
 					if (BufferGetBlockNumber(ptr->buffer) != blkno)
 						LockBuffer(ptr->buffer, GIST_UNLOCK);
-					WriteBuffer(ptr->buffer);
-					ptr = ptr->next;
+					ReleaseBuffer(ptr->buffer);
 				}
 
 				if (blkno == GIST_ROOT_BLKNO)
@@ -297,6 +293,8 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 
 	if (needwrite)
 	{
+		MarkBufferDirty(buffer);
+
 		if (!gv->index->rd_istemp)
 		{
 			XLogRecData *rdata;
@@ -317,13 +315,12 @@ gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
 		}
 		else
 			PageSetLSN(page, XLogRecPtrForTemp);
-		WriteBuffer(buffer);
 	}
-	else
-		ReleaseBuffer(buffer);
 
 	END_CRIT_SECTION();
 
+	UnlockReleaseBuffer(buffer);
+
 	if (ncompleted && !gv->index->rd_istemp)
 		gistxlogInsertCompletion(gv->index->rd_node, completed, ncompleted);
 
@@ -429,8 +426,7 @@ gistvacuumcleanup(PG_FUNCTION_ARGS)
 		}
 		else
 			lastFilledBlock = blkno;
-		LockBuffer(buffer, GIST_UNLOCK);
-		ReleaseBuffer(buffer);
+		UnlockReleaseBuffer(buffer);
 	}
 	lastBlock = npages - 1;
 
@@ -569,8 +565,7 @@ gistbulkdelete(PG_FUNCTION_ARGS)
 			if (stack->blkno == GIST_ROOT_BLKNO && !GistPageIsLeaf(page))
 			{
 				/* only the root can become non-leaf during relock */
-				LockBuffer(buffer, GIST_UNLOCK);
-				ReleaseBuffer(buffer);
+				UnlockReleaseBuffer(buffer);
 				/* one more check */
 				continue;
 			}
@@ -617,6 +612,8 @@ gistbulkdelete(PG_FUNCTION_ARGS)
 			{
 				GistMarkTuplesDeleted(page);
 
+				MarkBufferDirty(buffer);
+
 				if (!rel->rd_istemp)
 				{
 					XLogRecData *rdata;
@@ -638,7 +635,6 @@ gistbulkdelete(PG_FUNCTION_ARGS)
 				}
 				else
 					PageSetLSN(page, XLogRecPtrForTemp);
-				WriteNoReleaseBuffer(buffer);
 			}
 
 			END_CRIT_SECTION();
@@ -666,8 +662,7 @@ gistbulkdelete(PG_FUNCTION_ARGS)
 			}
 		}
 
-		LockBuffer(buffer, GIST_UNLOCK);
-		ReleaseBuffer(buffer);
+		UnlockReleaseBuffer(buffer);
 
 		ptr = stack->next;
 		pfree(stack);
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 12a521c75c9..fbceae29a96 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *			 $PostgreSQL: pgsql/src/backend/access/gist/gistxlog.c,v 1.13 2006/03/30 23:03:10 tgl Exp $
+ *			 $PostgreSQL: pgsql/src/backend/access/gist/gistxlog.c,v 1.14 2006/03/31 23:32:05 tgl Exp $
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
@@ -192,8 +192,7 @@ gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record, bool isnewroot)
 
 	if (XLByteLE(lsn, PageGetLSN(page)))
 	{
-		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-		ReleaseBuffer(buffer);
+		UnlockReleaseBuffer(buffer);
 		return;
 	}
 
@@ -236,8 +235,8 @@ gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record, bool isnewroot)
 	GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
 	PageSetLSN(page, lsn);
 	PageSetTLI(page, ThisTimeLineID);
-	LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-	WriteBuffer(buffer);
+	MarkBufferDirty(buffer);
+	UnlockReleaseBuffer(buffer);
 
 	if (ItemPointerIsValid(&(xlrec.data->key)))
 	{
@@ -313,8 +312,8 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
 
 		PageSetLSN(page, lsn);
 		PageSetTLI(page, ThisTimeLineID);
-		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-		WriteBuffer(buffer);
+		MarkBufferDirty(buffer);
+		UnlockReleaseBuffer(buffer);
 	}
 
 	if (ItemPointerIsValid(&(xlrec.data->key)))
@@ -346,8 +345,8 @@ gistRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
 	PageSetLSN(page, lsn);
 	PageSetTLI(page, ThisTimeLineID);
 
-	LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-	WriteBuffer(buffer);
+	MarkBufferDirty(buffer);
+	UnlockReleaseBuffer(buffer);
 }
 
 static void
@@ -561,8 +560,8 @@ gistContinueInsert(gistIncompleteInsert *insert)
 		PageSetLSN(page, insert->lsn);
 		PageSetTLI(page, ThisTimeLineID);
 
-		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-		WriteBuffer(buffer);
+		MarkBufferDirty(buffer);
+		UnlockReleaseBuffer(buffer);
 
 		/*
 		 * XXX fall out to avoid making LOG message at bottom of routine.
@@ -598,8 +597,7 @@ gistContinueInsert(gistIncompleteInsert *insert)
 
 			if (XLByteLE(insert->lsn, PageGetLSN(pages[numbuffer - 1])))
 			{
-				LockBuffer(buffers[numbuffer - 1], BUFFER_LOCK_UNLOCK);
-				ReleaseBuffer(buffers[numbuffer - 1]);
+				UnlockReleaseBuffer(buffers[numbuffer - 1]);
 				return;
 			}
 
@@ -685,8 +683,8 @@ gistContinueInsert(gistIncompleteInsert *insert)
 				PageSetLSN(pages[j], insert->lsn);
 				PageSetTLI(pages[j], ThisTimeLineID);
 				GistPageGetOpaque(pages[j])->rightlink = InvalidBlockNumber;
-				LockBuffer(buffers[j], BUFFER_LOCK_UNLOCK);
-				WriteBuffer(buffers[j]);
+				MarkBufferDirty(buffers[j]);
+				UnlockReleaseBuffer(buffers[j]);
 			}
 		}
 	}