slru-race-1.patch

*** src/backend/access/transam/clog.c.orig	Fri Oct 14 22:59:37 2005
--- src/backend/access/transam/clog.c	Mon Oct 31 10:08:24 2005
***************
*** 107,113 ****
  	byteval |= (status << bshift);
  	*byteptr = byteval;
  
! 	ClogCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  
  	LWLockRelease(CLogControlLock);
  }
--- 107,113 ----
  	byteval |= (status << bshift);
  	*byteptr = byteval;
  
! 	ClogCtl->shared->page_dirty[slotno] = true;
  
  	LWLockRelease(CLogControlLock);
  }
***************
*** 175,181 ****
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(ClogCtl, slotno, NULL);
! 	Assert(ClogCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  	LWLockRelease(CLogControlLock);
  }
--- 175,181 ----
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(ClogCtl, slotno, NULL);
! 	Assert(!ClogCtl->shared->page_dirty[slotno]);
  
  	LWLockRelease(CLogControlLock);
  }
***************
*** 246,252 ****
  		/* Zero the rest of the page */
  		MemSet(byteptr + 1, 0, BLCKSZ - byteno - 1);
  
! 		ClogCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  	}
  
  	LWLockRelease(CLogControlLock);
--- 246,252 ----
  		/* Zero the rest of the page */
  		MemSet(byteptr + 1, 0, BLCKSZ - byteno - 1);
  
! 		ClogCtl->shared->page_dirty[slotno] = true;
  	}
  
  	LWLockRelease(CLogControlLock);
***************
*** 404,410 ****
  
  		slotno = ZeroCLOGPage(pageno, false);
  		SimpleLruWritePage(ClogCtl, slotno, NULL);
! 		Assert(ClogCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  		LWLockRelease(CLogControlLock);
  	}
--- 404,410 ----
  
  		slotno = ZeroCLOGPage(pageno, false);
  		SimpleLruWritePage(ClogCtl, slotno, NULL);
! 		Assert(!ClogCtl->shared->page_dirty[slotno]);
  
  		LWLockRelease(CLogControlLock);
  	}
*** src/backend/access/transam/multixact.c.orig	Fri Oct 28 15:00:19 2005
--- src/backend/access/transam/multixact.c	Mon Oct 31 10:08:24 2005
***************
*** 714,720 ****
  
  	*offptr = offset;
  
! 	MultiXactOffsetCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  
  	/* Exchange our lock */
  	LWLockRelease(MultiXactOffsetControlLock);
--- 714,720 ----
  
  	*offptr = offset;
  
! 	MultiXactOffsetCtl->shared->page_dirty[slotno] = true;
  
  	/* Exchange our lock */
  	LWLockRelease(MultiXactOffsetControlLock);
***************
*** 742,748 ****
  
  		*memberptr = xids[i];
  
! 		MultiXactMemberCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
--- 742,748 ----
  
  		*memberptr = xids[i];
  
! 		MultiXactMemberCtl->shared->page_dirty[slotno] = true;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
***************
*** 1308,1314 ****
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(MultiXactOffsetCtl, slotno, NULL);
! 	Assert(MultiXactOffsetCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  	LWLockRelease(MultiXactOffsetControlLock);
  
--- 1308,1314 ----
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(MultiXactOffsetCtl, slotno, NULL);
! 	Assert(!MultiXactOffsetCtl->shared->page_dirty[slotno]);
  
  	LWLockRelease(MultiXactOffsetControlLock);
  
***************
*** 1319,1325 ****
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(MultiXactMemberCtl, slotno, NULL);
! 	Assert(MultiXactMemberCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  	LWLockRelease(MultiXactMemberControlLock);
  }
--- 1319,1325 ----
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(MultiXactMemberCtl, slotno, NULL);
! 	Assert(!MultiXactMemberCtl->shared->page_dirty[slotno]);
  
  	LWLockRelease(MultiXactMemberControlLock);
  }
***************
*** 1405,1411 ****
  
  		MemSet(offptr, 0, BLCKSZ - (entryno * sizeof(MultiXactOffset)));
  
! 		MultiXactOffsetCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  	}
  
  	LWLockRelease(MultiXactOffsetControlLock);
--- 1405,1411 ----
  
  		MemSet(offptr, 0, BLCKSZ - (entryno * sizeof(MultiXactOffset)));
  
! 		MultiXactOffsetCtl->shared->page_dirty[slotno] = true;
  	}
  
  	LWLockRelease(MultiXactOffsetControlLock);
***************
*** 1435,1441 ****
  
  		MemSet(xidptr, 0, BLCKSZ - (entryno * sizeof(TransactionId)));
  
! 		MultiXactMemberCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
--- 1435,1441 ----
  
  		MemSet(xidptr, 0, BLCKSZ - (entryno * sizeof(TransactionId)));
  
! 		MultiXactMemberCtl->shared->page_dirty[slotno] = true;
  	}
  
  	LWLockRelease(MultiXactMemberControlLock);
***************
*** 1829,1835 ****
  
  		slotno = ZeroMultiXactOffsetPage(pageno, false);
  		SimpleLruWritePage(MultiXactOffsetCtl, slotno, NULL);
! 		Assert(MultiXactOffsetCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  		LWLockRelease(MultiXactOffsetControlLock);
  	}
--- 1829,1835 ----
  
  		slotno = ZeroMultiXactOffsetPage(pageno, false);
  		SimpleLruWritePage(MultiXactOffsetCtl, slotno, NULL);
! 		Assert(!MultiXactOffsetCtl->shared->page_dirty[slotno]);
  
  		LWLockRelease(MultiXactOffsetControlLock);
  	}
***************
*** 1844,1850 ****
  
  		slotno = ZeroMultiXactMemberPage(pageno, false);
  		SimpleLruWritePage(MultiXactMemberCtl, slotno, NULL);
! 		Assert(MultiXactMemberCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  		LWLockRelease(MultiXactMemberControlLock);
  	}
--- 1844,1850 ----
  
  		slotno = ZeroMultiXactMemberPage(pageno, false);
  		SimpleLruWritePage(MultiXactMemberCtl, slotno, NULL);
! 		Assert(!MultiXactMemberCtl->shared->page_dirty[slotno]);
  
  		LWLockRelease(MultiXactMemberControlLock);
  	}
*** src/backend/access/transam/slru.c.orig	Sat Oct 29 14:52:52 2005
--- src/backend/access/transam/slru.c	Mon Oct 31 11:05:35 2005
***************
*** 14,48 ****
   * out the latest page (since we know it's going to be hit again eventually).
   *
   * We use a control LWLock to protect the shared data structures, plus
!  * per-buffer LWLocks that synchronize I/O for each buffer.  A process
!  * that is reading in or writing out a page buffer does not hold the control
!  * lock, only the per-buffer lock for the buffer it is working on.
!  *
!  * To change the page number or state of a buffer, one must normally hold
!  * the control lock.  (The sole exception to this rule is that a writer
!  * process changes the state from DIRTY to WRITE_IN_PROGRESS while holding
!  * only the per-buffer lock.)  If the buffer's state is neither EMPTY nor
!  * CLEAN, then there may be processes doing (or waiting to do) I/O on the
!  * buffer, so the page number may not be changed, and the only allowed state
!  * transition is to change WRITE_IN_PROGRESS to DIRTY after dirtying the page.
!  * To do any other state transition involving a buffer with potential I/O
!  * processes, one must hold both the per-buffer lock and the control lock.
!  * (Note the control lock must be acquired second; do not wait on a buffer
!  * lock while holding the control lock.)  A process wishing to read a page
!  * marks the buffer state as READ_IN_PROGRESS, then drops the control lock,
!  * acquires the per-buffer lock, and rechecks the state before proceeding.
!  * This recheck takes care of the possibility that someone else already did
!  * the read, while the early marking prevents someone else from trying to
!  * read the same page into a different buffer.
!  *
!  * Note we are assuming that read and write of the state value is atomic,
!  * since I/O processes may examine and change the state while not holding
!  * the control lock.
   *
   * As with the regular buffer manager, it is possible for another process
   * to re-dirty a page that is currently being written out.	This is handled
!  * by setting the page's state from WRITE_IN_PROGRESS to DIRTY.  The writing
!  * process must notice this and not mark the page CLEAN when it's done.
   *
   *
   * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
--- 14,37 ----
   * out the latest page (since we know it's going to be hit again eventually).
   *
   * We use a control LWLock to protect the shared data structures, plus
!  * per-buffer LWLocks that synchronize I/O for each buffer.  The control lock
!  * must be held to examine or modify any shared state.  A process that is
!  * reading in or writing out a page buffer does not hold the control lock,
!  * only the per-buffer lock for the buffer it is working on.
!  *
!  * When initiating I/O on a buffer, we acquire the per-buffer lock exclusively
!  * before releasing the control lock.  The per-buffer lock is released after
!  * completing the I/O, re-acquiring the control lock, and updating the shared
!  * state.  (Deadlock is not possible here, because we never try to initiate
!  * I/O when someone else is already doing I/O on the same buffer.)
!  * To wait for I/O to complete, release the control lock, acquire the
!  * per-buffer lock in shared mode, immediately release the per-buffer lock,
!  * reacquire the control lock, and then recheck state (since arbitrary things
!  * could have happened while we didn't have the lock).
   *
   * As with the regular buffer manager, it is possible for another process
   * to re-dirty a page that is currently being written out.	This is handled
!  * by re-setting the page's page_dirty flag.
   *
   *
   * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
***************
*** 175,180 ****
--- 164,170 ----
  		{
  			shared->page_buffer[slotno] = bufptr;
  			shared->page_status[slotno] = SLRU_PAGE_EMPTY;
+ 			shared->page_dirty[slotno] = false;
  			shared->page_lru_count[slotno] = 1;
  			shared->buffer_locks[slotno] = LWLockAssign();
  			bufptr += BLCKSZ;
***************
*** 211,222 ****
  	/* Find a suitable buffer slot for the page */
  	slotno = SlruSelectLRUPage(ctl, pageno);
  	Assert(shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 		   shared->page_status[slotno] == SLRU_PAGE_CLEAN ||
  		   shared->page_number[slotno] == pageno);
  
  	/* Mark the slot as containing this page */
  	shared->page_number[slotno] = pageno;
! 	shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  	SlruRecentlyUsed(shared, slotno);
  
  	/* Set the buffer to zeroes */
--- 201,214 ----
  	/* Find a suitable buffer slot for the page */
  	slotno = SlruSelectLRUPage(ctl, pageno);
  	Assert(shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 		   (shared->page_status[slotno] == SLRU_PAGE_VALID &&
! 			!shared->page_dirty[slotno]) ||
  		   shared->page_number[slotno] == pageno);
  
  	/* Mark the slot as containing this page */
  	shared->page_number[slotno] = pageno;
! 	shared->page_status[slotno] = SLRU_PAGE_VALID;
! 	shared->page_dirty[slotno] = true;
  	SlruRecentlyUsed(shared, slotno);
  
  	/* Set the buffer to zeroes */
***************
*** 229,234 ****
--- 221,268 ----
  }
  
  /*
+  * Wait for any active I/O on a page slot to finish.  (This does not
+  * guarantee that new I/O hasn't been started before we return, though.)
+  *
+  * Control lock must be held at entry, and will be held at exit.
+  */
+ static void
+ SimpleLruWaitIO(SlruCtl ctl, int slotno)
+ {
+ 	SlruShared	shared = ctl->shared;
+ 
+ 	/* See notes at top of file */
+ 	LWLockRelease(shared->ControlLock);
+ 	LWLockAcquire(shared->buffer_locks[slotno], LW_SHARED);
+ 	LWLockRelease(shared->buffer_locks[slotno]);
+ 	LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
+ 	/*
+ 	 * If the slot is still in an io-in-progress state, then either someone
+ 	 * already started a new I/O on the slot, or a previous I/O failed and
+ 	 * neglected to reset the page state.  That shouldn't happen, really,
+ 	 * but it seems worth a few extra cycles to check and recover from it.
+ 	 * We can cheaply test for failure by seeing if the buffer lock is still
+ 	 * held (we assume that transaction abort would release the lock).
+ 	 */
+ 	if (shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS ||
+ 		shared->page_status[slotno] == SLRU_PAGE_WRITE_IN_PROGRESS)
+ 	{
+ 		if (LWLockConditionalAcquire(shared->buffer_locks[slotno], LW_SHARED))
+ 		{
+ 			/* indeed, the I/O must have failed */
+ 			if (shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS)
+ 				shared->page_status[slotno] = SLRU_PAGE_EMPTY;
+ 			else				/* write_in_progress */
+ 			{
+ 				shared->page_status[slotno] = SLRU_PAGE_VALID;
+ 				shared->page_dirty[slotno] = true;
+ 			}
+ 			LWLockRelease(shared->buffer_locks[slotno]);
+ 		}
+ 	}
+ }
+ 
+ /*
   * Find a page in a shared buffer, reading it in if necessary.
   * The page number must correspond to an already-initialized page.
   *
***************
*** 245,251 ****
  {
  	SlruShared	shared = ctl->shared;
  
! 	/* Outer loop handles restart if we lose the buffer to someone else */
  	for (;;)
  	{
  		int			slotno;
--- 279,285 ----
  {
  	SlruShared	shared = ctl->shared;
  
! 	/* Outer loop handles restart if we must wait for someone else's I/O */
  	for (;;)
  	{
  		int			slotno;
***************
*** 258,281 ****
  		if (shared->page_number[slotno] == pageno &&
  			shared->page_status[slotno] != SLRU_PAGE_EMPTY)
  		{
! 			/* If page is still being read in, we cannot use it yet */
! 			if (shared->page_status[slotno] != SLRU_PAGE_READ_IN_PROGRESS)
  			{
! 				/* otherwise, it's ready to use */
! 				SlruRecentlyUsed(shared, slotno);
! 				return slotno;
  			}
! 		}
! 		else
! 		{
! 			/* We found no match; assert we selected a freeable slot */
! 			Assert(shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 				   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  		}
  
! 		/* Mark the slot read-busy (no-op if it already was) */
  		shared->page_number[slotno] = pageno;
  		shared->page_status[slotno] = SLRU_PAGE_READ_IN_PROGRESS;
  
  		/*
  		 * Temporarily mark page as recently-used to discourage
--- 292,321 ----
  		if (shared->page_number[slotno] == pageno &&
  			shared->page_status[slotno] != SLRU_PAGE_EMPTY)
  		{
! 			/* If page is still being read in, we must wait for I/O */
! 			if (shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS)
  			{
! 				SimpleLruWaitIO(ctl, slotno);
! 				/* Now we must recheck state from the top */
! 				continue;
  			}
! 			/* Otherwise, it's ready to use */
! 			SlruRecentlyUsed(shared, slotno);
! 			return slotno;
  		}
  
! 		/* We found no match; assert we selected a freeable slot */
! 		Assert(shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 			   (shared->page_status[slotno] == SLRU_PAGE_VALID &&
! 				!shared->page_dirty[slotno]));
! 
! 		/* Mark the slot read-busy */
  		shared->page_number[slotno] = pageno;
  		shared->page_status[slotno] = SLRU_PAGE_READ_IN_PROGRESS;
+ 		shared->page_dirty[slotno] = false;
+ 
+ 		/* Acquire per-buffer lock (cannot deadlock, see notes at top) */
+ 		LWLockAcquire(shared->buffer_locks[slotno], LW_EXCLUSIVE);
  
  		/*
  		 * Temporarily mark page as recently-used to discourage
***************
*** 283,314 ****
  		 */
  		SlruRecentlyUsed(shared, slotno);
  
! 		/* Release shared lock, grab per-buffer lock instead */
  		LWLockRelease(shared->ControlLock);
- 		LWLockAcquire(shared->buffer_locks[slotno], LW_EXCLUSIVE);
- 
- 		/*
- 		 * Check to see if someone else already did the read, or took the
- 		 * buffer away from us.  If so, restart from the top.
- 		 */
- 		if (shared->page_number[slotno] != pageno ||
- 			shared->page_status[slotno] != SLRU_PAGE_READ_IN_PROGRESS)
- 		{
- 			LWLockRelease(shared->buffer_locks[slotno]);
- 			LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
- 			continue;
- 		}
  
! 		/* Okay, do the read */
  		ok = SlruPhysicalReadPage(ctl, pageno, slotno);
  
! 		/* Re-acquire shared control lock and update page state */
  		LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
  
  		Assert(shared->page_number[slotno] == pageno &&
! 			   shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS);
  
! 		shared->page_status[slotno] = ok ? SLRU_PAGE_CLEAN : SLRU_PAGE_EMPTY;
  
  		LWLockRelease(shared->buffer_locks[slotno]);
  
--- 323,342 ----
  		 */
  		SlruRecentlyUsed(shared, slotno);
  
! 		/* Release control lock while doing I/O */
  		LWLockRelease(shared->ControlLock);
  
! 		/* Do the read */
  		ok = SlruPhysicalReadPage(ctl, pageno, slotno);
  
! 		/* Re-acquire control lock and update page state */
  		LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
  
  		Assert(shared->page_number[slotno] == pageno &&
! 			   shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS &&
! 			   !shared->page_dirty[slotno]);
  
! 		shared->page_status[slotno] = ok ? SLRU_PAGE_VALID : SLRU_PAGE_EMPTY;
  
  		LWLockRelease(shared->buffer_locks[slotno]);
  
***************
*** 336,385 ****
  SimpleLruWritePage(SlruCtl ctl, int slotno, SlruFlush fdata)
  {
  	SlruShared	shared = ctl->shared;
! 	int			pageno;
  	bool		ok;
  
! 	/* Do nothing if page does not need writing */
! 	if (shared->page_status[slotno] != SLRU_PAGE_DIRTY &&
! 		shared->page_status[slotno] != SLRU_PAGE_WRITE_IN_PROGRESS)
! 		return;
! 
! 	pageno = shared->page_number[slotno];
! 
! 	/* Release shared lock, grab per-buffer lock instead */
! 	LWLockRelease(shared->ControlLock);
! 	LWLockAcquire(shared->buffer_locks[slotno], LW_EXCLUSIVE);
  
  	/*
! 	 * Check to see if someone else already did the write, or took the buffer
! 	 * away from us.  If so, do nothing.  NOTE: we really should never see
! 	 * WRITE_IN_PROGRESS here, since that state should only occur while the
! 	 * writer is holding the buffer lock.  But accept it so that we have a
! 	 * recovery path if a writer aborts.
! 	 */
! 	if (shared->page_number[slotno] != pageno ||
! 		(shared->page_status[slotno] != SLRU_PAGE_DIRTY &&
! 		 shared->page_status[slotno] != SLRU_PAGE_WRITE_IN_PROGRESS))
! 	{
! 		LWLockRelease(shared->buffer_locks[slotno]);
! 		LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
  		return;
- 	}
  
  	/*
! 	 * Mark the slot write-busy.  After this point, a transaction status
! 	 * update on this page will mark it dirty again.  NB: we are assuming that
! 	 * read/write of the page status field is atomic, since we change the
! 	 * state while not holding control lock.  However, we cannot set this
! 	 * state any sooner, or we'd possibly fool a previous writer into thinking
! 	 * he's successfully dumped the page when he hasn't. (Scenario: other
! 	 * writer starts, page is redirtied, we come along and set
! 	 * WRITE_IN_PROGRESS again, other writer completes and sets CLEAN because
! 	 * redirty info has been lost, then we think it's clean too.)
  	 */
  	shared->page_status[slotno] = SLRU_PAGE_WRITE_IN_PROGRESS;
  
! 	/* Okay, do the write */
  	ok = SlruPhysicalWritePage(ctl, pageno, slotno, fdata);
  
  	/* If we failed, and we're in a flush, better close the files */
--- 364,402 ----
  SimpleLruWritePage(SlruCtl ctl, int slotno, SlruFlush fdata)
  {
  	SlruShared	shared = ctl->shared;
! 	int			pageno = shared->page_number[slotno];
  	bool		ok;
  
! 	/* If a write is in progress, wait for it to finish */
! 	while (shared->page_status[slotno] == SLRU_PAGE_WRITE_IN_PROGRESS &&
! 		   shared->page_number[slotno] == pageno)
! 	{
! 		SimpleLruWaitIO(ctl, slotno);
! 	}
  
  	/*
! 	 * Do nothing if page is not dirty, or if buffer no longer contains
! 	 * the same page we were called for.
! 	 */
! 	if (!shared->page_dirty[slotno] ||
! 		shared->page_status[slotno] != SLRU_PAGE_VALID ||
! 		shared->page_number[slotno] != pageno)
  		return;
  
  	/*
! 	 * Mark the slot write-busy, and clear the dirtybit.  After this point,
! 	 * a transaction status update on this page will mark it dirty again.
  	 */
  	shared->page_status[slotno] = SLRU_PAGE_WRITE_IN_PROGRESS;
+ 	shared->page_dirty[slotno] = false;
+ 
+ 	/* Acquire per-buffer lock (cannot deadlock, see notes at top) */
+ 	LWLockAcquire(shared->buffer_locks[slotno], LW_EXCLUSIVE);
+ 
+ 	/* Release control lock while doing I/O */
+ 	LWLockRelease(shared->ControlLock);
  
! 	/* Do the write */
  	ok = SlruPhysicalWritePage(ctl, pageno, slotno, fdata);
  
  	/* If we failed, and we're in a flush, better close the files */
***************
*** 391,406 ****
  			close(fdata->fd[i]);
  	}
  
! 	/* Re-acquire shared control lock and update page state */
  	LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
  
  	Assert(shared->page_number[slotno] == pageno &&
! 		   (shared->page_status[slotno] == SLRU_PAGE_WRITE_IN_PROGRESS ||
! 			shared->page_status[slotno] == SLRU_PAGE_DIRTY));
  
! 	/* Cannot set CLEAN if someone re-dirtied page since write started */
! 	if (shared->page_status[slotno] == SLRU_PAGE_WRITE_IN_PROGRESS)
! 		shared->page_status[slotno] = ok ? SLRU_PAGE_CLEAN : SLRU_PAGE_DIRTY;
  
  	LWLockRelease(shared->buffer_locks[slotno]);
  
--- 408,424 ----
  			close(fdata->fd[i]);
  	}
  
! 	/* Re-acquire control lock and update page state */
  	LWLockAcquire(shared->ControlLock, LW_EXCLUSIVE);
  
  	Assert(shared->page_number[slotno] == pageno &&
! 		   shared->page_status[slotno] == SLRU_PAGE_WRITE_IN_PROGRESS);
! 
! 	/* If we failed to write, mark the page dirty again */
! 	if (!ok)
! 		shared->page_dirty[slotno] = true;
  
! 	shared->page_status[slotno] = SLRU_PAGE_VALID;
  
  	LWLockRelease(shared->buffer_locks[slotno]);
  
***************
*** 739,758 ****
  		/*
  		 * If the selected page is clean, we're set.
  		 */
! 		if (shared->page_status[bestslot] == SLRU_PAGE_CLEAN)
  			return bestslot;
  
  		/*
! 		 * We need to do I/O.  Normal case is that we have to write it out,
! 		 * but it's possible in the worst case to have selected a read-busy
! 		 * page.  In that case we use SimpleLruReadPage to wait for the read
! 		 * to complete.
  		 */
! 		if (shared->page_status[bestslot] == SLRU_PAGE_READ_IN_PROGRESS)
! 			(void) SimpleLruReadPage(ctl, shared->page_number[bestslot],
! 									 InvalidTransactionId);
! 		else
  			SimpleLruWritePage(ctl, bestslot, NULL);
  
  		/*
  		 * Now loop back and try again.  This is the easiest way of dealing
--- 757,776 ----
  		/*
  		 * If the selected page is clean, we're set.
  		 */
! 		if (shared->page_status[bestslot] == SLRU_PAGE_VALID &&
! 			!shared->page_dirty[bestslot])
  			return bestslot;
  
  		/*
! 		 * We need to wait for I/O.  Normal case is that it's dirty and we
! 		 * must initiate a write, but it's possible that the page is already
! 		 * write-busy, or in the worst case still read-busy.  In those cases
! 		 * we wait for the existing I/O to complete.
  		 */
! 		if (shared->page_status[bestslot] == SLRU_PAGE_VALID)
  			SimpleLruWritePage(ctl, bestslot, NULL);
+ 		else
+ 			SimpleLruWaitIO(ctl, bestslot);
  
  		/*
  		 * Now loop back and try again.  This is the easiest way of dealing
***************
*** 793,799 ****
  		 */
  		Assert(checkpoint ||
  			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 			   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  	}
  
  	LWLockRelease(shared->ControlLock);
--- 811,818 ----
  		 */
  		Assert(checkpoint ||
  			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
! 			   (shared->page_status[slotno] == SLRU_PAGE_VALID &&
! 				!shared->page_dirty[slotno]));
  	}
  
  	LWLockRelease(shared->ControlLock);
***************
*** 871,879 ****
  			continue;
  
  		/*
! 		 * If page is CLEAN, just change state to EMPTY (expected case).
  		 */
! 		if (shared->page_status[slotno] == SLRU_PAGE_CLEAN)
  		{
  			shared->page_status[slotno] = SLRU_PAGE_EMPTY;
  			continue;
--- 890,899 ----
  			continue;
  
  		/*
! 		 * If page is clean, just change state to EMPTY (expected case).
  		 */
! 		if (shared->page_status[slotno] == SLRU_PAGE_VALID &&
! 			!shared->page_dirty[slotno])
  		{
  			shared->page_status[slotno] = SLRU_PAGE_EMPTY;
  			continue;
***************
*** 882,894 ****
  		/*
  		 * Hmm, we have (or may have) I/O operations acting on the page, so
  		 * we've got to wait for them to finish and then start again. This is
! 		 * the same logic as in SlruSelectLRUPage.
  		 */
! 		if (shared->page_status[slotno] == SLRU_PAGE_READ_IN_PROGRESS)
! 			(void) SimpleLruReadPage(ctl, shared->page_number[slotno],
! 									 InvalidTransactionId);
! 		else
  			SimpleLruWritePage(ctl, slotno, NULL);
  		goto restart;
  	}
  
--- 902,915 ----
  		/*
  		 * Hmm, we have (or may have) I/O operations acting on the page, so
  		 * we've got to wait for them to finish and then start again. This is
! 		 * the same logic as in SlruSelectLRUPage.  (XXX if page is dirty,
! 		 * wouldn't it be OK to just discard it without writing it?  For now,
! 		 * keep the logic the same as it was.)
  		 */
! 		if (shared->page_status[slotno] == SLRU_PAGE_VALID)
  			SimpleLruWritePage(ctl, slotno, NULL);
+ 		else
+ 			SimpleLruWaitIO(ctl, slotno);
  		goto restart;
  	}
  
*** src/backend/access/transam/subtrans.c.orig	Fri Oct 14 22:59:37 2005
--- src/backend/access/transam/subtrans.c	Mon Oct 31 10:08:25 2005
***************
*** 86,92 ****
  
  	*ptr = parent;
  
! 	SubTransCtl->shared->page_status[slotno] = SLRU_PAGE_DIRTY;
  
  	LWLockRelease(SubtransControlLock);
  }
--- 86,92 ----
  
  	*ptr = parent;
  
! 	SubTransCtl->shared->page_dirty[slotno] = true;
  
  	LWLockRelease(SubtransControlLock);
  }
***************
*** 199,205 ****
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(SubTransCtl, slotno, NULL);
! 	Assert(SubTransCtl->shared->page_status[slotno] == SLRU_PAGE_CLEAN);
  
  	LWLockRelease(SubtransControlLock);
  }
--- 199,205 ----
  
  	/* Make sure it's written out */
  	SimpleLruWritePage(SubTransCtl, slotno, NULL);
! 	Assert(!SubTransCtl->shared->page_dirty[slotno]);
  
  	LWLockRelease(SubtransControlLock);
  }
*** src/include/access/slru.h.orig	Sun Oct 30 11:16:01 2005
--- src/include/access/slru.h	Mon Oct 31 11:24:07 2005
***************
*** 23,35 ****
   */
  #define NUM_SLRU_BUFFERS	8
  
! /* Page status codes */
  typedef enum
  {
  	SLRU_PAGE_EMPTY,			/* buffer is not in use */
  	SLRU_PAGE_READ_IN_PROGRESS, /* page is being read in */
! 	SLRU_PAGE_CLEAN,			/* page is valid and not dirty */
! 	SLRU_PAGE_DIRTY,			/* page is valid but needs write */
  	SLRU_PAGE_WRITE_IN_PROGRESS /* page is being written out */
  } SlruPageStatus;
  
--- 23,39 ----
   */
  #define NUM_SLRU_BUFFERS	8
  
! /*
!  * Page status codes.  Note that these do not include the "dirty" bit.
!  * page_dirty can be TRUE only in the VALID or WRITE_IN_PROGRESS states;
!  * in the latter case it implies that the page has been re-dirtied since
!  * the write started.
!  */
  typedef enum
  {
  	SLRU_PAGE_EMPTY,			/* buffer is not in use */
  	SLRU_PAGE_READ_IN_PROGRESS, /* page is being read in */
! 	SLRU_PAGE_VALID,			/* page is valid and not being written */
  	SLRU_PAGE_WRITE_IN_PROGRESS /* page is being written out */
  } SlruPageStatus;
  
***************
*** 48,53 ****
--- 52,58 ----
  	 */
  	char	   *page_buffer[NUM_SLRU_BUFFERS];
  	SlruPageStatus page_status[NUM_SLRU_BUFFERS];
+ 	bool		page_dirty[NUM_SLRU_BUFFERS];
  	int			page_number[NUM_SLRU_BUFFERS];
  	unsigned int page_lru_count[NUM_SLRU_BUFFERS];
  	LWLockId	buffer_locks[NUM_SLRU_BUFFERS];