Re: Simple LRU for clog and (later) subtrans

От:

Bruce Momjian <pgman@candle.pha.pa.us>

Дата:

1 июня 2003 г. в 00:52:08


(Applying newer version, but posting this because it has a description.)

Your patch has been added to the PostgreSQL unapplied patches list at:

	http://momjian.postgresql.org/cgi-bin/pgpatches

I will try to apply it within the next 48 hours.

---------------------------------------------------------------------------


Manfred Koizar wrote:
> This patch extracts page buffer pooling and the simple
> least-recently-used strategy from clog.c into slru.c.  It doesn't
> change any visible behaviour and passes all regression tests plus a
> TruncateCLOG test done manually.
> 
> Apart from refactoring I made a little change to SlruRecentlyUsed,
> formerly ClogRecentlyUsed:  It now skips incrementing lru_counts, if
> slotno is already the LRU slot, thus saving a few CPU cycles.  To make
> this work, lru_counts are initialised to 1 in SimpleLruInit.
> 
> SimpleLru will be used by pg_subtrans (part of the nested transactions
> project), so the main purpose of this patch is to avoid future code
> duplication.

-- 
  Bruce Momjian                        |  http://candle.pha.pa.us
  pgman@candle.pha.pa.us               |  (610) 359-1001
  +  If your life is a hard drive,     |  13 Roberts Road
  +  Christ can be your backup.        |  Newtown Square, Pennsylvania 19073

Re: Simple LRU for clog and (later) subtrans

От:

Bruce Momjian <pgman@candle.pha.pa.us>

Дата:

11 июня 2003 г. в 19:38:42


Newest version applied --- new files added.  Thanks.

---------------------------------------------------------------------------

Manfred Koizar wrote:
> This patch extracts page buffer pooling and the simple
> least-recently-used strategy from clog.c into slru.c.  It doesn't
> change any visible behaviour and passes all regression tests plus a
> TruncateCLOG test done manually.
> 
> Apart from refactoring I made a little change to SlruRecentlyUsed,
> formerly ClogRecentlyUsed:  It now skips incrementing lru_counts, if
> slotno is already the LRU slot, thus saving a few CPU cycles.  To make
> this work, lru_counts are initialised to 1 in SimpleLruInit.
> 
> SimpleLru will be used by pg_subtrans (part of the nested transactions
> project), so the main purpose of this patch is to avoid future code
> duplication.
> 
> Servus
>  Manfred

> diff -ruN ../base/src/backend/access/transam/Makefile src/backend/access/transam/Makefile
> --- ../base/src/backend/access/transam/Makefile	2001-08-25 20:52:41.000000000 +0200
> +++ src/backend/access/transam/Makefile	2003-05-01 20:11:44.000000000 +0200
> @@ -12,7 +12,7 @@
>  top_builddir = ../../../..
>  include $(top_builddir)/src/Makefile.global
>  
> -OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o
> +OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o slru.o
>  
>  all: SUBSYS.o
>  
> diff -ruN ../base/src/backend/access/transam/clog.c src/backend/access/transam/clog.c
> --- ../base/src/backend/access/transam/clog.c	2003-04-18 19:09:07.000000000 +0200
> +++ src/backend/access/transam/clog.c	2003-05-02 00:20:07.000000000 +0200
> @@ -26,15 +26,14 @@
>  #include 
>  
>  #include "access/clog.h"
> +#include "access/slru.h"
>  #include "storage/lwlock.h"
>  #include "miscadmin.h"
>  
>  
>  /*
>   * Defines for CLOG page and segment sizes.  A page is the same BLCKSZ
> - * as is used everywhere else in Postgres.	The CLOG segment size can be
> - * chosen somewhat arbitrarily; we make it 1 million transactions by default,
> - * or 256Kb.
> + * as is used everywhere else in Postgres.
>   *
>   * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
>   * CLOG page numbering also wraps around at 0xFFFFFFFF/CLOG_XACTS_PER_PAGE,
> @@ -43,17 +42,12 @@
>   * segment and page numbers in TruncateCLOG (see CLOGPagePrecedes).
>   */
>  
> -#define CLOG_BLCKSZ			BLCKSZ
> -
>  /* We need two bits per xact, so four xacts fit in a byte */
>  #define CLOG_BITS_PER_XACT	2
>  #define CLOG_XACTS_PER_BYTE 4
> -#define CLOG_XACTS_PER_PAGE (CLOG_BLCKSZ * CLOG_XACTS_PER_BYTE)
> +#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
>  #define CLOG_XACT_BITMASK	((1 << CLOG_BITS_PER_XACT) - 1)
>  
> -#define CLOG_XACTS_PER_SEGMENT	0x100000
> -#define CLOG_PAGES_PER_SEGMENT	(CLOG_XACTS_PER_SEGMENT / CLOG_XACTS_PER_PAGE)
> -
>  #define TransactionIdToPage(xid)	((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
>  #define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
>  #define TransactionIdToByte(xid)	(TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
> @@ -63,47 +57,6 @@
>  /*----------
>   * Shared-memory data structures for CLOG control
>   *
> - * We use a simple least-recently-used scheme to manage a pool of page
> - * buffers for the CLOG.  Under ordinary circumstances we expect that write
> - * traffic will occur mostly to the latest CLOG page (and to the just-prior
> - * page, soon after a page transition).  Read traffic will probably touch
> - * a larger span of pages, but in any case a fairly small number of page
> - * buffers should be sufficient.  So, we just search the buffers using plain
> - * linear search; there's no need for a hashtable or anything fancy.
> - * The management algorithm is straight LRU except that we will never swap
> - * out the latest page (since we know it's going to be hit again eventually).
> - *
> - * We use an overall 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.
> - *
>   * XLOG interactions: this module generates an XLOG record whenever a new
>   * CLOG page is initialized to zeroes.	Other writes of CLOG come from
>   * recording of transaction commit or abort in xact.c, which generates its
> @@ -117,90 +70,12 @@
>   *----------
>   */
>  
> -typedef enum
> -{
> -	CLOG_PAGE_EMPTY,			/* CLOG buffer is not in use */
> -	CLOG_PAGE_READ_IN_PROGRESS, /* CLOG page is being read in */
> -	CLOG_PAGE_CLEAN,			/* CLOG page is valid and not dirty */
> -	CLOG_PAGE_DIRTY,			/* CLOG page is valid but needs write */
> -	CLOG_PAGE_WRITE_IN_PROGRESS /* CLOG page is being written out */
> -} ClogPageStatus;
> -
> -/*
> - * Shared-memory state for CLOG.
> - */
> -typedef struct ClogCtlData
> -{
> -	/*
> -	 * Info for each buffer slot.  Page number is undefined when status is
> -	 * EMPTY.  lru_count is essentially the number of operations since
> -	 * last use of this page; the page with highest lru_count is the best
> -	 * candidate to replace.
> -	 */
> -	char	   *page_buffer[NUM_CLOG_BUFFERS];
> -	ClogPageStatus page_status[NUM_CLOG_BUFFERS];
> -	int			page_number[NUM_CLOG_BUFFERS];
> -	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
> -
> -	/*
> -	 * latest_page_number is the page number of the current end of the
> -	 * CLOG; this is not critical data, since we use it only to avoid
> -	 * swapping out the latest page.
> -	 */
> -	int			latest_page_number;
> -} ClogCtlData;
> -
> -static ClogCtlData *ClogCtl = NULL;
> -
> -/*
> - * ClogBufferLocks is set during CLOGShmemInit and does not change thereafter.
> - * The value is automatically inherited by backends via fork, and
> - * doesn't need to be in shared memory.
> - */
> -static LWLockId ClogBufferLocks[NUM_CLOG_BUFFERS];		/* Per-buffer I/O locks */
> -
> -/*
> - * ClogDir is set during CLOGShmemInit and does not change thereafter.
> - * The value is automatically inherited by backends via fork, and
> - * doesn't need to be in shared memory.
> - */
> -static char ClogDir[MAXPGPATH];
> -
> -#define ClogFileName(path, seg) \
> -	snprintf(path, MAXPGPATH, "%s/%04X", ClogDir, seg)
> -
> -/*
> - * Macro to mark a buffer slot "most recently used".
> - */
> -#define ClogRecentlyUsed(slotno)	\
> -	do { \
> -		int		iilru; \
> -		for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
> -			ClogCtl->page_lru_count[iilru]++; \
> -		ClogCtl->page_lru_count[slotno] = 0; \
> -	} while (0)
> -
> -/* Saved info for CLOGReportIOError */
> -typedef enum
> -{
> -	CLOG_OPEN_FAILED,
> -	CLOG_CREATE_FAILED,
> -	CLOG_SEEK_FAILED,
> -	CLOG_READ_FAILED,
> -	CLOG_WRITE_FAILED
> -} ClogErrorCause;
> -static ClogErrorCause clog_errcause;
> -static int	clog_errno;
>  
> +static SlruCtlData ClogCtlData;
> +static SlruCtl ClogCtl = &ClogCtlData;
> +  
>  
>  static int	ZeroCLOGPage(int pageno, bool writeXlog);
> -static int	ReadCLOGPage(int pageno, TransactionId xid);
> -static void WriteCLOGPage(int slotno);
> -static bool CLOGPhysicalReadPage(int pageno, int slotno);
> -static bool CLOGPhysicalWritePage(int pageno, int slotno);
> -static void CLOGReportIOError(int pageno, TransactionId xid);
> -static int	SelectLRUCLOGPage(int pageno);
> -static bool ScanCLOGDirectory(int cutoffPage, bool doDeletions);
>  static bool CLOGPagePrecedes(int page1, int page2);
>  static void WriteZeroPageXlogRec(int pageno);
>  
> @@ -217,16 +92,15 @@
>  	int			pageno = TransactionIdToPage(xid);
>  	int			byteno = TransactionIdToByte(xid);
>  	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
> -	int			slotno;
>  	char	   *byteptr;
>  
>  	Assert(status == TRANSACTION_STATUS_COMMITTED ||
>  		   status == TRANSACTION_STATUS_ABORTED);
>  
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> +	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
>  
> -	slotno = ReadCLOGPage(pageno, xid);
> -	byteptr = ClogCtl->page_buffer[slotno] + byteno;
> +	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, true);
> +	byteptr += byteno;
>  
>  	/* Current state should be 0 or target state */
>  	Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
> @@ -234,9 +108,9 @@
>  
>  	*byteptr |= (status << bshift);
>  
> -	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
> +	/* ...->page_status[slotno] = CLOG_PAGE_DIRTY; already done */
>  
> -	LWLockRelease(CLogControlLock);
> +	LWLockRelease(ClogCtl->ControlLock);
>  }
>  
>  /*
> @@ -251,18 +125,17 @@
>  	int			pageno = TransactionIdToPage(xid);
>  	int			byteno = TransactionIdToByte(xid);
>  	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
> -	int			slotno;
>  	char	   *byteptr;
>  	XidStatus	status;
>  
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> +	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
>  
> -	slotno = ReadCLOGPage(pageno, xid);
> -	byteptr = ClogCtl->page_buffer[slotno] + byteno;
> +	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, false);
> +	byteptr += byteno;
>  
>  	status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
>  
> -	LWLockRelease(CLogControlLock);
> +	LWLockRelease(ClogCtl->ControlLock);
>  
>  	return status;
>  }
> @@ -275,40 +148,14 @@
>  int
>  CLOGShmemSize(void)
>  {
> -	return MAXALIGN(sizeof(ClogCtlData) + CLOG_BLCKSZ * NUM_CLOG_BUFFERS);
> +	return SimpleLruShmemSize();
>  }
>  
>  void
>  CLOGShmemInit(void)
>  {
> -	bool		found;
> -	char	   *bufptr;
> -	int			slotno;
> -
> -	/* this must agree with space requested by CLOGShmemSize() */
> -	ClogCtl = (ClogCtlData *)
> -		ShmemInitStruct("CLOG Ctl",
> -						MAXALIGN(sizeof(ClogCtlData) +
> -								 CLOG_BLCKSZ * NUM_CLOG_BUFFERS),
> -						&found);
> -	Assert(!found);
> -
> -	memset(ClogCtl, 0, sizeof(ClogCtlData));
> -
> -	bufptr = ((char *) ClogCtl) + sizeof(ClogCtlData);
> -
> -	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -	{
> -		ClogCtl->page_buffer[slotno] = bufptr;
> -		ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
> -		ClogBufferLocks[slotno] = LWLockAssign();
> -		bufptr += CLOG_BLCKSZ;
> -	}
> -
> -	/* ClogCtl->latest_page_number will be set later */
> -
> -	/* Init CLOG directory path */
> -	snprintf(ClogDir, MAXPGPATH, "%s/pg_clog", DataDir);
> +	SimpleLruInit(ClogCtl, "CLOG Ctl", "pg_clog");
> +	ClogCtl->PagePrecedes = CLOGPagePrecedes;
>  }
>  
>  /*
> @@ -322,16 +169,16 @@
>  {
>  	int			slotno;
>  
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> +	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
>  
>  	/* Create and zero the first page of the commit log */
>  	slotno = ZeroCLOGPage(0, false);
>  
>  	/* Make sure it's written out */
> -	WriteCLOGPage(slotno);
> -	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
> +	SimpleLruWritePage(ClogCtl, slotno);
> +	/* Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN); */
>  
> -	LWLockRelease(CLogControlLock);
> +	LWLockRelease(ClogCtl->ControlLock);
>  }
>  
>  /*
> @@ -346,24 +193,7 @@
>  static int
>  ZeroCLOGPage(int pageno, bool writeXlog)
>  {
> -	int			slotno;
> -
> -	/* Find a suitable buffer slot for the page */
> -	slotno = SelectLRUCLOGPage(pageno);
> -	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
> -		   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN ||
> -		   ClogCtl->page_number[slotno] == pageno);
> -
> -	/* Mark the slot as containing this page */
> -	ClogCtl->page_number[slotno] = pageno;
> -	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
> -	ClogRecentlyUsed(slotno);
> -
> -	/* Set the buffer to zeroes */
> -	MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
> -
> -	/* Assume this page is now the latest active page */
> -	ClogCtl->latest_page_number = pageno;
> +	int			slotno = SimpleLruZeroPage(ClogCtl, pageno);
>  
>  	if (writeXlog)
>  		WriteZeroPageXlogRec(pageno);
> @@ -372,430 +202,6 @@
>  }
>  
>  /*
> - * Find a CLOG page in a shared buffer, reading it in if necessary.
> - * The page number must correspond to an already-initialized page.
> - *
> - * The passed-in xid is used only for error reporting, and may be
> - * InvalidTransactionId if no specific xid is associated with the action.
> - *
> - * Return value is the shared-buffer slot number now holding the page.
> - * The buffer's LRU access info is updated.
> - *
> - * Control lock must be held at entry, and will be held at exit.
> - */
> -static int
> -ReadCLOGPage(int pageno, TransactionId xid)
> -{
> -	/* Outer loop handles restart if we lose the buffer to someone else */
> -	for (;;)
> -	{
> -		int			slotno;
> -		bool		ok;
> -
> -		/* See if page already is in memory; if not, pick victim slot */
> -		slotno = SelectLRUCLOGPage(pageno);
> -
> -		/* Did we find the page in memory? */
> -		if (ClogCtl->page_number[slotno] == pageno &&
> -			ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
> -		{
> -			/* If page is still being read in, we cannot use it yet */
> -			if (ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
> -			{
> -				/* otherwise, it's ready to use */
> -				ClogRecentlyUsed(slotno);
> -				return slotno;
> -			}
> -		}
> -		else
> -		{
> -			/* We found no match; assert we selected a freeable slot */
> -			Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
> -				   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
> -		}
> -
> -		/* Mark the slot read-busy (no-op if it already was) */
> -		ClogCtl->page_number[slotno] = pageno;
> -		ClogCtl->page_status[slotno] = CLOG_PAGE_READ_IN_PROGRESS;
> -
> -		/*
> -		 * Temporarily mark page as recently-used to discourage
> -		 * SelectLRUCLOGPage from selecting it again for someone else.
> -		 */
> -		ClogCtl->page_lru_count[slotno] = 0;
> -
> -		/* Release shared lock, grab per-buffer lock instead */
> -		LWLockRelease(CLogControlLock);
> -		LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
> -			ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
> -		{
> -			LWLockRelease(ClogBufferLocks[slotno]);
> -			LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -			continue;
> -		}
> -
> -		/* Okay, do the read */
> -		ok = CLOGPhysicalReadPage(pageno, slotno);
> -
> -		/* Re-acquire shared control lock and update page state */
> -		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -
> -		Assert(ClogCtl->page_number[slotno] == pageno &&
> -			 ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS);
> -
> -		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_EMPTY;
> -
> -		LWLockRelease(ClogBufferLocks[slotno]);
> -
> -		/* Now it's okay to elog if we failed */
> -		if (!ok)
> -			CLOGReportIOError(pageno, xid);
> -
> -		ClogRecentlyUsed(slotno);
> -		return slotno;
> -	}
> -}
> -
> -/*
> - * Write a CLOG page from a shared buffer, if necessary.
> - * Does nothing if the specified slot is not dirty.
> - *
> - * NOTE: only one write attempt is made here.  Hence, it is possible that
> - * the page is still dirty at exit (if someone else re-dirtied it during
> - * the write).	However, we *do* attempt a fresh write even if the page
> - * is already being written; this is for checkpoints.
> - *
> - * Control lock must be held at entry, and will be held at exit.
> - */
> -static void
> -WriteCLOGPage(int slotno)
> -{
> -	int			pageno;
> -	bool		ok;
> -
> -	/* Do nothing if page does not need writing */
> -	if (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
> -		ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS)
> -		return;
> -
> -	pageno = ClogCtl->page_number[slotno];
> -
> -	/* Release shared lock, grab per-buffer lock instead */
> -	LWLockRelease(CLogControlLock);
> -	LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
> -		(ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
> -		 ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS))
> -	{
> -		LWLockRelease(ClogBufferLocks[slotno]);
> -		LWLockAcquire(CLogControlLock, 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.)
> -	 */
> -	ClogCtl->page_status[slotno] = CLOG_PAGE_WRITE_IN_PROGRESS;
> -
> -	/* Okay, do the write */
> -	ok = CLOGPhysicalWritePage(pageno, slotno);
> -
> -	/* Re-acquire shared control lock and update page state */
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -
> -	Assert(ClogCtl->page_number[slotno] == pageno &&
> -		   (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS ||
> -			ClogCtl->page_status[slotno] == CLOG_PAGE_DIRTY));
> -
> -	/* Cannot set CLEAN if someone re-dirtied page since write started */
> -	if (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS)
> -		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_DIRTY;
> -
> -	LWLockRelease(ClogBufferLocks[slotno]);
> -
> -	/* Now it's okay to elog if we failed */
> -	if (!ok)
> -		CLOGReportIOError(pageno, InvalidTransactionId);
> -}
> -
> -/*
> - * Physical read of a (previously existing) page into a buffer slot
> - *
> - * On failure, we cannot just elog(ERROR) since caller has put state in
> - * shared memory that must be undone.  So, we return FALSE and save enough
> - * info in static variables to let CLOGReportIOError make the report.
> - *
> - * For now, assume it's not worth keeping a file pointer open across
> - * read/write operations.  We could cache one virtual file pointer ...
> - */
> -static bool
> -CLOGPhysicalReadPage(int pageno, int slotno)
> -{
> -	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
> -	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
> -	int			offset = rpageno * CLOG_BLCKSZ;
> -	char		path[MAXPGPATH];
> -	int			fd;
> -
> -	ClogFileName(path, segno);
> -
> -	/*
> -	 * In a crash-and-restart situation, it's possible for us to receive
> -	 * commands to set the commit status of transactions whose bits are in
> -	 * already-truncated segments of the commit log (see notes in
> -	 * CLOGPhysicalWritePage).	Hence, if we are InRecovery, allow the
> -	 * case where the file doesn't exist, and return zeroes instead.
> -	 */
> -	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
> -	if (fd < 0)
> -	{
> -		if (errno != ENOENT || !InRecovery)
> -		{
> -			clog_errcause = CLOG_OPEN_FAILED;
> -			clog_errno = errno;
> -			return false;
> -		}
> -
> -		elog(LOG, "clog file %s doesn't exist, reading as zeroes", path);
> -		MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
> -		return true;
> -	}
> -
> -	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
> -	{
> -		clog_errcause = CLOG_SEEK_FAILED;
> -		clog_errno = errno;
> -		return false;
> -	}
> -
> -	errno = 0;
> -	if (read(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
> -	{
> -		clog_errcause = CLOG_READ_FAILED;
> -		clog_errno = errno;
> -		return false;
> -	}
> -
> -	close(fd);
> -	return true;
> -}
> -
> -/*
> - * Physical write of a page from a buffer slot
> - *
> - * On failure, we cannot just elog(ERROR) since caller has put state in
> - * shared memory that must be undone.  So, we return FALSE and save enough
> - * info in static variables to let CLOGReportIOError make the report.
> - *
> - * For now, assume it's not worth keeping a file pointer open across
> - * read/write operations.  We could cache one virtual file pointer ...
> - */
> -static bool
> -CLOGPhysicalWritePage(int pageno, int slotno)
> -{
> -	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
> -	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
> -	int			offset = rpageno * CLOG_BLCKSZ;
> -	char		path[MAXPGPATH];
> -	int			fd;
> -
> -	ClogFileName(path, segno);
> -
> -	/*
> -	 * If the file doesn't already exist, we should create it.  It is
> -	 * possible for this to need to happen when writing a page that's not
> -	 * first in its segment; we assume the OS can cope with that.  (Note:
> -	 * it might seem that it'd be okay to create files only when
> -	 * ZeroCLOGPage is called for the first page of a segment.	However,
> -	 * if after a crash and restart the REDO logic elects to replay the
> -	 * log from a checkpoint before the latest one, then it's possible
> -	 * that we will get commands to set transaction status of transactions
> -	 * that have already been truncated from the commit log.  Easiest way
> -	 * to deal with that is to accept references to nonexistent files here
> -	 * and in CLOGPhysicalReadPage.)
> -	 */
> -	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
> -	if (fd < 0)
> -	{
> -		if (errno != ENOENT)
> -		{
> -			clog_errcause = CLOG_OPEN_FAILED;
> -			clog_errno = errno;
> -			return false;
> -		}
> -
> -		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
> -						   S_IRUSR | S_IWUSR);
> -		if (fd < 0)
> -		{
> -			clog_errcause = CLOG_CREATE_FAILED;
> -			clog_errno = errno;
> -			return false;
> -		}
> -	}
> -
> -	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
> -	{
> -		clog_errcause = CLOG_SEEK_FAILED;
> -		clog_errno = errno;
> -		return false;
> -	}
> -
> -	errno = 0;
> -	if (write(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
> -	{
> -		/* if write didn't set errno, assume problem is no disk space */
> -		if (errno == 0)
> -			errno = ENOSPC;
> -		clog_errcause = CLOG_WRITE_FAILED;
> -		clog_errno = errno;
> -		return false;
> -	}
> -
> -	close(fd);
> -	return true;
> -}
> -
> -/*
> - * Issue the error message after failure of CLOGPhysicalReadPage or
> - * CLOGPhysicalWritePage.  Call this after cleaning up shared-memory state.
> - */
> -static void
> -CLOGReportIOError(int pageno, TransactionId xid)
> -{
> -	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
> -	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
> -	int			offset = rpageno * CLOG_BLCKSZ;
> -	char		path[MAXPGPATH];
> -
> -	/* XXX TODO: provide xid as context in error messages */
> -
> -	ClogFileName(path, segno);
> -	errno = clog_errno;
> -	switch (clog_errcause)
> -	{
> -		case CLOG_OPEN_FAILED:
> -			elog(ERROR, "open of %s failed: %m", path);
> -			break;
> -		case CLOG_CREATE_FAILED:
> -			elog(ERROR, "creation of file %s failed: %m", path);
> -			break;
> -		case CLOG_SEEK_FAILED:
> -			elog(ERROR, "lseek of clog file %u, offset %u failed: %m",
> -				 segno, offset);
> -			break;
> -		case CLOG_READ_FAILED:
> -			elog(ERROR, "read of clog file %u, offset %u failed: %m",
> -				 segno, offset);
> -			break;
> -		case CLOG_WRITE_FAILED:
> -			elog(ERROR, "write of clog file %u, offset %u failed: %m",
> -				 segno, offset);
> -			break;
> -		default:
> -			/* can't get here, we trust */
> -			elog(ERROR, "unknown CLOG I/O error");
> -			break;
> -	}
> -}
> -
> -/*
> - * Select the slot to re-use when we need a free slot.
> - *
> - * The target page number is passed because we need to consider the
> - * possibility that some other process reads in the target page while
> - * we are doing I/O to free a slot.  Hence, check or recheck to see if
> - * any slot already holds the target page, and return that slot if so.
> - * Thus, the returned slot is *either* a slot already holding the pageno
> - * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
> - * or CLEAN).
> - *
> - * Control lock must be held at entry, and will be held at exit.
> - */
> -static int
> -SelectLRUCLOGPage(int pageno)
> -{
> -	/* Outer loop handles restart after I/O */
> -	for (;;)
> -	{
> -		int			slotno;
> -		int			bestslot = 0;
> -		unsigned int bestcount = 0;
> -
> -		/* See if page already has a buffer assigned */
> -		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -		{
> -			if (ClogCtl->page_number[slotno] == pageno &&
> -				ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
> -				return slotno;
> -		}
> -
> -		/*
> -		 * If we find any EMPTY slot, just select that one. Else locate
> -		 * the least-recently-used slot that isn't the latest CLOG page.
> -		 */
> -		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -		{
> -			if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
> -				return slotno;
> -			if (ClogCtl->page_lru_count[slotno] > bestcount &&
> -			 ClogCtl->page_number[slotno] != ClogCtl->latest_page_number)
> -			{
> -				bestslot = slotno;
> -				bestcount = ClogCtl->page_lru_count[slotno];
> -			}
> -		}
> -
> -		/*
> -		 * If the selected page is clean, we're set.
> -		 */
> -		if (ClogCtl->page_status[bestslot] == CLOG_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 ReadCLOGPage to wait for
> -		 * the read to complete.
> -		 */
> -		if (ClogCtl->page_status[bestslot] == CLOG_PAGE_READ_IN_PROGRESS)
> -			(void) ReadCLOGPage(ClogCtl->page_number[bestslot],
> -								InvalidTransactionId);
> -		else
> -			WriteCLOGPage(bestslot);
> -
> -		/*
> -		 * Now loop back and try again.  This is the easiest way of
> -		 * dealing with corner cases such as the victim page being
> -		 * re-dirtied while we wrote it.
> -		 */
> -	}
> -}
> -
> -/*
>   * This must be called ONCE during postmaster or standalone-backend startup,
>   * after StartupXLOG has initialized ShmemVariableCache->nextXid.
>   */
> @@ -805,7 +211,7 @@
>  	/*
>  	 * Initialize our idea of the latest page number.
>  	 */
> -	ClogCtl->latest_page_number = TransactionIdToPage(ShmemVariableCache->nextXid);
> +	SimpleLruSetLatestPage(ClogCtl, TransactionIdToPage(ShmemVariableCache->nextXid));
>  }
>  
>  /*
> @@ -814,18 +220,7 @@
>  void
>  ShutdownCLOG(void)
>  {
> -	int			slotno;
> -
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -
> -	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -	{
> -		WriteCLOGPage(slotno);
> -		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
> -			   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
> -	}
> -
> -	LWLockRelease(CLogControlLock);
> +	SimpleLruFlush(ClogCtl, false);
>  }
>  
>  /*
> @@ -834,21 +229,7 @@
>  void
>  CheckPointCLOG(void)
>  {
> -	int			slotno;
> -
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -
> -	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -	{
> -		WriteCLOGPage(slotno);
> -
> -		/*
> -		 * We cannot assert that the slot is clean now, since another
> -		 * process might have re-dirtied it already.  That's okay.
> -		 */
> -	}
> -
> -	LWLockRelease(CLogControlLock);
> +	SimpleLruFlush(ClogCtl, true);
>  }
>  
>  
> @@ -875,12 +256,12 @@
>  
>  	pageno = TransactionIdToPage(newestXact);
>  
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> +	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
>  
>  	/* Zero the page and make an XLOG entry about it */
>  	ZeroCLOGPage(pageno, true);
>  
> -	LWLockRelease(CLogControlLock);
> +	LWLockRelease(ClogCtl->ControlLock);
>  }
>  
>  
> @@ -902,126 +283,15 @@
>  TruncateCLOG(TransactionId oldestXact)
>  {
>  	int			cutoffPage;
> -	int			slotno;
>  
>  	/*
>  	 * The cutoff point is the start of the segment containing oldestXact.
> +	 * We pass the *page* containing oldestXact to SimpleLruTruncate.
>  	 */
> -	oldestXact -= oldestXact % CLOG_XACTS_PER_SEGMENT;
>  	cutoffPage = TransactionIdToPage(oldestXact);
> -
> -	if (!ScanCLOGDirectory(cutoffPage, false))
> -		return;					/* nothing to remove */
> -
> -	/* Perform a forced CHECKPOINT */
> -	CreateCheckPoint(false, true);
> -
> -	/*
> -	 * Scan CLOG shared memory and remove any pages preceding the cutoff
> -	 * page, to ensure we won't rewrite them later.  (Any dirty pages
> -	 * should have been flushed already during the checkpoint, we're just
> -	 * being extra careful here.)
> -	 */
> -	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> -
> -restart:;
> -
> -	/*
> -	 * While we are holding the lock, make an important safety check: the
> -	 * planned cutoff point must be <= the current CLOG endpoint page.
> -	 * Otherwise we have already wrapped around, and proceeding with the
> -	 * truncation would risk removing the current CLOG segment.
> -	 */
> -	if (CLOGPagePrecedes(ClogCtl->latest_page_number, cutoffPage))
> -	{
> -		LWLockRelease(CLogControlLock);
> -		elog(LOG, "unable to truncate commit log: apparent wraparound");
> -		return;
> -	}
> -
> -	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> -	{
> -		if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
> -			continue;
> -		if (!CLOGPagePrecedes(ClogCtl->page_number[slotno], cutoffPage))
> -			continue;
> -
> -		/*
> -		 * If page is CLEAN, just change state to EMPTY (expected case).
> -		 */
> -		if (ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN)
> -		{
> -			ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
> -			continue;
> -		}
> -
> -		/*
> -		 * 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 SelectLRUCLOGPage.
> -		 */
> -		if (ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS)
> -			(void) ReadCLOGPage(ClogCtl->page_number[slotno],
> -								InvalidTransactionId);
> -		else
> -			WriteCLOGPage(slotno);
> -		goto restart;
> -	}
> -
> -	LWLockRelease(CLogControlLock);
> -
> -	/* Now we can remove the old CLOG segment(s) */
> -	(void) ScanCLOGDirectory(cutoffPage, true);
> +	SimpleLruTruncate(ClogCtl, cutoffPage);
>  }
>  
> -/*
> - * TruncateCLOG subroutine: scan CLOG directory for removable segments.
> - * Actually remove them iff doDeletions is true.  Return TRUE iff any
> - * removable segments were found.  Note: no locking is needed.
> - */
> -static bool
> -ScanCLOGDirectory(int cutoffPage, bool doDeletions)
> -{
> -	bool		found = false;
> -	DIR		   *cldir;
> -	struct dirent *clde;
> -	int			segno;
> -	int			segpage;
> -	char		path[MAXPGPATH];
> -
> -	cldir = opendir(ClogDir);
> -	if (cldir == NULL)
> -		elog(ERROR, "could not open transaction-commit log directory (%s): %m",
> -			 ClogDir);
> -
> -	errno = 0;
> -	while ((clde = readdir(cldir)) != NULL)
> -	{
> -		if (strlen(clde->d_name) == 4 &&
> -			strspn(clde->d_name, "0123456789ABCDEF") == 4)
> -		{
> -			segno = (int) strtol(clde->d_name, NULL, 16);
> -			segpage = segno * CLOG_PAGES_PER_SEGMENT;
> -			if (CLOGPagePrecedes(segpage, cutoffPage))
> -			{
> -				found = true;
> -				if (doDeletions)
> -				{
> -					elog(LOG, "removing commit log file %s", clde->d_name);
> -					snprintf(path, MAXPGPATH, "%s/%s", ClogDir, clde->d_name);
> -					unlink(path);
> -				}
> -			}
> -		}
> -		errno = 0;
> -	}
> -	if (errno)
> -		elog(ERROR, "could not read transaction-commit log directory (%s): %m",
> -			 ClogDir);
> -	closedir(cldir);
> -
> -	return found;
> -}
>  
>  /*
>   * Decide which of two CLOG page numbers is "older" for truncation purposes.
> @@ -1081,13 +351,13 @@
>  
>  		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
>  
> -		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
> +		LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
>  
>  		slotno = ZeroCLOGPage(pageno, false);
> -		WriteCLOGPage(slotno);
> -		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
> +		SimpleLruWritePage(ClogCtl, slotno);
> +		/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
>  
> -		LWLockRelease(CLogControlLock);
> +		LWLockRelease(ClogCtl->ControlLock);
>  	}
>  }
>  
> diff -ruN ../base/src/backend/access/transam/slru.c src/backend/access/transam/slru.c
> --- ../base/src/backend/access/transam/slru.c	1970-01-01 01:00:00.000000000 +0100
> +++ src/backend/access/transam/slru.c	2003-05-02 00:15:26.000000000 +0200
> @@ -0,0 +1,850 @@
> +/*-------------------------------------------------------------------------
> + *
> + * slru.c
> + *		Simple LRU
> + *
> + * This module replaces the old "pg_log" access code, which treated pg_log
> + * essentially like a relation, in that it went through the regular buffer
> + * manager.  The problem with that was that there wasn't any good way to
> + * recycle storage space for transactions so old that they'll never be
> + * looked up again.  Now we use specialized access code so that the commit
> + * log can be broken into relatively small, independent segments.
> + *
> + * Portions Copyright (c) 2003, PostgreSQL Global Development Group
> + * Portions Copyright (c) 1994, Regents of the University of California
> + *
> + * $Header: $
> + *
> + *-------------------------------------------------------------------------
> + */
> +#include "postgres.h"
> +
> +#include 
> +#include 
> +#include 
> +#include 
> +#include 
> +
> +#include "access/slru.h"
> +#include "storage/lwlock.h"
> +#include "miscadmin.h"
> +
> +
> +/*
> + * Define segment size.  A page is the same BLCKSZ as is used everywhere
> + * else in Postgres.  The segment size can be chosen somewhat arbitrarily;
> + * we make it 32 pages by default, or 256Kb, i.e. 1M transactions for CLOG
> + * or 64K transactions for SUBTRANS.
> + *
> + * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
> + * page numbering also wraps around at 0xFFFFFFFF/xxxx_XACTS_PER_PAGE (where
> + * xxxx is CLOG or SUBTRANS, respectively), and segment numbering at
> + * 0xFFFFFFFF/xxxx_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.  We need
> + * take no explicit notice of that fact in this module, except when comparing
> + * segment and page numbers in SimpleLruTruncate (see PagePrecedes()).
> + */
> +
> +#define SLRU_PAGES_PER_SEGMENT	32
> +
> +
> +/*----------
> + * Shared-memory data structures for SLRU control
> + *
> + * We use a simple least-recently-used scheme to manage a pool of page
> + * buffers.  Under ordinary circumstances we expect that write
> + * traffic will occur mostly to the latest page (and to the just-prior
> + * page, soon after a page transition).  Read traffic will probably touch
> + * a larger span of pages, but in any case a fairly small number of page
> + * buffers should be sufficient.  So, we just search the buffers using plain
> + * linear search; there's no need for a hashtable or anything fancy.
> + * The management algorithm is straight LRU except that we will never swap
> + * 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.
> + *----------
> + */
> +
> +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;
> +
> +/*
> + * Shared-memory state
> + */
> +typedef struct SlruSharedData
> +{
> +	/*
> +	 * Info for each buffer slot.  Page number is undefined when status is
> +	 * EMPTY.  lru_count is essentially the number of page switches since
> +	 * last use of this page; the page with highest lru_count is the best
> +	 * candidate to replace.
> +	 */
> +	char	   *page_buffer[NUM_CLOG_BUFFERS];
> +	SlruPageStatus page_status[NUM_CLOG_BUFFERS];
> +	int			page_number[NUM_CLOG_BUFFERS];
> +	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
> +
> +	/*
> +	 * latest_page_number is the page number of the current end of the
> +	 * CLOG; this is not critical data, since we use it only to avoid
> +	 * swapping out the latest page.
> +	 */
> +	int			latest_page_number;
> +} SlruSharedData;
> +typedef SlruSharedData *SlruShared;
> +
> +
> +#define SlruFileName(ctl, path, seg) \
> +	snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir, seg)
> +
> +/*
> + * Macro to mark a buffer slot "most recently used".
> + */
> +#define SlruRecentlyUsed(shared, slotno)	\
> +	do { \
> +		if ((shared)->page_lru_count[slotno] != 0) { \
> +			int		iilru; \
> +			for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
> +				(shared)->page_lru_count[iilru]++; \
> +			(shared)->page_lru_count[slotno] = 0; \
> +		} \
> +	} while (0)
> +
> +/* Saved info for SlruReportIOError */
> +typedef enum
> +{
> +	SLRU_OPEN_FAILED,
> +	SLRU_CREATE_FAILED,
> +	SLRU_SEEK_FAILED,
> +	SLRU_READ_FAILED,
> +	SLRU_WRITE_FAILED
> +} SlruErrorCause;
> +static SlruErrorCause slru_errcause;
> +static int	slru_errno;
> +
> +
> +static bool SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno);
> +static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno);
> +static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
> +static int	SlruSelectLRUPage(SlruCtl ctl, int pageno);
> +static bool SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions);
> +
> +
> +/*
> + * Initialization of shared memory
> + */
> +
> +int
> +SimpleLruShmemSize(void)
> +{
> +	return MAXALIGN(sizeof(SlruSharedData)) + BLCKSZ * NUM_CLOG_BUFFERS;
> +}
> +
> +void
> +SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir)
> +{
> +	bool		found;
> +	char	   *bufptr;
> +	int			slotno;
> +	SlruShared	shared;
> +
> +	shared = (SlruShared) ShmemInitStruct(name, SimpleLruShmemSize(), &found);
> +	Assert(!found);
> +
> +	ctl->ControlLock = LWLockAssign();
> +	ctl->shared = shared;
> +	memset(shared, 0, sizeof(SlruSharedData));
> +
> +	bufptr = ((char *) shared) + MAXALIGN(sizeof(SlruSharedData));
> +
> +	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> +	{
> +		shared->page_buffer[slotno] = bufptr;
> +		shared->page_status[slotno] = SLRU_PAGE_EMPTY;
> +		shared->page_lru_count[slotno] = 1;
> +		ctl->BufferLocks[slotno] = LWLockAssign();
> +		bufptr += BLCKSZ;
> +	}
> +
> +	/* shared->latest_page_number will be set later */
> +
> +	/* Init directory path */
> +	snprintf(ctl->Dir, MAXPGPATH, "%s/%s", DataDir, subdir);
> +}
> +
> +/*
> + * Initialize (or reinitialize) a page to zeroes.
> + *
> + * The page is not actually written, just set up in shared memory.
> + * The slot number of the new page is returned.
> + *
> + * Control lock must be held at entry, and will be held at exit.
> + */
> +int
> +SimpleLruZeroPage(SlruCtl ctl, int pageno)
> +{
> +	int			slotno;
> +	SlruShared shared = (SlruShared) ctl->shared;
> +
> +	/* 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 */
> +	MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
> +
> +	/* Assume this page is now the latest active page */
> +	shared->latest_page_number = pageno;
> +
> +	return slotno;
> +}
> +
> +/*
> + * Find a page in a shared buffer, reading it in if necessary.
> + * The page number must correspond to an already-initialized page.
> + *
> + * The passed-in xid is used only for error reporting, and may be
> + * InvalidTransactionId if no specific xid is associated with the action.
> + *
> + * Return value is the shared-buffer address of the page.
> + * The buffer's LRU access info is updated.
> + * If forwrite is true, the buffer is marked as dirty.
> + *
> + * Control lock must be held at entry, and will be held at exit.
> + */
> +char *
> +SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite)
> +{
> +	SlruShared shared = (SlruShared) ctl->shared;
> +
> +	/* Outer loop handles restart if we lose the buffer to someone else */
> +	for (;;)
> +	{
> +		int			slotno;
> +		bool		ok;
> +
> +		/* See if page already is in memory; if not, pick victim slot */
> +		slotno = SlruSelectLRUPage(ctl, pageno);
> +
> +		/* Did we find the page in memory? */
> +		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);
> +				if (forwrite)
> +					shared->page_status[slotno] = SLRU_PAGE_DIRTY;
> +				return shared->page_buffer[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
> +		 * SlruSelectLRUPage from selecting it again for someone else.
> +		 */
> +		SlruRecentlyUsed(shared, slotno);
> +
> +		/* Release shared lock, grab per-buffer lock instead */
> +		LWLockRelease(ctl->ControlLock);
> +		LWLockAcquire(ctl->BufferLocks[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(ctl->BufferLocks[slotno]);
> +			LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
> +			continue;
> +		}
> +
> +		/* Okay, do the read */
> +		ok = SlruPhysicalReadPage(ctl, pageno, slotno);
> +
> +		/* Re-acquire shared control lock and update page state */
> +		LWLockAcquire(ctl->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(ctl->BufferLocks[slotno]);
> +
> +		/* Now it's okay to elog if we failed */
> +		if (!ok)
> +			SlruReportIOError(ctl, pageno, xid);
> +
> +		SlruRecentlyUsed(shared, slotno);
> +		if (forwrite)
> +			shared->page_status[slotno] = SLRU_PAGE_DIRTY;
> +		return shared->page_buffer[slotno];
> +	}
> +}
> +
> +/*
> + * Write a page from a shared buffer, if necessary.
> + * Does nothing if the specified slot is not dirty.
> + *
> + * NOTE: only one write attempt is made here.  Hence, it is possible that
> + * the page is still dirty at exit (if someone else re-dirtied it during
> + * the write).	However, we *do* attempt a fresh write even if the page
> + * is already being written; this is for checkpoints.
> + *
> + * Control lock must be held at entry, and will be held at exit.
> + */
> +void
> +SimpleLruWritePage(SlruCtl ctl, int slotno)
> +{
> +	int			pageno;
> +	bool		ok;
> +	SlruShared shared = (SlruShared) ctl->shared;
> +
> +	/* 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(ctl->ControlLock);
> +	LWLockAcquire(ctl->BufferLocks[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(ctl->BufferLocks[slotno]);
> +		LWLockAcquire(ctl->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);
> +
> +	/* Re-acquire shared control lock and update page state */
> +	LWLockAcquire(ctl->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(ctl->BufferLocks[slotno]);
> +
> +	/* Now it's okay to elog if we failed */
> +	if (!ok)
> +		SlruReportIOError(ctl, pageno, InvalidTransactionId);
> +}
> +
> +/*
> + * Physical read of a (previously existing) page into a buffer slot
> + *
> + * On failure, we cannot just elog(ERROR) since caller has put state in
> + * shared memory that must be undone.  So, we return FALSE and save enough
> + * info in static variables to let SlruReportIOError make the report.
> + *
> + * For now, assume it's not worth keeping a file pointer open across
> + * read/write operations.  We could cache one virtual file pointer ...
> + */
> +static bool
> +SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
> +{
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
> +	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
> +	int			offset = rpageno * BLCKSZ;
> +	char		path[MAXPGPATH];
> +	int			fd;
> +
> +	SlruFileName(ctl, path, segno);
> +
> +	/*
> +	 * In a crash-and-restart situation, it's possible for us to receive
> +	 * commands to set the commit status of transactions whose bits are in
> +	 * already-truncated segments of the commit log (see notes in
> +	 * SlruPhysicalWritePage).	Hence, if we are InRecovery, allow the
> +	 * case where the file doesn't exist, and return zeroes instead.
> +	 */
> +	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
> +	if (fd < 0)
> +	{
> +		if (errno != ENOENT || !InRecovery)
> +		{
> +			slru_errcause = SLRU_OPEN_FAILED;
> +			slru_errno = errno;
> +			return false;
> +		}
> +
> +		elog(LOG, "file %s doesn't exist, reading as zeroes", path);
> +		MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
> +		return true;
> +	}
> +
> +	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
> +	{
> +		slru_errcause = SLRU_SEEK_FAILED;
> +		slru_errno = errno;
> +		return false;
> +	}
> +
> +	errno = 0;
> +	if (read(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
> +	{
> +		slru_errcause = SLRU_READ_FAILED;
> +		slru_errno = errno;
> +		return false;
> +	}
> +
> +	close(fd);
> +	return true;
> +}
> +
> +/*
> + * Physical write of a page from a buffer slot
> + *
> + * On failure, we cannot just elog(ERROR) since caller has put state in
> + * shared memory that must be undone.  So, we return FALSE and save enough
> + * info in static variables to let SlruReportIOError make the report.
> + *
> + * For now, assume it's not worth keeping a file pointer open across
> + * read/write operations.  We could cache one virtual file pointer ...
> + */
> +static bool
> +SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno)
> +{
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
> +	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
> +	int			offset = rpageno * BLCKSZ;
> +	char		path[MAXPGPATH];
> +	int			fd;
> +
> +	SlruFileName(ctl, path, segno);
> +
> +	/*
> +	 * If the file doesn't already exist, we should create it.  It is
> +	 * possible for this to need to happen when writing a page that's not
> +	 * first in its segment; we assume the OS can cope with that.  (Note:
> +	 * it might seem that it'd be okay to create files only when
> +	 * SimpleLruZeroPage is called for the first page of a segment.	However,
> +	 * if after a crash and restart the REDO logic elects to replay the
> +	 * log from a checkpoint before the latest one, then it's possible
> +	 * that we will get commands to set transaction status of transactions
> +	 * that have already been truncated from the commit log.  Easiest way
> +	 * to deal with that is to accept references to nonexistent files here
> +	 * and in SlruPhysicalReadPage.)
> +	 */
> +	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
> +	if (fd < 0)
> +	{
> +		if (errno != ENOENT)
> +		{
> +			slru_errcause = SLRU_OPEN_FAILED;
> +			slru_errno = errno;
> +			return false;
> +		}
> +
> +		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
> +						   S_IRUSR | S_IWUSR);
> +		if (fd < 0)
> +		{
> +			slru_errcause = SLRU_CREATE_FAILED;
> +			slru_errno = errno;
> +			return false;
> +		}
> +	}
> +
> +	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
> +	{
> +		slru_errcause = SLRU_SEEK_FAILED;
> +		slru_errno = errno;
> +		return false;
> +	}
> +
> +	errno = 0;
> +	if (write(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
> +	{
> +		/* if write didn't set errno, assume problem is no disk space */
> +		if (errno == 0)
> +			errno = ENOSPC;
> +		slru_errcause = SLRU_WRITE_FAILED;
> +		slru_errno = errno;
> +		return false;
> +	}
> +
> +	close(fd);
> +	return true;
> +}
> +
> +/*
> + * Issue the error message after failure of SlruPhysicalReadPage or
> + * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
> + */
> +static void
> +SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
> +{
> +	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
> +	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
> +	int			offset = rpageno * BLCKSZ;
> +	char		path[MAXPGPATH];
> +
> +	/* XXX TODO: provide xid as context in error messages */
> +
> +	SlruFileName(ctl, path, segno);
> +	errno = slru_errno;
> +	switch (slru_errcause)
> +	{
> +		case SLRU_OPEN_FAILED:
> +			elog(ERROR, "open of %s failed: %m", path);
> +			break;
> +		case SLRU_CREATE_FAILED:
> +			elog(ERROR, "creation of file %s failed: %m", path);
> +			break;
> +		case SLRU_SEEK_FAILED:
> +			elog(ERROR, "lseek of file %s, offset %u failed: %m",
> +				 path, offset);
> +			break;
> +		case SLRU_READ_FAILED:
> +			elog(ERROR, "read of file %s, offset %u failed: %m",
> +				 path, offset);
> +			break;
> +		case SLRU_WRITE_FAILED:
> +			elog(ERROR, "write of file %s, offset %u failed: %m",
> +				 path, offset);
> +			break;
> +		default:
> +			/* can't get here, we trust */
> +			elog(ERROR, "unknown SimpleLru I/O error");
> +			break;
> +	}
> +}
> +
> +/*
> + * Select the slot to re-use when we need a free slot.
> + *
> + * The target page number is passed because we need to consider the
> + * possibility that some other process reads in the target page while
> + * we are doing I/O to free a slot.  Hence, check or recheck to see if
> + * any slot already holds the target page, and return that slot if so.
> + * Thus, the returned slot is *either* a slot already holding the pageno
> + * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
> + * or CLEAN).
> + *
> + * Control lock must be held at entry, and will be held at exit.
> + */
> +static int
> +SlruSelectLRUPage(SlruCtl ctl, int pageno)
> +{
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +	/* Outer loop handles restart after I/O */
> +	for (;;)
> +	{
> +		int			slotno;
> +		int			bestslot = 0;
> +		unsigned int bestcount = 0;
> +
> +		/* See if page already has a buffer assigned */
> +		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> +		{
> +			if (shared->page_number[slotno] == pageno &&
> +				shared->page_status[slotno] != SLRU_PAGE_EMPTY)
> +				return slotno;
> +		}
> +
> +		/*
> +		 * If we find any EMPTY slot, just select that one. Else locate
> +		 * the least-recently-used slot that isn't the latest page.
> +		 */
> +		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> +		{
> +			if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
> +				return slotno;
> +			if (shared->page_lru_count[slotno] > bestcount &&
> +			    shared->page_number[slotno] != shared->latest_page_number)
> +			{
> +				bestslot = slotno;
> +				bestcount = shared->page_lru_count[slotno];
> +			}
> +		}
> +
> +		/*
> +		 * 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, false);
> +		else
> +			SimpleLruWritePage(ctl, bestslot);
> +
> +		/*
> +		 * Now loop back and try again.  This is the easiest way of
> +		 * dealing with corner cases such as the victim page being
> +		 * re-dirtied while we wrote it.
> +		 */
> +	}
> +}
> +
> +/*
> + * This must be called ONCE during postmaster or standalone-backend startup
> + */
> +void
> +SimpleLruSetLatestPage(SlruCtl ctl, int pageno)
> +{
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +
> +	shared->latest_page_number = pageno;
> +}
> +
> +/*
> + * This is called during checkpoint and postmaster/standalone-backend shutdown
> + */
> +void
> +SimpleLruFlush(SlruCtl ctl, bool checkpoint)
> +{
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +	int			slotno;
> +
> +	LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
> +
> +	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> +	{
> +		SimpleLruWritePage(ctl, slotno);
> +		/*
> +		 * When called during a checkpoint,
> +		 * we cannot assert that the slot is clean now, since another
> +		 * process might have re-dirtied it already.  That's okay.
> +		 */
> +		Assert(checkpoint ||
> +			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
> +			   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
> +	}
> +
> +	LWLockRelease(ctl->ControlLock);
> +}
> +
> +/*
> + * Remove all segments before the one holding the passed page number
> + *
> + * When this is called, we know that the database logically contains no
> + * reference to transaction IDs older than oldestXact.	However, we must
> + * not remove any segment until we have performed a checkpoint, to ensure
> + * that no such references remain on disk either; else a crash just after
> + * the truncation might leave us with a problem.  Since CLOG segments hold
> + * a large number of transactions, the opportunity to actually remove a
> + * segment is fairly rare, and so it seems best not to do the checkpoint
> + * unless we have confirmed that there is a removable segment.	Therefore
> + * we issue the checkpoint command here, not in higher-level code as might
> + * seem cleaner.
> + */
> +void
> +SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
> +{
> +	int			slotno;
> +	SlruShared	shared = (SlruShared) ctl->shared;
> +
> +	/*
> +	 * The cutoff point is the start of the segment containing cutoffPage.
> +	 */
> +	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;
> +
> +	if (!SlruScanDirectory(ctl, cutoffPage, false))
> +		return;					/* nothing to remove */
> +
> +	/* Perform a forced CHECKPOINT */
> +	CreateCheckPoint(false, true);
> +
> +	/*
> +	 * Scan shared memory and remove any pages preceding the cutoff
> +	 * page, to ensure we won't rewrite them later.  (Any dirty pages
> +	 * should have been flushed already during the checkpoint, we're just
> +	 * being extra careful here.)
> +	 */
> +	LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
> +
> +restart:;
> +
> +	/*
> +	 * While we are holding the lock, make an important safety check: the
> +	 * planned cutoff point must be <= the current endpoint page.
> +	 * Otherwise we have already wrapped around, and proceeding with the
> +	 * truncation would risk removing the current segment.
> +	 */
> +	if (ctl->PagePrecedes(shared->latest_page_number, cutoffPage))
> +	{
> +		LWLockRelease(ctl->ControlLock);
> +		elog(LOG, "unable to truncate %s: apparent wraparound", ctl->Dir);
> +		return;
> +	}
> +
> +	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
> +	{
> +		if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
> +			continue;
> +		if (!ctl->PagePrecedes(shared->page_number[slotno], cutoffPage))
> +			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;
> +		}
> +
> +		/*
> +		 * 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, false);
> +		else
> +			SimpleLruWritePage(ctl, slotno);
> +		goto restart;
> +	}
> +
> +	LWLockRelease(ctl->ControlLock);
> +
> +	/* Now we can remove the old segment(s) */
> +	(void) SlruScanDirectory(ctl, cutoffPage, true);
> +}
> +
> +/*
> + * SlruTruncate subroutine: scan directory for removable segments.
> + * Actually remove them iff doDeletions is true.  Return TRUE iff any
> + * removable segments were found.  Note: no locking is needed.
> + */
> +static bool
> +SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
> +{
> +	bool		found = false;
> +	DIR		   *cldir;
> +	struct dirent *clde;
> +	int			segno;
> +	int			segpage;
> +	char		path[MAXPGPATH];
> +
> +	cldir = opendir(ctl->Dir);
> +	if (cldir == NULL)
> +		elog(ERROR, "could not open directory (%s): %m", ctl->Dir);
> +
> +	errno = 0;
> +	while ((clde = readdir(cldir)) != NULL)
> +	{
> +		if (strlen(clde->d_name) == 4 &&
> +			strspn(clde->d_name, "0123456789ABCDEF") == 4)
> +		{
> +			segno = (int) strtol(clde->d_name, NULL, 16);
> +			segpage = segno * SLRU_PAGES_PER_SEGMENT;
> +			if (ctl->PagePrecedes(segpage, cutoffPage))
> +			{
> +				found = true;
> +				if (doDeletions)
> +				{
> +					elog(LOG, "removing file %s/%s", ctl->Dir, clde->d_name);
> +					snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, clde->d_name);
> +					unlink(path);
> +				}
> +			}
> +		}
> +		errno = 0;
> +	}
> +	if (errno)
> +		elog(ERROR, "could not read directory (%s): %m", ctl->Dir);
> +	closedir(cldir);
> +
> +	return found;
> +}
> diff -ruN ../base/src/backend/storage/lmgr/lwlock.c src/backend/storage/lmgr/lwlock.c
> --- ../base/src/backend/storage/lmgr/lwlock.c	2002-10-10 19:07:27.000000000 +0200
> +++ src/backend/storage/lmgr/lwlock.c	2003-05-01 21:31:34.000000000 +0200
> @@ -108,8 +108,8 @@
>  	/* bufmgr.c needs two for each shared buffer */
>  	numLocks += 2 * NBuffers;
>  
> -	/* clog.c needs one per CLOG buffer */
> -	numLocks += NUM_CLOG_BUFFERS;
> +	/* clog.c needs one per CLOG buffer + one control lock */
> +	numLocks += NUM_CLOG_BUFFERS + 1;
>  
>  	/* Perhaps create a few more for use by user-defined modules? */
>  
> diff -ruN ../base/src/include/access/slru.h src/include/access/slru.h
> --- ../base/src/include/access/slru.h	1970-01-01 01:00:00.000000000 +0100
> +++ src/include/access/slru.h	2003-05-01 20:50:14.000000000 +0200
> @@ -0,0 +1,57 @@
> +/*
> + * slru.h
> + *
> + * Simple LRU
> + *
> + * Portions Copyright (c) 2003, PostgreSQL Global Development Group
> + * Portions Copyright (c) 1994, Regents of the University of California
> + *
> + * $Id: $
> + */
> +#ifndef SLRU_H
> +#define SLRU_H
> +
> +#include "access/xlog.h"
> +
> +/* exported because lwlock.c needs it */
> +#define NUM_CLOG_BUFFERS	8
> +
> +
> +typedef struct SlruCtlData
> +{
> +	void *shared;	/* pointer to SlruSharedData */
> +	LWLockId ControlLock;   // = CLogControlLock
> +/*
> + * BufferLocks is set during CLOGShmemInit and does not change thereafter.
> + * The value is automatically inherited by backends via fork, and
> + * doesn't need to be in shared memory.
> + */
> +	LWLockId BufferLocks[NUM_CLOG_BUFFERS];		/* Per-buffer I/O locks */
> +
> +/*
> + * Dir is set during SimpleLruShmemInit and does not change thereafter.
> + * The value is automatically inherited by backends via fork, and
> + * doesn't need to be in shared memory.
> + */
> +	char Dir[MAXPGPATH];
> +	
> +/*
> + * Decide which of two page numbers is "older" for truncation purposes.
> + * We need to use comparison of TransactionIds here in order to do the right
> + * thing with wraparound XID arithmetic.
> + */
> +	bool (*PagePrecedes)(int, int);
> +
> +} SlruCtlData;
> +typedef SlruCtlData *SlruCtl;
> +
> +extern int	SimpleLruShmemSize(void);
> +extern void SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir);
> +extern int	SimpleLruZeroPage(SlruCtl ctl, int pageno);
> +extern char	*SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite);
> +extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
> +extern void SimpleLruSetLatestPage(SlruCtl ctl, int pageno);
> +extern void SimpleLruFlush(SlruCtl ctl, bool checkpoint);
> +extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
> +
> +#endif   /* SLRU_H */
> diff -ruN ../base/src/include/storage/lwlock.h src/include/storage/lwlock.h
> --- ../base/src/include/storage/lwlock.h	2002-06-21 02:12:30.000000000 +0200
> +++ src/include/storage/lwlock.h	2003-05-01 21:32:21.000000000 +0200
> @@ -37,7 +37,6 @@
>  	WALWriteLock,
>  	ControlFileLock,
>  	CheckpointLock,
> -	CLogControlLock,
>  	RelCacheInitLock,
>  
>  	NumFixedLWLocks,			/* must be last except for

> 
> ---------------------------(end of broadcast)---------------------------
> TIP 4: Don't 'kill -9' the postmaster

-- 
  Bruce Momjian                        |  http://candle.pha.pa.us
  pgman@candle.pha.pa.us               |  (610) 359-1001
  +  If your life is a hard drive,     |  13 Roberts Road
  +  Christ can be your backup.        |  Newtown Square, Pennsylvania 19073

Simple LRU for clog and (later) subtrans

От:

Manfred Koizar <mkoi-pg@aon.at>

Дата:

2 мая 2003 г. в 06:02:12

This patch extracts page buffer pooling and the simple
least-recently-used strategy from clog.c into slru.c.  It doesn't
change any visible behaviour and passes all regression tests plus a
TruncateCLOG test done manually.

Apart from refactoring I made a little change to SlruRecentlyUsed,
formerly ClogRecentlyUsed:  It now skips incrementing lru_counts, if
slotno is already the LRU slot, thus saving a few CPU cycles.  To make
this work, lru_counts are initialised to 1 in SimpleLruInit.

SimpleLru will be used by pg_subtrans (part of the nested transactions
project), so the main purpose of this patch is to avoid future code
duplication.

Servus
 Manfred
diff -ruN ../base/src/backend/access/transam/Makefile src/backend/access/transam/Makefile
--- ../base/src/backend/access/transam/Makefile	2001-08-25 20:52:41.000000000 +0200
+++ src/backend/access/transam/Makefile	2003-05-01 20:11:44.000000000 +0200
@@ -12,7 +12,7 @@
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o
+OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o slru.o
 
 all: SUBSYS.o
 
diff -ruN ../base/src/backend/access/transam/clog.c src/backend/access/transam/clog.c
--- ../base/src/backend/access/transam/clog.c	2003-04-18 19:09:07.000000000 +0200
+++ src/backend/access/transam/clog.c	2003-05-02 00:20:07.000000000 +0200
@@ -26,15 +26,14 @@
 #include 
 
 #include "access/clog.h"
+#include "access/slru.h"
 #include "storage/lwlock.h"
 #include "miscadmin.h"
 
 
 /*
  * Defines for CLOG page and segment sizes.  A page is the same BLCKSZ
- * as is used everywhere else in Postgres.	The CLOG segment size can be
- * chosen somewhat arbitrarily; we make it 1 million transactions by default,
- * or 256Kb.
+ * as is used everywhere else in Postgres.
  *
  * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
  * CLOG page numbering also wraps around at 0xFFFFFFFF/CLOG_XACTS_PER_PAGE,
@@ -43,17 +42,12 @@
  * segment and page numbers in TruncateCLOG (see CLOGPagePrecedes).
  */
 
-#define CLOG_BLCKSZ			BLCKSZ
-
 /* We need two bits per xact, so four xacts fit in a byte */
 #define CLOG_BITS_PER_XACT	2
 #define CLOG_XACTS_PER_BYTE 4
-#define CLOG_XACTS_PER_PAGE (CLOG_BLCKSZ * CLOG_XACTS_PER_BYTE)
+#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
 #define CLOG_XACT_BITMASK	((1 << CLOG_BITS_PER_XACT) - 1)
 
-#define CLOG_XACTS_PER_SEGMENT	0x100000
-#define CLOG_PAGES_PER_SEGMENT	(CLOG_XACTS_PER_SEGMENT / CLOG_XACTS_PER_PAGE)
-
 #define TransactionIdToPage(xid)	((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToByte(xid)	(TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
@@ -63,47 +57,6 @@
 /*----------
  * Shared-memory data structures for CLOG control
  *
- * We use a simple least-recently-used scheme to manage a pool of page
- * buffers for the CLOG.  Under ordinary circumstances we expect that write
- * traffic will occur mostly to the latest CLOG page (and to the just-prior
- * page, soon after a page transition).  Read traffic will probably touch
- * a larger span of pages, but in any case a fairly small number of page
- * buffers should be sufficient.  So, we just search the buffers using plain
- * linear search; there's no need for a hashtable or anything fancy.
- * The management algorithm is straight LRU except that we will never swap
- * out the latest page (since we know it's going to be hit again eventually).
- *
- * We use an overall 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.
- *
  * XLOG interactions: this module generates an XLOG record whenever a new
  * CLOG page is initialized to zeroes.	Other writes of CLOG come from
  * recording of transaction commit or abort in xact.c, which generates its
@@ -117,90 +70,12 @@
  *----------
  */
 
-typedef enum
-{
-	CLOG_PAGE_EMPTY,			/* CLOG buffer is not in use */
-	CLOG_PAGE_READ_IN_PROGRESS, /* CLOG page is being read in */
-	CLOG_PAGE_CLEAN,			/* CLOG page is valid and not dirty */
-	CLOG_PAGE_DIRTY,			/* CLOG page is valid but needs write */
-	CLOG_PAGE_WRITE_IN_PROGRESS /* CLOG page is being written out */
-} ClogPageStatus;
-
-/*
- * Shared-memory state for CLOG.
- */
-typedef struct ClogCtlData
-{
-	/*
-	 * Info for each buffer slot.  Page number is undefined when status is
-	 * EMPTY.  lru_count is essentially the number of operations since
-	 * last use of this page; the page with highest lru_count is the best
-	 * candidate to replace.
-	 */
-	char	   *page_buffer[NUM_CLOG_BUFFERS];
-	ClogPageStatus page_status[NUM_CLOG_BUFFERS];
-	int			page_number[NUM_CLOG_BUFFERS];
-	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
-
-	/*
-	 * latest_page_number is the page number of the current end of the
-	 * CLOG; this is not critical data, since we use it only to avoid
-	 * swapping out the latest page.
-	 */
-	int			latest_page_number;
-} ClogCtlData;
-
-static ClogCtlData *ClogCtl = NULL;
-
-/*
- * ClogBufferLocks is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
- */
-static LWLockId ClogBufferLocks[NUM_CLOG_BUFFERS];		/* Per-buffer I/O locks */
-
-/*
- * ClogDir is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
- */
-static char ClogDir[MAXPGPATH];
-
-#define ClogFileName(path, seg) \
-	snprintf(path, MAXPGPATH, "%s/%04X", ClogDir, seg)
-
-/*
- * Macro to mark a buffer slot "most recently used".
- */
-#define ClogRecentlyUsed(slotno)	\
-	do { \
-		int		iilru; \
-		for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
-			ClogCtl->page_lru_count[iilru]++; \
-		ClogCtl->page_lru_count[slotno] = 0; \
-	} while (0)
-
-/* Saved info for CLOGReportIOError */
-typedef enum
-{
-	CLOG_OPEN_FAILED,
-	CLOG_CREATE_FAILED,
-	CLOG_SEEK_FAILED,
-	CLOG_READ_FAILED,
-	CLOG_WRITE_FAILED
-} ClogErrorCause;
-static ClogErrorCause clog_errcause;
-static int	clog_errno;
 
+static SlruCtlData ClogCtlData;
+static SlruCtl ClogCtl = &ClogCtlData;
+  
 
 static int	ZeroCLOGPage(int pageno, bool writeXlog);
-static int	ReadCLOGPage(int pageno, TransactionId xid);
-static void WriteCLOGPage(int slotno);
-static bool CLOGPhysicalReadPage(int pageno, int slotno);
-static bool CLOGPhysicalWritePage(int pageno, int slotno);
-static void CLOGReportIOError(int pageno, TransactionId xid);
-static int	SelectLRUCLOGPage(int pageno);
-static bool ScanCLOGDirectory(int cutoffPage, bool doDeletions);
 static bool CLOGPagePrecedes(int page1, int page2);
 static void WriteZeroPageXlogRec(int pageno);
 
@@ -217,16 +92,15 @@
 	int			pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);
 	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
-	int			slotno;
 	char	   *byteptr;
 
 	Assert(status == TRANSACTION_STATUS_COMMITTED ||
 		   status == TRANSACTION_STATUS_ABORTED);
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
-	slotno = ReadCLOGPage(pageno, xid);
-	byteptr = ClogCtl->page_buffer[slotno] + byteno;
+	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, true);
+	byteptr += byteno;
 
 	/* Current state should be 0 or target state */
 	Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
@@ -234,9 +108,9 @@
 
 	*byteptr |= (status << bshift);
 
-	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
+	/* ...->page_status[slotno] = CLOG_PAGE_DIRTY; already done */
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->ControlLock);
 }
 
 /*
@@ -251,18 +125,17 @@
 	int			pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);
 	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
-	int			slotno;
 	char	   *byteptr;
 	XidStatus	status;
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
-	slotno = ReadCLOGPage(pageno, xid);
-	byteptr = ClogCtl->page_buffer[slotno] + byteno;
+	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, false);
+	byteptr += byteno;
 
 	status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->ControlLock);
 
 	return status;
 }
@@ -275,40 +148,14 @@
 int
 CLOGShmemSize(void)
 {
-	return MAXALIGN(sizeof(ClogCtlData) + CLOG_BLCKSZ * NUM_CLOG_BUFFERS);
+	return SimpleLruShmemSize();
 }
 
 void
 CLOGShmemInit(void)
 {
-	bool		found;
-	char	   *bufptr;
-	int			slotno;
-
-	/* this must agree with space requested by CLOGShmemSize() */
-	ClogCtl = (ClogCtlData *)
-		ShmemInitStruct("CLOG Ctl",
-						MAXALIGN(sizeof(ClogCtlData) +
-								 CLOG_BLCKSZ * NUM_CLOG_BUFFERS),
-						&found);
-	Assert(!found);
-
-	memset(ClogCtl, 0, sizeof(ClogCtlData));
-
-	bufptr = ((char *) ClogCtl) + sizeof(ClogCtlData);
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		ClogCtl->page_buffer[slotno] = bufptr;
-		ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
-		ClogBufferLocks[slotno] = LWLockAssign();
-		bufptr += CLOG_BLCKSZ;
-	}
-
-	/* ClogCtl->latest_page_number will be set later */
-
-	/* Init CLOG directory path */
-	snprintf(ClogDir, MAXPGPATH, "%s/pg_clog", DataDir);
+	SimpleLruInit(ClogCtl, "CLOG Ctl", "pg_clog");
+	ClogCtl->PagePrecedes = CLOGPagePrecedes;
 }
 
 /*
@@ -322,16 +169,16 @@
 {
 	int			slotno;
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
 	/* Create and zero the first page of the commit log */
 	slotno = ZeroCLOGPage(0, false);
 
 	/* Make sure it's written out */
-	WriteCLOGPage(slotno);
-	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+	SimpleLruWritePage(ClogCtl, slotno);
+	/* Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN); */
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->ControlLock);
 }
 
 /*
@@ -346,24 +193,7 @@
 static int
 ZeroCLOGPage(int pageno, bool writeXlog)
 {
-	int			slotno;
-
-	/* Find a suitable buffer slot for the page */
-	slotno = SelectLRUCLOGPage(pageno);
-	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-		   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN ||
-		   ClogCtl->page_number[slotno] == pageno);
-
-	/* Mark the slot as containing this page */
-	ClogCtl->page_number[slotno] = pageno;
-	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
-	ClogRecentlyUsed(slotno);
-
-	/* Set the buffer to zeroes */
-	MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
-
-	/* Assume this page is now the latest active page */
-	ClogCtl->latest_page_number = pageno;
+	int			slotno = SimpleLruZeroPage(ClogCtl, pageno);
 
 	if (writeXlog)
 		WriteZeroPageXlogRec(pageno);
@@ -372,430 +202,6 @@
 }
 
 /*
- * Find a CLOG page in a shared buffer, reading it in if necessary.
- * The page number must correspond to an already-initialized page.
- *
- * The passed-in xid is used only for error reporting, and may be
- * InvalidTransactionId if no specific xid is associated with the action.
- *
- * Return value is the shared-buffer slot number now holding the page.
- * The buffer's LRU access info is updated.
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static int
-ReadCLOGPage(int pageno, TransactionId xid)
-{
-	/* Outer loop handles restart if we lose the buffer to someone else */
-	for (;;)
-	{
-		int			slotno;
-		bool		ok;
-
-		/* See if page already is in memory; if not, pick victim slot */
-		slotno = SelectLRUCLOGPage(pageno);
-
-		/* Did we find the page in memory? */
-		if (ClogCtl->page_number[slotno] == pageno &&
-			ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
-		{
-			/* If page is still being read in, we cannot use it yet */
-			if (ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
-			{
-				/* otherwise, it's ready to use */
-				ClogRecentlyUsed(slotno);
-				return slotno;
-			}
-		}
-		else
-		{
-			/* We found no match; assert we selected a freeable slot */
-			Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-				   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
-		}
-
-		/* Mark the slot read-busy (no-op if it already was) */
-		ClogCtl->page_number[slotno] = pageno;
-		ClogCtl->page_status[slotno] = CLOG_PAGE_READ_IN_PROGRESS;
-
-		/*
-		 * Temporarily mark page as recently-used to discourage
-		 * SelectLRUCLOGPage from selecting it again for someone else.
-		 */
-		ClogCtl->page_lru_count[slotno] = 0;
-
-		/* Release shared lock, grab per-buffer lock instead */
-		LWLockRelease(CLogControlLock);
-		LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
-			ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
-		{
-			LWLockRelease(ClogBufferLocks[slotno]);
-			LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-			continue;
-		}
-
-		/* Okay, do the read */
-		ok = CLOGPhysicalReadPage(pageno, slotno);
-
-		/* Re-acquire shared control lock and update page state */
-		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-		Assert(ClogCtl->page_number[slotno] == pageno &&
-			 ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS);
-
-		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_EMPTY;
-
-		LWLockRelease(ClogBufferLocks[slotno]);
-
-		/* Now it's okay to elog if we failed */
-		if (!ok)
-			CLOGReportIOError(pageno, xid);
-
-		ClogRecentlyUsed(slotno);
-		return slotno;
-	}
-}
-
-/*
- * Write a CLOG page from a shared buffer, if necessary.
- * Does nothing if the specified slot is not dirty.
- *
- * NOTE: only one write attempt is made here.  Hence, it is possible that
- * the page is still dirty at exit (if someone else re-dirtied it during
- * the write).	However, we *do* attempt a fresh write even if the page
- * is already being written; this is for checkpoints.
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static void
-WriteCLOGPage(int slotno)
-{
-	int			pageno;
-	bool		ok;
-
-	/* Do nothing if page does not need writing */
-	if (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
-		ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS)
-		return;
-
-	pageno = ClogCtl->page_number[slotno];
-
-	/* Release shared lock, grab per-buffer lock instead */
-	LWLockRelease(CLogControlLock);
-	LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
-		(ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
-		 ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS))
-	{
-		LWLockRelease(ClogBufferLocks[slotno]);
-		LWLockAcquire(CLogControlLock, 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.)
-	 */
-	ClogCtl->page_status[slotno] = CLOG_PAGE_WRITE_IN_PROGRESS;
-
-	/* Okay, do the write */
-	ok = CLOGPhysicalWritePage(pageno, slotno);
-
-	/* Re-acquire shared control lock and update page state */
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	Assert(ClogCtl->page_number[slotno] == pageno &&
-		   (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS ||
-			ClogCtl->page_status[slotno] == CLOG_PAGE_DIRTY));
-
-	/* Cannot set CLEAN if someone re-dirtied page since write started */
-	if (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS)
-		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_DIRTY;
-
-	LWLockRelease(ClogBufferLocks[slotno]);
-
-	/* Now it's okay to elog if we failed */
-	if (!ok)
-		CLOGReportIOError(pageno, InvalidTransactionId);
-}
-
-/*
- * Physical read of a (previously existing) page into a buffer slot
- *
- * On failure, we cannot just elog(ERROR) since caller has put state in
- * shared memory that must be undone.  So, we return FALSE and save enough
- * info in static variables to let CLOGReportIOError make the report.
- *
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations.  We could cache one virtual file pointer ...
- */
-static bool
-CLOGPhysicalReadPage(int pageno, int slotno)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-	int			fd;
-
-	ClogFileName(path, segno);
-
-	/*
-	 * In a crash-and-restart situation, it's possible for us to receive
-	 * commands to set the commit status of transactions whose bits are in
-	 * already-truncated segments of the commit log (see notes in
-	 * CLOGPhysicalWritePage).	Hence, if we are InRecovery, allow the
-	 * case where the file doesn't exist, and return zeroes instead.
-	 */
-	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
-	if (fd < 0)
-	{
-		if (errno != ENOENT || !InRecovery)
-		{
-			clog_errcause = CLOG_OPEN_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-
-		elog(LOG, "clog file %s doesn't exist, reading as zeroes", path);
-		MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
-		return true;
-	}
-
-	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
-	{
-		clog_errcause = CLOG_SEEK_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	errno = 0;
-	if (read(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
-	{
-		clog_errcause = CLOG_READ_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	close(fd);
-	return true;
-}
-
-/*
- * Physical write of a page from a buffer slot
- *
- * On failure, we cannot just elog(ERROR) since caller has put state in
- * shared memory that must be undone.  So, we return FALSE and save enough
- * info in static variables to let CLOGReportIOError make the report.
- *
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations.  We could cache one virtual file pointer ...
- */
-static bool
-CLOGPhysicalWritePage(int pageno, int slotno)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-	int			fd;
-
-	ClogFileName(path, segno);
-
-	/*
-	 * If the file doesn't already exist, we should create it.  It is
-	 * possible for this to need to happen when writing a page that's not
-	 * first in its segment; we assume the OS can cope with that.  (Note:
-	 * it might seem that it'd be okay to create files only when
-	 * ZeroCLOGPage is called for the first page of a segment.	However,
-	 * if after a crash and restart the REDO logic elects to replay the
-	 * log from a checkpoint before the latest one, then it's possible
-	 * that we will get commands to set transaction status of transactions
-	 * that have already been truncated from the commit log.  Easiest way
-	 * to deal with that is to accept references to nonexistent files here
-	 * and in CLOGPhysicalReadPage.)
-	 */
-	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
-	if (fd < 0)
-	{
-		if (errno != ENOENT)
-		{
-			clog_errcause = CLOG_OPEN_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-
-		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
-						   S_IRUSR | S_IWUSR);
-		if (fd < 0)
-		{
-			clog_errcause = CLOG_CREATE_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-	}
-
-	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
-	{
-		clog_errcause = CLOG_SEEK_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	errno = 0;
-	if (write(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
-	{
-		/* if write didn't set errno, assume problem is no disk space */
-		if (errno == 0)
-			errno = ENOSPC;
-		clog_errcause = CLOG_WRITE_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	close(fd);
-	return true;
-}
-
-/*
- * Issue the error message after failure of CLOGPhysicalReadPage or
- * CLOGPhysicalWritePage.  Call this after cleaning up shared-memory state.
- */
-static void
-CLOGReportIOError(int pageno, TransactionId xid)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-
-	/* XXX TODO: provide xid as context in error messages */
-
-	ClogFileName(path, segno);
-	errno = clog_errno;
-	switch (clog_errcause)
-	{
-		case CLOG_OPEN_FAILED:
-			elog(ERROR, "open of %s failed: %m", path);
-			break;
-		case CLOG_CREATE_FAILED:
-			elog(ERROR, "creation of file %s failed: %m", path);
-			break;
-		case CLOG_SEEK_FAILED:
-			elog(ERROR, "lseek of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		case CLOG_READ_FAILED:
-			elog(ERROR, "read of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		case CLOG_WRITE_FAILED:
-			elog(ERROR, "write of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		default:
-			/* can't get here, we trust */
-			elog(ERROR, "unknown CLOG I/O error");
-			break;
-	}
-}
-
-/*
- * Select the slot to re-use when we need a free slot.
- *
- * The target page number is passed because we need to consider the
- * possibility that some other process reads in the target page while
- * we are doing I/O to free a slot.  Hence, check or recheck to see if
- * any slot already holds the target page, and return that slot if so.
- * Thus, the returned slot is *either* a slot already holding the pageno
- * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
- * or CLEAN).
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static int
-SelectLRUCLOGPage(int pageno)
-{
-	/* Outer loop handles restart after I/O */
-	for (;;)
-	{
-		int			slotno;
-		int			bestslot = 0;
-		unsigned int bestcount = 0;
-
-		/* See if page already has a buffer assigned */
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-		{
-			if (ClogCtl->page_number[slotno] == pageno &&
-				ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
-				return slotno;
-		}
-
-		/*
-		 * If we find any EMPTY slot, just select that one. Else locate
-		 * the least-recently-used slot that isn't the latest CLOG page.
-		 */
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-		{
-			if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
-				return slotno;
-			if (ClogCtl->page_lru_count[slotno] > bestcount &&
-			 ClogCtl->page_number[slotno] != ClogCtl->latest_page_number)
-			{
-				bestslot = slotno;
-				bestcount = ClogCtl->page_lru_count[slotno];
-			}
-		}
-
-		/*
-		 * If the selected page is clean, we're set.
-		 */
-		if (ClogCtl->page_status[bestslot] == CLOG_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 ReadCLOGPage to wait for
-		 * the read to complete.
-		 */
-		if (ClogCtl->page_status[bestslot] == CLOG_PAGE_READ_IN_PROGRESS)
-			(void) ReadCLOGPage(ClogCtl->page_number[bestslot],
-								InvalidTransactionId);
-		else
-			WriteCLOGPage(bestslot);
-
-		/*
-		 * Now loop back and try again.  This is the easiest way of
-		 * dealing with corner cases such as the victim page being
-		 * re-dirtied while we wrote it.
-		 */
-	}
-}
-
-/*
  * This must be called ONCE during postmaster or standalone-backend startup,
  * after StartupXLOG has initialized ShmemVariableCache->nextXid.
  */
@@ -805,7 +211,7 @@
 	/*
 	 * Initialize our idea of the latest page number.
 	 */
-	ClogCtl->latest_page_number = TransactionIdToPage(ShmemVariableCache->nextXid);
+	SimpleLruSetLatestPage(ClogCtl, TransactionIdToPage(ShmemVariableCache->nextXid));
 }
 
 /*
@@ -814,18 +220,7 @@
 void
 ShutdownCLOG(void)
 {
-	int			slotno;
-
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		WriteCLOGPage(slotno);
-		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-			   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
-	}
-
-	LWLockRelease(CLogControlLock);
+	SimpleLruFlush(ClogCtl, false);
 }
 
 /*
@@ -834,21 +229,7 @@
 void
 CheckPointCLOG(void)
 {
-	int			slotno;
-
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		WriteCLOGPage(slotno);
-
-		/*
-		 * We cannot assert that the slot is clean now, since another
-		 * process might have re-dirtied it already.  That's okay.
-		 */
-	}
-
-	LWLockRelease(CLogControlLock);
+	SimpleLruFlush(ClogCtl, true);
 }
 
 
@@ -875,12 +256,12 @@
 
 	pageno = TransactionIdToPage(newestXact);
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
 	/* Zero the page and make an XLOG entry about it */
 	ZeroCLOGPage(pageno, true);
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->ControlLock);
 }
 
 
@@ -902,126 +283,15 @@
 TruncateCLOG(TransactionId oldestXact)
 {
 	int			cutoffPage;
-	int			slotno;
 
 	/*
 	 * The cutoff point is the start of the segment containing oldestXact.
+	 * We pass the *page* containing oldestXact to SimpleLruTruncate.
 	 */
-	oldestXact -= oldestXact % CLOG_XACTS_PER_SEGMENT;
 	cutoffPage = TransactionIdToPage(oldestXact);
-
-	if (!ScanCLOGDirectory(cutoffPage, false))
-		return;					/* nothing to remove */
-
-	/* Perform a forced CHECKPOINT */
-	CreateCheckPoint(false, true);
-
-	/*
-	 * Scan CLOG shared memory and remove any pages preceding the cutoff
-	 * page, to ensure we won't rewrite them later.  (Any dirty pages
-	 * should have been flushed already during the checkpoint, we're just
-	 * being extra careful here.)
-	 */
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-restart:;
-
-	/*
-	 * While we are holding the lock, make an important safety check: the
-	 * planned cutoff point must be <= the current CLOG endpoint page.
-	 * Otherwise we have already wrapped around, and proceeding with the
-	 * truncation would risk removing the current CLOG segment.
-	 */
-	if (CLOGPagePrecedes(ClogCtl->latest_page_number, cutoffPage))
-	{
-		LWLockRelease(CLogControlLock);
-		elog(LOG, "unable to truncate commit log: apparent wraparound");
-		return;
-	}
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
-			continue;
-		if (!CLOGPagePrecedes(ClogCtl->page_number[slotno], cutoffPage))
-			continue;
-
-		/*
-		 * If page is CLEAN, just change state to EMPTY (expected case).
-		 */
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN)
-		{
-			ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
-			continue;
-		}
-
-		/*
-		 * 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 SelectLRUCLOGPage.
-		 */
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS)
-			(void) ReadCLOGPage(ClogCtl->page_number[slotno],
-								InvalidTransactionId);
-		else
-			WriteCLOGPage(slotno);
-		goto restart;
-	}
-
-	LWLockRelease(CLogControlLock);
-
-	/* Now we can remove the old CLOG segment(s) */
-	(void) ScanCLOGDirectory(cutoffPage, true);
+	SimpleLruTruncate(ClogCtl, cutoffPage);
 }
 
-/*
- * TruncateCLOG subroutine: scan CLOG directory for removable segments.
- * Actually remove them iff doDeletions is true.  Return TRUE iff any
- * removable segments were found.  Note: no locking is needed.
- */
-static bool
-ScanCLOGDirectory(int cutoffPage, bool doDeletions)
-{
-	bool		found = false;
-	DIR		   *cldir;
-	struct dirent *clde;
-	int			segno;
-	int			segpage;
-	char		path[MAXPGPATH];
-
-	cldir = opendir(ClogDir);
-	if (cldir == NULL)
-		elog(ERROR, "could not open transaction-commit log directory (%s): %m",
-			 ClogDir);
-
-	errno = 0;
-	while ((clde = readdir(cldir)) != NULL)
-	{
-		if (strlen(clde->d_name) == 4 &&
-			strspn(clde->d_name, "0123456789ABCDEF") == 4)
-		{
-			segno = (int) strtol(clde->d_name, NULL, 16);
-			segpage = segno * CLOG_PAGES_PER_SEGMENT;
-			if (CLOGPagePrecedes(segpage, cutoffPage))
-			{
-				found = true;
-				if (doDeletions)
-				{
-					elog(LOG, "removing commit log file %s", clde->d_name);
-					snprintf(path, MAXPGPATH, "%s/%s", ClogDir, clde->d_name);
-					unlink(path);
-				}
-			}
-		}
-		errno = 0;
-	}
-	if (errno)
-		elog(ERROR, "could not read transaction-commit log directory (%s): %m",
-			 ClogDir);
-	closedir(cldir);
-
-	return found;
-}
 
 /*
  * Decide which of two CLOG page numbers is "older" for truncation purposes.
@@ -1081,13 +351,13 @@
 
 		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
 
-		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+		LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
 		slotno = ZeroCLOGPage(pageno, false);
-		WriteCLOGPage(slotno);
-		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+		SimpleLruWritePage(ClogCtl, slotno);
+		/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
 
-		LWLockRelease(CLogControlLock);
+		LWLockRelease(ClogCtl->ControlLock);
 	}
 }
 
diff -ruN ../base/src/backend/access/transam/slru.c src/backend/access/transam/slru.c
--- ../base/src/backend/access/transam/slru.c	1970-01-01 01:00:00.000000000 +0100
+++ src/backend/access/transam/slru.c	2003-05-02 00:15:26.000000000 +0200
@@ -0,0 +1,850 @@
+/*-------------------------------------------------------------------------
+ *
+ * slru.c
+ *		Simple LRU
+ *
+ * This module replaces the old "pg_log" access code, which treated pg_log
+ * essentially like a relation, in that it went through the regular buffer
+ * manager.  The problem with that was that there wasn't any good way to
+ * recycle storage space for transactions so old that they'll never be
+ * looked up again.  Now we use specialized access code so that the commit
+ * log can be broken into relatively small, independent segments.
+ *
+ * Portions Copyright (c) 2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $Header: $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include 
+#include 
+#include 
+#include 
+#include 
+
+#include "access/slru.h"
+#include "storage/lwlock.h"
+#include "miscadmin.h"
+
+
+/*
+ * Define segment size.  A page is the same BLCKSZ as is used everywhere
+ * else in Postgres.  The segment size can be chosen somewhat arbitrarily;
+ * we make it 32 pages by default, or 256Kb, i.e. 1M transactions for CLOG
+ * or 64K transactions for SUBTRANS.
+ *
+ * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
+ * page numbering also wraps around at 0xFFFFFFFF/xxxx_XACTS_PER_PAGE (where
+ * xxxx is CLOG or SUBTRANS, respectively), and segment numbering at
+ * 0xFFFFFFFF/xxxx_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.  We need
+ * take no explicit notice of that fact in this module, except when comparing
+ * segment and page numbers in SimpleLruTruncate (see PagePrecedes()).
+ */
+
+#define SLRU_PAGES_PER_SEGMENT	32
+
+
+/*----------
+ * Shared-memory data structures for SLRU control
+ *
+ * We use a simple least-recently-used scheme to manage a pool of page
+ * buffers.  Under ordinary circumstances we expect that write
+ * traffic will occur mostly to the latest page (and to the just-prior
+ * page, soon after a page transition).  Read traffic will probably touch
+ * a larger span of pages, but in any case a fairly small number of page
+ * buffers should be sufficient.  So, we just search the buffers using plain
+ * linear search; there's no need for a hashtable or anything fancy.
+ * The management algorithm is straight LRU except that we will never swap
+ * 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.
+ *----------
+ */
+
+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;
+
+/*
+ * Shared-memory state
+ */
+typedef struct SlruSharedData
+{
+	/*
+	 * Info for each buffer slot.  Page number is undefined when status is
+	 * EMPTY.  lru_count is essentially the number of page switches since
+	 * last use of this page; the page with highest lru_count is the best
+	 * candidate to replace.
+	 */
+	char	   *page_buffer[NUM_CLOG_BUFFERS];
+	SlruPageStatus page_status[NUM_CLOG_BUFFERS];
+	int			page_number[NUM_CLOG_BUFFERS];
+	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
+
+	/*
+	 * latest_page_number is the page number of the current end of the
+	 * CLOG; this is not critical data, since we use it only to avoid
+	 * swapping out the latest page.
+	 */
+	int			latest_page_number;
+} SlruSharedData;
+typedef SlruSharedData *SlruShared;
+
+
+#define SlruFileName(ctl, path, seg) \
+	snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir, seg)
+
+/*
+ * Macro to mark a buffer slot "most recently used".
+ */
+#define SlruRecentlyUsed(shared, slotno)	\
+	do { \
+		if ((shared)->page_lru_count[slotno] != 0) { \
+			int		iilru; \
+			for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
+				(shared)->page_lru_count[iilru]++; \
+			(shared)->page_lru_count[slotno] = 0; \
+		} \
+	} while (0)
+
+/* Saved info for SlruReportIOError */
+typedef enum
+{
+	SLRU_OPEN_FAILED,
+	SLRU_CREATE_FAILED,
+	SLRU_SEEK_FAILED,
+	SLRU_READ_FAILED,
+	SLRU_WRITE_FAILED
+} SlruErrorCause;
+static SlruErrorCause slru_errcause;
+static int	slru_errno;
+
+
+static bool SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno);
+static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno);
+static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
+static int	SlruSelectLRUPage(SlruCtl ctl, int pageno);
+static bool SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions);
+
+
+/*
+ * Initialization of shared memory
+ */
+
+int
+SimpleLruShmemSize(void)
+{
+	return MAXALIGN(sizeof(SlruSharedData)) + BLCKSZ * NUM_CLOG_BUFFERS;
+}
+
+void
+SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir)
+{
+	bool		found;
+	char	   *bufptr;
+	int			slotno;
+	SlruShared	shared;
+
+	shared = (SlruShared) ShmemInitStruct(name, SimpleLruShmemSize(), &found);
+	Assert(!found);
+
+	ctl->ControlLock = LWLockAssign();
+	ctl->shared = shared;
+	memset(shared, 0, sizeof(SlruSharedData));
+
+	bufptr = ((char *) shared) + MAXALIGN(sizeof(SlruSharedData));
+
+	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+	{
+		shared->page_buffer[slotno] = bufptr;
+		shared->page_status[slotno] = SLRU_PAGE_EMPTY;
+		shared->page_lru_count[slotno] = 1;
+		ctl->BufferLocks[slotno] = LWLockAssign();
+		bufptr += BLCKSZ;
+	}
+
+	/* shared->latest_page_number will be set later */
+
+	/* Init directory path */
+	snprintf(ctl->Dir, MAXPGPATH, "%s/%s", DataDir, subdir);
+}
+
+/*
+ * Initialize (or reinitialize) a page to zeroes.
+ *
+ * The page is not actually written, just set up in shared memory.
+ * The slot number of the new page is returned.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+int
+SimpleLruZeroPage(SlruCtl ctl, int pageno)
+{
+	int			slotno;
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* 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 */
+	MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
+
+	/* Assume this page is now the latest active page */
+	shared->latest_page_number = pageno;
+
+	return slotno;
+}
+
+/*
+ * Find a page in a shared buffer, reading it in if necessary.
+ * The page number must correspond to an already-initialized page.
+ *
+ * The passed-in xid is used only for error reporting, and may be
+ * InvalidTransactionId if no specific xid is associated with the action.
+ *
+ * Return value is the shared-buffer address of the page.
+ * The buffer's LRU access info is updated.
+ * If forwrite is true, the buffer is marked as dirty.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+char *
+SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite)
+{
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* Outer loop handles restart if we lose the buffer to someone else */
+	for (;;)
+	{
+		int			slotno;
+		bool		ok;
+
+		/* See if page already is in memory; if not, pick victim slot */
+		slotno = SlruSelectLRUPage(ctl, pageno);
+
+		/* Did we find the page in memory? */
+		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);
+				if (forwrite)
+					shared->page_status[slotno] = SLRU_PAGE_DIRTY;
+				return shared->page_buffer[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
+		 * SlruSelectLRUPage from selecting it again for someone else.
+		 */
+		SlruRecentlyUsed(shared, slotno);
+
+		/* Release shared lock, grab per-buffer lock instead */
+		LWLockRelease(ctl->ControlLock);
+		LWLockAcquire(ctl->BufferLocks[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(ctl->BufferLocks[slotno]);
+			LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
+			continue;
+		}
+
+		/* Okay, do the read */
+		ok = SlruPhysicalReadPage(ctl, pageno, slotno);
+
+		/* Re-acquire shared control lock and update page state */
+		LWLockAcquire(ctl->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(ctl->BufferLocks[slotno]);
+
+		/* Now it's okay to elog if we failed */
+		if (!ok)
+			SlruReportIOError(ctl, pageno, xid);
+
+		SlruRecentlyUsed(shared, slotno);
+		if (forwrite)
+			shared->page_status[slotno] = SLRU_PAGE_DIRTY;
+		return shared->page_buffer[slotno];
+	}
+}
+
+/*
+ * Write a page from a shared buffer, if necessary.
+ * Does nothing if the specified slot is not dirty.
+ *
+ * NOTE: only one write attempt is made here.  Hence, it is possible that
+ * the page is still dirty at exit (if someone else re-dirtied it during
+ * the write).	However, we *do* attempt a fresh write even if the page
+ * is already being written; this is for checkpoints.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+void
+SimpleLruWritePage(SlruCtl ctl, int slotno)
+{
+	int			pageno;
+	bool		ok;
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* 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(ctl->ControlLock);
+	LWLockAcquire(ctl->BufferLocks[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(ctl->BufferLocks[slotno]);
+		LWLockAcquire(ctl->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);
+
+	/* Re-acquire shared control lock and update page state */
+	LWLockAcquire(ctl->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(ctl->BufferLocks[slotno]);
+
+	/* Now it's okay to elog if we failed */
+	if (!ok)
+		SlruReportIOError(ctl, pageno, InvalidTransactionId);
+}
+
+/*
+ * Physical read of a (previously existing) page into a buffer slot
+ *
+ * On failure, we cannot just elog(ERROR) since caller has put state in
+ * shared memory that must be undone.  So, we return FALSE and save enough
+ * info in static variables to let SlruReportIOError make the report.
+ *
+ * For now, assume it's not worth keeping a file pointer open across
+ * read/write operations.  We could cache one virtual file pointer ...
+ */
+static bool
+SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+	int			fd;
+
+	SlruFileName(ctl, path, segno);
+
+	/*
+	 * In a crash-and-restart situation, it's possible for us to receive
+	 * commands to set the commit status of transactions whose bits are in
+	 * already-truncated segments of the commit log (see notes in
+	 * SlruPhysicalWritePage).	Hence, if we are InRecovery, allow the
+	 * case where the file doesn't exist, and return zeroes instead.
+	 */
+	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
+	if (fd < 0)
+	{
+		if (errno != ENOENT || !InRecovery)
+		{
+			slru_errcause = SLRU_OPEN_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+
+		elog(LOG, "file %s doesn't exist, reading as zeroes", path);
+		MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
+		return true;
+	}
+
+	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
+	{
+		slru_errcause = SLRU_SEEK_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	errno = 0;
+	if (read(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
+	{
+		slru_errcause = SLRU_READ_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	close(fd);
+	return true;
+}
+
+/*
+ * Physical write of a page from a buffer slot
+ *
+ * On failure, we cannot just elog(ERROR) since caller has put state in
+ * shared memory that must be undone.  So, we return FALSE and save enough
+ * info in static variables to let SlruReportIOError make the report.
+ *
+ * For now, assume it's not worth keeping a file pointer open across
+ * read/write operations.  We could cache one virtual file pointer ...
+ */
+static bool
+SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+	int			fd;
+
+	SlruFileName(ctl, path, segno);
+
+	/*
+	 * If the file doesn't already exist, we should create it.  It is
+	 * possible for this to need to happen when writing a page that's not
+	 * first in its segment; we assume the OS can cope with that.  (Note:
+	 * it might seem that it'd be okay to create files only when
+	 * SimpleLruZeroPage is called for the first page of a segment.	However,
+	 * if after a crash and restart the REDO logic elects to replay the
+	 * log from a checkpoint before the latest one, then it's possible
+	 * that we will get commands to set transaction status of transactions
+	 * that have already been truncated from the commit log.  Easiest way
+	 * to deal with that is to accept references to nonexistent files here
+	 * and in SlruPhysicalReadPage.)
+	 */
+	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
+	if (fd < 0)
+	{
+		if (errno != ENOENT)
+		{
+			slru_errcause = SLRU_OPEN_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+
+		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
+						   S_IRUSR | S_IWUSR);
+		if (fd < 0)
+		{
+			slru_errcause = SLRU_CREATE_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+	}
+
+	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
+	{
+		slru_errcause = SLRU_SEEK_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	errno = 0;
+	if (write(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
+	{
+		/* if write didn't set errno, assume problem is no disk space */
+		if (errno == 0)
+			errno = ENOSPC;
+		slru_errcause = SLRU_WRITE_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	close(fd);
+	return true;
+}
+
+/*
+ * Issue the error message after failure of SlruPhysicalReadPage or
+ * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
+ */
+static void
+SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
+{
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+
+	/* XXX TODO: provide xid as context in error messages */
+
+	SlruFileName(ctl, path, segno);
+	errno = slru_errno;
+	switch (slru_errcause)
+	{
+		case SLRU_OPEN_FAILED:
+			elog(ERROR, "open of %s failed: %m", path);
+			break;
+		case SLRU_CREATE_FAILED:
+			elog(ERROR, "creation of file %s failed: %m", path);
+			break;
+		case SLRU_SEEK_FAILED:
+			elog(ERROR, "lseek of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		case SLRU_READ_FAILED:
+			elog(ERROR, "read of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		case SLRU_WRITE_FAILED:
+			elog(ERROR, "write of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		default:
+			/* can't get here, we trust */
+			elog(ERROR, "unknown SimpleLru I/O error");
+			break;
+	}
+}
+
+/*
+ * Select the slot to re-use when we need a free slot.
+ *
+ * The target page number is passed because we need to consider the
+ * possibility that some other process reads in the target page while
+ * we are doing I/O to free a slot.  Hence, check or recheck to see if
+ * any slot already holds the target page, and return that slot if so.
+ * Thus, the returned slot is *either* a slot already holding the pageno
+ * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
+ * or CLEAN).
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+static int
+SlruSelectLRUPage(SlruCtl ctl, int pageno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	/* Outer loop handles restart after I/O */
+	for (;;)
+	{
+		int			slotno;
+		int			bestslot = 0;
+		unsigned int bestcount = 0;
+
+		/* See if page already has a buffer assigned */
+		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+		{
+			if (shared->page_number[slotno] == pageno &&
+				shared->page_status[slotno] != SLRU_PAGE_EMPTY)
+				return slotno;
+		}
+
+		/*
+		 * If we find any EMPTY slot, just select that one. Else locate
+		 * the least-recently-used slot that isn't the latest page.
+		 */
+		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+		{
+			if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
+				return slotno;
+			if (shared->page_lru_count[slotno] > bestcount &&
+			    shared->page_number[slotno] != shared->latest_page_number)
+			{
+				bestslot = slotno;
+				bestcount = shared->page_lru_count[slotno];
+			}
+		}
+
+		/*
+		 * 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, false);
+		else
+			SimpleLruWritePage(ctl, bestslot);
+
+		/*
+		 * Now loop back and try again.  This is the easiest way of
+		 * dealing with corner cases such as the victim page being
+		 * re-dirtied while we wrote it.
+		 */
+	}
+}
+
+/*
+ * This must be called ONCE during postmaster or standalone-backend startup
+ */
+void
+SimpleLruSetLatestPage(SlruCtl ctl, int pageno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+
+	shared->latest_page_number = pageno;
+}
+
+/*
+ * This is called during checkpoint and postmaster/standalone-backend shutdown
+ */
+void
+SimpleLruFlush(SlruCtl ctl, bool checkpoint)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			slotno;
+
+	LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
+
+	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+	{
+		SimpleLruWritePage(ctl, slotno);
+		/*
+		 * When called during a checkpoint,
+		 * we cannot assert that the slot is clean now, since another
+		 * process might have re-dirtied it already.  That's okay.
+		 */
+		Assert(checkpoint ||
+			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
+			   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
+	}
+
+	LWLockRelease(ctl->ControlLock);
+}
+
+/*
+ * Remove all segments before the one holding the passed page number
+ *
+ * When this is called, we know that the database logically contains no
+ * reference to transaction IDs older than oldestXact.	However, we must
+ * not remove any segment until we have performed a checkpoint, to ensure
+ * that no such references remain on disk either; else a crash just after
+ * the truncation might leave us with a problem.  Since CLOG segments hold
+ * a large number of transactions, the opportunity to actually remove a
+ * segment is fairly rare, and so it seems best not to do the checkpoint
+ * unless we have confirmed that there is a removable segment.	Therefore
+ * we issue the checkpoint command here, not in higher-level code as might
+ * seem cleaner.
+ */
+void
+SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
+{
+	int			slotno;
+	SlruShared	shared = (SlruShared) ctl->shared;
+
+	/*
+	 * The cutoff point is the start of the segment containing cutoffPage.
+	 */
+	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;
+
+	if (!SlruScanDirectory(ctl, cutoffPage, false))
+		return;					/* nothing to remove */
+
+	/* Perform a forced CHECKPOINT */
+	CreateCheckPoint(false, true);
+
+	/*
+	 * Scan shared memory and remove any pages preceding the cutoff
+	 * page, to ensure we won't rewrite them later.  (Any dirty pages
+	 * should have been flushed already during the checkpoint, we're just
+	 * being extra careful here.)
+	 */
+	LWLockAcquire(ctl->ControlLock, LW_EXCLUSIVE);
+
+restart:;
+
+	/*
+	 * While we are holding the lock, make an important safety check: the
+	 * planned cutoff point must be <= the current endpoint page.
+	 * Otherwise we have already wrapped around, and proceeding with the
+	 * truncation would risk removing the current segment.
+	 */
+	if (ctl->PagePrecedes(shared->latest_page_number, cutoffPage))
+	{
+		LWLockRelease(ctl->ControlLock);
+		elog(LOG, "unable to truncate %s: apparent wraparound", ctl->Dir);
+		return;
+	}
+
+	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+	{
+		if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
+			continue;
+		if (!ctl->PagePrecedes(shared->page_number[slotno], cutoffPage))
+			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;
+		}
+
+		/*
+		 * 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, false);
+		else
+			SimpleLruWritePage(ctl, slotno);
+		goto restart;
+	}
+
+	LWLockRelease(ctl->ControlLock);
+
+	/* Now we can remove the old segment(s) */
+	(void) SlruScanDirectory(ctl, cutoffPage, true);
+}
+
+/*
+ * SlruTruncate subroutine: scan directory for removable segments.
+ * Actually remove them iff doDeletions is true.  Return TRUE iff any
+ * removable segments were found.  Note: no locking is needed.
+ */
+static bool
+SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
+{
+	bool		found = false;
+	DIR		   *cldir;
+	struct dirent *clde;
+	int			segno;
+	int			segpage;
+	char		path[MAXPGPATH];
+
+	cldir = opendir(ctl->Dir);
+	if (cldir == NULL)
+		elog(ERROR, "could not open directory (%s): %m", ctl->Dir);
+
+	errno = 0;
+	while ((clde = readdir(cldir)) != NULL)
+	{
+		if (strlen(clde->d_name) == 4 &&
+			strspn(clde->d_name, "0123456789ABCDEF") == 4)
+		{
+			segno = (int) strtol(clde->d_name, NULL, 16);
+			segpage = segno * SLRU_PAGES_PER_SEGMENT;
+			if (ctl->PagePrecedes(segpage, cutoffPage))
+			{
+				found = true;
+				if (doDeletions)
+				{
+					elog(LOG, "removing file %s/%s", ctl->Dir, clde->d_name);
+					snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, clde->d_name);
+					unlink(path);
+				}
+			}
+		}
+		errno = 0;
+	}
+	if (errno)
+		elog(ERROR, "could not read directory (%s): %m", ctl->Dir);
+	closedir(cldir);
+
+	return found;
+}
diff -ruN ../base/src/backend/storage/lmgr/lwlock.c src/backend/storage/lmgr/lwlock.c
--- ../base/src/backend/storage/lmgr/lwlock.c	2002-10-10 19:07:27.000000000 +0200
+++ src/backend/storage/lmgr/lwlock.c	2003-05-01 21:31:34.000000000 +0200
@@ -108,8 +108,8 @@
 	/* bufmgr.c needs two for each shared buffer */
 	numLocks += 2 * NBuffers;
 
-	/* clog.c needs one per CLOG buffer */
-	numLocks += NUM_CLOG_BUFFERS;
+	/* clog.c needs one per CLOG buffer + one control lock */
+	numLocks += NUM_CLOG_BUFFERS + 1;
 
 	/* Perhaps create a few more for use by user-defined modules? */
 
diff -ruN ../base/src/include/access/slru.h src/include/access/slru.h
--- ../base/src/include/access/slru.h	1970-01-01 01:00:00.000000000 +0100
+++ src/include/access/slru.h	2003-05-01 20:50:14.000000000 +0200
@@ -0,0 +1,57 @@
+/*
+ * slru.h
+ *
+ * Simple LRU
+ *
+ * Portions Copyright (c) 2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $Id: $
+ */
+#ifndef SLRU_H
+#define SLRU_H
+
+#include "access/xlog.h"
+
+/* exported because lwlock.c needs it */
+#define NUM_CLOG_BUFFERS	8
+
+
+typedef struct SlruCtlData
+{
+	void *shared;	/* pointer to SlruSharedData */
+	LWLockId ControlLock;   // = CLogControlLock
+/*
+ * BufferLocks is set during CLOGShmemInit and does not change thereafter.
+ * The value is automatically inherited by backends via fork, and
+ * doesn't need to be in shared memory.
+ */
+	LWLockId BufferLocks[NUM_CLOG_BUFFERS];		/* Per-buffer I/O locks */
+
+/*
+ * Dir is set during SimpleLruShmemInit and does not change thereafter.
+ * The value is automatically inherited by backends via fork, and
+ * doesn't need to be in shared memory.
+ */
+	char Dir[MAXPGPATH];
+	
+/*
+ * Decide which of two page numbers is "older" for truncation purposes.
+ * We need to use comparison of TransactionIds here in order to do the right
+ * thing with wraparound XID arithmetic.
+ */
+	bool (*PagePrecedes)(int, int);
+
+} SlruCtlData;
+typedef SlruCtlData *SlruCtl;
+
+extern int	SimpleLruShmemSize(void);
+extern void SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir);
+extern int	SimpleLruZeroPage(SlruCtl ctl, int pageno);
+extern char	*SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite);
+extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
+extern void SimpleLruSetLatestPage(SlruCtl ctl, int pageno);
+extern void SimpleLruFlush(SlruCtl ctl, bool checkpoint);
+extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
+
+#endif   /* SLRU_H */
diff -ruN ../base/src/include/storage/lwlock.h src/include/storage/lwlock.h
--- ../base/src/include/storage/lwlock.h	2002-06-21 02:12:30.000000000 +0200
+++ src/include/storage/lwlock.h	2003-05-01 21:32:21.000000000 +0200
@@ -37,7 +37,6 @@
 	WALWriteLock,
 	ControlFileLock,
 	CheckpointLock,
-	CLogControlLock,
 	RelCacheInitLock,
 
 	NumFixedLWLocks,			/* must be last except for

Re: Simple LRU for clog and (later) subtrans

От:

Manfred Koizar <mkoi-pg@aon.at>

Дата:

5 мая 2003 г. в 19:08:04

On Fri, 02 May 2003 12:01:34 +0200, I wrote:
>This patch extracts page buffer pooling and the simple
>least-recently-used strategy from clog.c into slru.c.

Here's a new version that does not conflict with Bruce's recent clog.c
changes.

Servus
 Manfred
diff -ruN ../base/src/backend/access/transam/Makefile src/backend/access/transam/Makefile
--- ../base/src/backend/access/transam/Makefile	2001-08-25 20:52:41.000000000 +0200
+++ src/backend/access/transam/Makefile	2003-05-05 21:22:54.000000000 +0200
@@ -12,7 +12,7 @@
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o
+OBJS = clog.o transam.o varsup.o xact.o xid.o xlog.o xlogutils.o rmgr.o slru.o
 
 all: SUBSYS.o
 
diff -ruN ../base/src/backend/access/transam/clog.c src/backend/access/transam/clog.c
--- ../base/src/backend/access/transam/clog.c	2003-05-05 19:22:05.000000000 +0200
+++ src/backend/access/transam/clog.c	2003-05-06 00:03:49.000000000 +0200
@@ -26,15 +26,14 @@
 #include 
 
 #include "access/clog.h"
+#include "access/slru.h"
 #include "storage/lwlock.h"
 #include "miscadmin.h"
 
 
 /*
  * Defines for CLOG page and segment sizes.  A page is the same BLCKSZ
- * as is used everywhere else in Postgres.	The CLOG segment size can be
- * chosen somewhat arbitrarily; we make it 1 million transactions by default,
- * or 256Kb.
+ * as is used everywhere else in Postgres.
  *
  * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
  * CLOG page numbering also wraps around at 0xFFFFFFFF/CLOG_XACTS_PER_PAGE,
@@ -43,17 +42,12 @@
  * segment and page numbers in TruncateCLOG (see CLOGPagePrecedes).
  */
 
-#define CLOG_BLCKSZ			BLCKSZ
-
 /* We need two bits per xact, so four xacts fit in a byte */
 #define CLOG_BITS_PER_XACT	2
 #define CLOG_XACTS_PER_BYTE 4
-#define CLOG_XACTS_PER_PAGE (CLOG_BLCKSZ * CLOG_XACTS_PER_BYTE)
+#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
 #define CLOG_XACT_BITMASK	((1 << CLOG_BITS_PER_XACT) - 1)
 
-#define CLOG_XACTS_PER_SEGMENT	0x100000
-#define CLOG_PAGES_PER_SEGMENT	(CLOG_XACTS_PER_SEGMENT / CLOG_XACTS_PER_PAGE)
-
 #define TransactionIdToPage(xid)	((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToByte(xid)	(TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
@@ -63,47 +57,6 @@
 /*----------
  * Shared-memory data structures for CLOG control
  *
- * We use a simple least-recently-used scheme to manage a pool of page
- * buffers for the CLOG.  Under ordinary circumstances we expect that write
- * traffic will occur mostly to the latest CLOG page (and to the just-prior
- * page, soon after a page transition).  Read traffic will probably touch
- * a larger span of pages, but in any case a fairly small number of page
- * buffers should be sufficient.  So, we just search the buffers using plain
- * linear search; there's no need for a hashtable or anything fancy.
- * The management algorithm is straight LRU except that we will never swap
- * out the latest page (since we know it's going to be hit again eventually).
- *
- * We use an overall 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.
- *
  * XLOG interactions: this module generates an XLOG record whenever a new
  * CLOG page is initialized to zeroes.	Other writes of CLOG come from
  * recording of transaction commit or abort in xact.c, which generates its
@@ -117,90 +70,12 @@
  *----------
  */
 
-typedef enum
-{
-	CLOG_PAGE_EMPTY,			/* CLOG buffer is not in use */
-	CLOG_PAGE_READ_IN_PROGRESS, /* CLOG page is being read in */
-	CLOG_PAGE_CLEAN,			/* CLOG page is valid and not dirty */
-	CLOG_PAGE_DIRTY,			/* CLOG page is valid but needs write */
-	CLOG_PAGE_WRITE_IN_PROGRESS /* CLOG page is being written out */
-} ClogPageStatus;
-
-/*
- * Shared-memory state for CLOG.
- */
-typedef struct ClogCtlData
-{
-	/*
-	 * Info for each buffer slot.  Page number is undefined when status is
-	 * EMPTY.  lru_count is essentially the number of operations since
-	 * last use of this page; the page with highest lru_count is the best
-	 * candidate to replace.
-	 */
-	char	   *page_buffer[NUM_CLOG_BUFFERS];
-	ClogPageStatus page_status[NUM_CLOG_BUFFERS];
-	int			page_number[NUM_CLOG_BUFFERS];
-	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
-
-	/*
-	 * latest_page_number is the page number of the current end of the
-	 * CLOG; this is not critical data, since we use it only to avoid
-	 * swapping out the latest page.
-	 */
-	int			latest_page_number;
-} ClogCtlData;
-
-static ClogCtlData *ClogCtl = NULL;
-
-/*
- * ClogBufferLocks is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
- */
-static LWLockId *ClogBufferLocks;		/* Per-buffer I/O locks */
-
-/*
- * ClogDir is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
- */
-static char ClogDir[MAXPGPATH];
-
-#define ClogFileName(path, seg) \
-	snprintf(path, MAXPGPATH, "%s/%04X", ClogDir, seg)
-
-/*
- * Macro to mark a buffer slot "most recently used".
- */
-#define ClogRecentlyUsed(slotno)	\
-	do { \
-		int		iilru; \
-		for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
-			ClogCtl->page_lru_count[iilru]++; \
-		ClogCtl->page_lru_count[slotno] = 0; \
-	} while (0)
-
-/* Saved info for CLOGReportIOError */
-typedef enum
-{
-	CLOG_OPEN_FAILED,
-	CLOG_CREATE_FAILED,
-	CLOG_SEEK_FAILED,
-	CLOG_READ_FAILED,
-	CLOG_WRITE_FAILED
-} ClogErrorCause;
-static ClogErrorCause clog_errcause;
-static int	clog_errno;
 
+static SlruCtlData ClogCtlData;
+static SlruCtl ClogCtl = &ClogCtlData;
+  
 
 static int	ZeroCLOGPage(int pageno, bool writeXlog);
-static int	ReadCLOGPage(int pageno, TransactionId xid);
-static void WriteCLOGPage(int slotno);
-static bool CLOGPhysicalReadPage(int pageno, int slotno);
-static bool CLOGPhysicalWritePage(int pageno, int slotno);
-static void CLOGReportIOError(int pageno, TransactionId xid);
-static int	SelectLRUCLOGPage(int pageno);
-static bool ScanCLOGDirectory(int cutoffPage, bool doDeletions);
 static bool CLOGPagePrecedes(int page1, int page2);
 static void WriteZeroPageXlogRec(int pageno);
 
@@ -217,16 +92,15 @@
 	int			pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);
 	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
-	int			slotno;
 	char	   *byteptr;
 
 	Assert(status == TRANSACTION_STATUS_COMMITTED ||
 		   status == TRANSACTION_STATUS_ABORTED);
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
 
-	slotno = ReadCLOGPage(pageno, xid);
-	byteptr = ClogCtl->page_buffer[slotno] + byteno;
+	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, true);
+	byteptr += byteno;
 
 	/* Current state should be 0 or target state */
 	Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
@@ -234,9 +108,9 @@
 
 	*byteptr |= (status << bshift);
 
-	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
+	/* ...->page_status[slotno] = CLOG_PAGE_DIRTY; already done */
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->locks->ControlLock);
 }
 
 /*
@@ -251,18 +125,17 @@
 	int			pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);
 	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
-	int			slotno;
 	char	   *byteptr;
 	XidStatus	status;
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
 
-	slotno = ReadCLOGPage(pageno, xid);
-	byteptr = ClogCtl->page_buffer[slotno] + byteno;
+	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, false);
+	byteptr += byteno;
 
 	status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->locks->ControlLock);
 
 	return status;
 }
@@ -271,70 +144,18 @@
 /*
  * Initialization of shared memory for CLOG
  */
+
 int
 CLOGShmemSize(void)
 {
-	return MAXALIGN(sizeof(ClogCtlData) + CLOG_BLCKSZ * NUM_CLOG_BUFFERS)
-#ifdef EXEC_BACKEND
-			+ MAXALIGN(NUM_CLOG_BUFFERS * sizeof(LWLockId))
-#endif
-	;
+	return SimpleLruShmemSize();
 }
 
-
 void
 CLOGShmemInit(void)
 {
-	bool		found;
-	int			slotno;
-
-	/* Handle ClogCtl */
-	
-	/* this must agree with space requested by CLOGShmemSize() */
-	ClogCtl = (ClogCtlData *) ShmemInitStruct("CLOG Ctl",
-				MAXALIGN(sizeof(ClogCtlData) +
-				CLOG_BLCKSZ * NUM_CLOG_BUFFERS), &found);
-
-	if (!IsUnderPostmaster)
-	/* Initialize ClogCtl shared memory area */
-	{
-		char	   *bufptr;
-
-		Assert(!found);
-
-		memset(ClogCtl, 0, sizeof(ClogCtlData));
-
-		bufptr = (char *)ClogCtl + sizeof(ClogCtlData);
-	
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-		{
-			ClogCtl->page_buffer[slotno] = bufptr;
-			ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
-			bufptr += CLOG_BLCKSZ;
-		}
-
-		/* ClogCtl->latest_page_number will be set later */
-	}
-	else
-		Assert(found);
-
-	/* Handle ClogBufferLocks */
-	
-#ifdef EXEC_BACKEND
-	ClogBufferLocks = (LWLockId *) ShmemInitStruct("CLOG Buffer Locks",
-						NUM_CLOG_BUFFERS * sizeof(LWLockId), &found);
-	Assert((!found && !IsUnderPostmaster) || (found && IsUnderPostmaster));
-#else
-	ClogBufferLocks = malloc(NUM_CLOG_BUFFERS * sizeof(LWLockId));
-	Assert(ClogBufferLocks);
-#endif
-
-	if (!IsUnderPostmaster)
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-			ClogBufferLocks[slotno] = LWLockAssign();
-
-	/* Init CLOG directory path */
-	snprintf(ClogDir, MAXPGPATH, "%s/pg_clog", DataDir);
+	SimpleLruInit(ClogCtl, "CLOG Ctl", "pg_clog");
+	ClogCtl->PagePrecedes = CLOGPagePrecedes;
 }
 
 /*
@@ -348,16 +169,16 @@
 {
 	int			slotno;
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
 
 	/* Create and zero the first page of the commit log */
 	slotno = ZeroCLOGPage(0, false);
 
 	/* Make sure it's written out */
-	WriteCLOGPage(slotno);
-	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+	SimpleLruWritePage(ClogCtl, slotno);
+	/* Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN); */
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->locks->ControlLock);
 }
 
 /*
@@ -372,24 +193,7 @@
 static int
 ZeroCLOGPage(int pageno, bool writeXlog)
 {
-	int			slotno;
-
-	/* Find a suitable buffer slot for the page */
-	slotno = SelectLRUCLOGPage(pageno);
-	Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-		   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN ||
-		   ClogCtl->page_number[slotno] == pageno);
-
-	/* Mark the slot as containing this page */
-	ClogCtl->page_number[slotno] = pageno;
-	ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
-	ClogRecentlyUsed(slotno);
-
-	/* Set the buffer to zeroes */
-	MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
-
-	/* Assume this page is now the latest active page */
-	ClogCtl->latest_page_number = pageno;
+	int			slotno = SimpleLruZeroPage(ClogCtl, pageno);
 
 	if (writeXlog)
 		WriteZeroPageXlogRec(pageno);
@@ -398,430 +202,6 @@
 }
 
 /*
- * Find a CLOG page in a shared buffer, reading it in if necessary.
- * The page number must correspond to an already-initialized page.
- *
- * The passed-in xid is used only for error reporting, and may be
- * InvalidTransactionId if no specific xid is associated with the action.
- *
- * Return value is the shared-buffer slot number now holding the page.
- * The buffer's LRU access info is updated.
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static int
-ReadCLOGPage(int pageno, TransactionId xid)
-{
-	/* Outer loop handles restart if we lose the buffer to someone else */
-	for (;;)
-	{
-		int			slotno;
-		bool		ok;
-
-		/* See if page already is in memory; if not, pick victim slot */
-		slotno = SelectLRUCLOGPage(pageno);
-
-		/* Did we find the page in memory? */
-		if (ClogCtl->page_number[slotno] == pageno &&
-			ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
-		{
-			/* If page is still being read in, we cannot use it yet */
-			if (ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
-			{
-				/* otherwise, it's ready to use */
-				ClogRecentlyUsed(slotno);
-				return slotno;
-			}
-		}
-		else
-		{
-			/* We found no match; assert we selected a freeable slot */
-			Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-				   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
-		}
-
-		/* Mark the slot read-busy (no-op if it already was) */
-		ClogCtl->page_number[slotno] = pageno;
-		ClogCtl->page_status[slotno] = CLOG_PAGE_READ_IN_PROGRESS;
-
-		/*
-		 * Temporarily mark page as recently-used to discourage
-		 * SelectLRUCLOGPage from selecting it again for someone else.
-		 */
-		ClogCtl->page_lru_count[slotno] = 0;
-
-		/* Release shared lock, grab per-buffer lock instead */
-		LWLockRelease(CLogControlLock);
-		LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
-			ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
-		{
-			LWLockRelease(ClogBufferLocks[slotno]);
-			LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-			continue;
-		}
-
-		/* Okay, do the read */
-		ok = CLOGPhysicalReadPage(pageno, slotno);
-
-		/* Re-acquire shared control lock and update page state */
-		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-		Assert(ClogCtl->page_number[slotno] == pageno &&
-			 ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS);
-
-		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_EMPTY;
-
-		LWLockRelease(ClogBufferLocks[slotno]);
-
-		/* Now it's okay to elog if we failed */
-		if (!ok)
-			CLOGReportIOError(pageno, xid);
-
-		ClogRecentlyUsed(slotno);
-		return slotno;
-	}
-}
-
-/*
- * Write a CLOG page from a shared buffer, if necessary.
- * Does nothing if the specified slot is not dirty.
- *
- * NOTE: only one write attempt is made here.  Hence, it is possible that
- * the page is still dirty at exit (if someone else re-dirtied it during
- * the write).	However, we *do* attempt a fresh write even if the page
- * is already being written; this is for checkpoints.
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static void
-WriteCLOGPage(int slotno)
-{
-	int			pageno;
-	bool		ok;
-
-	/* Do nothing if page does not need writing */
-	if (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
-		ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS)
-		return;
-
-	pageno = ClogCtl->page_number[slotno];
-
-	/* Release shared lock, grab per-buffer lock instead */
-	LWLockRelease(CLogControlLock);
-	LWLockAcquire(ClogBufferLocks[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 (ClogCtl->page_number[slotno] != pageno ||
-		(ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
-		 ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS))
-	{
-		LWLockRelease(ClogBufferLocks[slotno]);
-		LWLockAcquire(CLogControlLock, 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.)
-	 */
-	ClogCtl->page_status[slotno] = CLOG_PAGE_WRITE_IN_PROGRESS;
-
-	/* Okay, do the write */
-	ok = CLOGPhysicalWritePage(pageno, slotno);
-
-	/* Re-acquire shared control lock and update page state */
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	Assert(ClogCtl->page_number[slotno] == pageno &&
-		   (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS ||
-			ClogCtl->page_status[slotno] == CLOG_PAGE_DIRTY));
-
-	/* Cannot set CLEAN if someone re-dirtied page since write started */
-	if (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS)
-		ClogCtl->page_status[slotno] = ok ? CLOG_PAGE_CLEAN : CLOG_PAGE_DIRTY;
-
-	LWLockRelease(ClogBufferLocks[slotno]);
-
-	/* Now it's okay to elog if we failed */
-	if (!ok)
-		CLOGReportIOError(pageno, InvalidTransactionId);
-}
-
-/*
- * Physical read of a (previously existing) page into a buffer slot
- *
- * On failure, we cannot just elog(ERROR) since caller has put state in
- * shared memory that must be undone.  So, we return FALSE and save enough
- * info in static variables to let CLOGReportIOError make the report.
- *
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations.  We could cache one virtual file pointer ...
- */
-static bool
-CLOGPhysicalReadPage(int pageno, int slotno)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-	int			fd;
-
-	ClogFileName(path, segno);
-
-	/*
-	 * In a crash-and-restart situation, it's possible for us to receive
-	 * commands to set the commit status of transactions whose bits are in
-	 * already-truncated segments of the commit log (see notes in
-	 * CLOGPhysicalWritePage).	Hence, if we are InRecovery, allow the
-	 * case where the file doesn't exist, and return zeroes instead.
-	 */
-	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
-	if (fd < 0)
-	{
-		if (errno != ENOENT || !InRecovery)
-		{
-			clog_errcause = CLOG_OPEN_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-
-		elog(LOG, "clog file %s doesn't exist, reading as zeroes", path);
-		MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
-		return true;
-	}
-
-	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
-	{
-		clog_errcause = CLOG_SEEK_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	errno = 0;
-	if (read(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
-	{
-		clog_errcause = CLOG_READ_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	close(fd);
-	return true;
-}
-
-/*
- * Physical write of a page from a buffer slot
- *
- * On failure, we cannot just elog(ERROR) since caller has put state in
- * shared memory that must be undone.  So, we return FALSE and save enough
- * info in static variables to let CLOGReportIOError make the report.
- *
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations.  We could cache one virtual file pointer ...
- */
-static bool
-CLOGPhysicalWritePage(int pageno, int slotno)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-	int			fd;
-
-	ClogFileName(path, segno);
-
-	/*
-	 * If the file doesn't already exist, we should create it.  It is
-	 * possible for this to need to happen when writing a page that's not
-	 * first in its segment; we assume the OS can cope with that.  (Note:
-	 * it might seem that it'd be okay to create files only when
-	 * ZeroCLOGPage is called for the first page of a segment.	However,
-	 * if after a crash and restart the REDO logic elects to replay the
-	 * log from a checkpoint before the latest one, then it's possible
-	 * that we will get commands to set transaction status of transactions
-	 * that have already been truncated from the commit log.  Easiest way
-	 * to deal with that is to accept references to nonexistent files here
-	 * and in CLOGPhysicalReadPage.)
-	 */
-	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
-	if (fd < 0)
-	{
-		if (errno != ENOENT)
-		{
-			clog_errcause = CLOG_OPEN_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-
-		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
-						   S_IRUSR | S_IWUSR);
-		if (fd < 0)
-		{
-			clog_errcause = CLOG_CREATE_FAILED;
-			clog_errno = errno;
-			return false;
-		}
-	}
-
-	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
-	{
-		clog_errcause = CLOG_SEEK_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	errno = 0;
-	if (write(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
-	{
-		/* if write didn't set errno, assume problem is no disk space */
-		if (errno == 0)
-			errno = ENOSPC;
-		clog_errcause = CLOG_WRITE_FAILED;
-		clog_errno = errno;
-		return false;
-	}
-
-	close(fd);
-	return true;
-}
-
-/*
- * Issue the error message after failure of CLOGPhysicalReadPage or
- * CLOGPhysicalWritePage.  Call this after cleaning up shared-memory state.
- */
-static void
-CLOGReportIOError(int pageno, TransactionId xid)
-{
-	int			segno = pageno / CLOG_PAGES_PER_SEGMENT;
-	int			rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
-	int			offset = rpageno * CLOG_BLCKSZ;
-	char		path[MAXPGPATH];
-
-	/* XXX TODO: provide xid as context in error messages */
-
-	ClogFileName(path, segno);
-	errno = clog_errno;
-	switch (clog_errcause)
-	{
-		case CLOG_OPEN_FAILED:
-			elog(ERROR, "open of %s failed: %m", path);
-			break;
-		case CLOG_CREATE_FAILED:
-			elog(ERROR, "creation of file %s failed: %m", path);
-			break;
-		case CLOG_SEEK_FAILED:
-			elog(ERROR, "lseek of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		case CLOG_READ_FAILED:
-			elog(ERROR, "read of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		case CLOG_WRITE_FAILED:
-			elog(ERROR, "write of clog file %u, offset %u failed: %m",
-				 segno, offset);
-			break;
-		default:
-			/* can't get here, we trust */
-			elog(ERROR, "unknown CLOG I/O error");
-			break;
-	}
-}
-
-/*
- * Select the slot to re-use when we need a free slot.
- *
- * The target page number is passed because we need to consider the
- * possibility that some other process reads in the target page while
- * we are doing I/O to free a slot.  Hence, check or recheck to see if
- * any slot already holds the target page, and return that slot if so.
- * Thus, the returned slot is *either* a slot already holding the pageno
- * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
- * or CLEAN).
- *
- * Control lock must be held at entry, and will be held at exit.
- */
-static int
-SelectLRUCLOGPage(int pageno)
-{
-	/* Outer loop handles restart after I/O */
-	for (;;)
-	{
-		int			slotno;
-		int			bestslot = 0;
-		unsigned int bestcount = 0;
-
-		/* See if page already has a buffer assigned */
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-		{
-			if (ClogCtl->page_number[slotno] == pageno &&
-				ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
-				return slotno;
-		}
-
-		/*
-		 * If we find any EMPTY slot, just select that one. Else locate
-		 * the least-recently-used slot that isn't the latest CLOG page.
-		 */
-		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-		{
-			if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
-				return slotno;
-			if (ClogCtl->page_lru_count[slotno] > bestcount &&
-			 ClogCtl->page_number[slotno] != ClogCtl->latest_page_number)
-			{
-				bestslot = slotno;
-				bestcount = ClogCtl->page_lru_count[slotno];
-			}
-		}
-
-		/*
-		 * If the selected page is clean, we're set.
-		 */
-		if (ClogCtl->page_status[bestslot] == CLOG_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 ReadCLOGPage to wait for
-		 * the read to complete.
-		 */
-		if (ClogCtl->page_status[bestslot] == CLOG_PAGE_READ_IN_PROGRESS)
-			(void) ReadCLOGPage(ClogCtl->page_number[bestslot],
-								InvalidTransactionId);
-		else
-			WriteCLOGPage(bestslot);
-
-		/*
-		 * Now loop back and try again.  This is the easiest way of
-		 * dealing with corner cases such as the victim page being
-		 * re-dirtied while we wrote it.
-		 */
-	}
-}
-
-/*
  * This must be called ONCE during postmaster or standalone-backend startup,
  * after StartupXLOG has initialized ShmemVariableCache->nextXid.
  */
@@ -831,7 +211,7 @@
 	/*
 	 * Initialize our idea of the latest page number.
 	 */
-	ClogCtl->latest_page_number = TransactionIdToPage(ShmemVariableCache->nextXid);
+	SimpleLruSetLatestPage(ClogCtl, TransactionIdToPage(ShmemVariableCache->nextXid));
 }
 
 /*
@@ -840,18 +220,7 @@
 void
 ShutdownCLOG(void)
 {
-	int			slotno;
-
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		WriteCLOGPage(slotno);
-		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
-			   ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
-	}
-
-	LWLockRelease(CLogControlLock);
+	SimpleLruFlush(ClogCtl, false);
 }
 
 /*
@@ -860,21 +229,7 @@
 void
 CheckPointCLOG(void)
 {
-	int			slotno;
-
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		WriteCLOGPage(slotno);
-
-		/*
-		 * We cannot assert that the slot is clean now, since another
-		 * process might have re-dirtied it already.  That's okay.
-		 */
-	}
-
-	LWLockRelease(CLogControlLock);
+	SimpleLruFlush(ClogCtl, true);
 }
 
 
@@ -901,12 +256,12 @@
 
 	pageno = TransactionIdToPage(newestXact);
 
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+	LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
 
 	/* Zero the page and make an XLOG entry about it */
 	ZeroCLOGPage(pageno, true);
 
-	LWLockRelease(CLogControlLock);
+	LWLockRelease(ClogCtl->locks->ControlLock);
 }
 
 
@@ -928,126 +283,15 @@
 TruncateCLOG(TransactionId oldestXact)
 {
 	int			cutoffPage;
-	int			slotno;
 
 	/*
 	 * The cutoff point is the start of the segment containing oldestXact.
+	 * We pass the *page* containing oldestXact to SimpleLruTruncate.
 	 */
-	oldestXact -= oldestXact % CLOG_XACTS_PER_SEGMENT;
 	cutoffPage = TransactionIdToPage(oldestXact);
-
-	if (!ScanCLOGDirectory(cutoffPage, false))
-		return;					/* nothing to remove */
-
-	/* Perform a forced CHECKPOINT */
-	CreateCheckPoint(false, true);
-
-	/*
-	 * Scan CLOG shared memory and remove any pages preceding the cutoff
-	 * page, to ensure we won't rewrite them later.  (Any dirty pages
-	 * should have been flushed already during the checkpoint, we're just
-	 * being extra careful here.)
-	 */
-	LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-restart:;
-
-	/*
-	 * While we are holding the lock, make an important safety check: the
-	 * planned cutoff point must be <= the current CLOG endpoint page.
-	 * Otherwise we have already wrapped around, and proceeding with the
-	 * truncation would risk removing the current CLOG segment.
-	 */
-	if (CLOGPagePrecedes(ClogCtl->latest_page_number, cutoffPage))
-	{
-		LWLockRelease(CLogControlLock);
-		elog(LOG, "unable to truncate commit log: apparent wraparound");
-		return;
-	}
-
-	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
-	{
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
-			continue;
-		if (!CLOGPagePrecedes(ClogCtl->page_number[slotno], cutoffPage))
-			continue;
-
-		/*
-		 * If page is CLEAN, just change state to EMPTY (expected case).
-		 */
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN)
-		{
-			ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
-			continue;
-		}
-
-		/*
-		 * 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 SelectLRUCLOGPage.
-		 */
-		if (ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS)
-			(void) ReadCLOGPage(ClogCtl->page_number[slotno],
-								InvalidTransactionId);
-		else
-			WriteCLOGPage(slotno);
-		goto restart;
-	}
-
-	LWLockRelease(CLogControlLock);
-
-	/* Now we can remove the old CLOG segment(s) */
-	(void) ScanCLOGDirectory(cutoffPage, true);
+	SimpleLruTruncate(ClogCtl, cutoffPage);
 }
 
-/*
- * TruncateCLOG subroutine: scan CLOG directory for removable segments.
- * Actually remove them iff doDeletions is true.  Return TRUE iff any
- * removable segments were found.  Note: no locking is needed.
- */
-static bool
-ScanCLOGDirectory(int cutoffPage, bool doDeletions)
-{
-	bool		found = false;
-	DIR		   *cldir;
-	struct dirent *clde;
-	int			segno;
-	int			segpage;
-	char		path[MAXPGPATH];
-
-	cldir = opendir(ClogDir);
-	if (cldir == NULL)
-		elog(ERROR, "could not open transaction-commit log directory (%s): %m",
-			 ClogDir);
-
-	errno = 0;
-	while ((clde = readdir(cldir)) != NULL)
-	{
-		if (strlen(clde->d_name) == 4 &&
-			strspn(clde->d_name, "0123456789ABCDEF") == 4)
-		{
-			segno = (int) strtol(clde->d_name, NULL, 16);
-			segpage = segno * CLOG_PAGES_PER_SEGMENT;
-			if (CLOGPagePrecedes(segpage, cutoffPage))
-			{
-				found = true;
-				if (doDeletions)
-				{
-					elog(LOG, "removing commit log file %s", clde->d_name);
-					snprintf(path, MAXPGPATH, "%s/%s", ClogDir, clde->d_name);
-					unlink(path);
-				}
-			}
-		}
-		errno = 0;
-	}
-	if (errno)
-		elog(ERROR, "could not read transaction-commit log directory (%s): %m",
-			 ClogDir);
-	closedir(cldir);
-
-	return found;
-}
 
 /*
  * Decide which of two CLOG page numbers is "older" for truncation purposes.
@@ -1107,13 +351,13 @@
 
 		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
 
-		LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+		LWLockAcquire(ClogCtl->locks->ControlLock, LW_EXCLUSIVE);
 
 		slotno = ZeroCLOGPage(pageno, false);
-		WriteCLOGPage(slotno);
-		Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+		SimpleLruWritePage(ClogCtl, slotno);
+		/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
 
-		LWLockRelease(CLogControlLock);
+		LWLockRelease(ClogCtl->locks->ControlLock);
 	}
 }
 
diff -ruN ../base/src/backend/access/transam/slru.c src/backend/access/transam/slru.c
--- ../base/src/backend/access/transam/slru.c	1970-01-01 01:00:00.000000000 +0100
+++ src/backend/access/transam/slru.c	2003-05-06 00:11:04.000000000 +0200
@@ -0,0 +1,886 @@
+/*-------------------------------------------------------------------------
+ *
+ * slru.c
+ *		Simple LRU
+ *
+ * This module replaces the old "pg_log" access code, which treated pg_log
+ * essentially like a relation, in that it went through the regular buffer
+ * manager.  The problem with that was that there wasn't any good way to
+ * recycle storage space for transactions so old that they'll never be
+ * looked up again.  Now we use specialized access code so that the commit
+ * log can be broken into relatively small, independent segments.
+ *
+ * Portions Copyright (c) 2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $Header: $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include 
+#include 
+#include 
+#include 
+#include 
+
+#include "access/slru.h"
+#include "storage/lwlock.h"
+#include "miscadmin.h"
+
+
+/*
+ * Define segment size.  A page is the same BLCKSZ as is used everywhere
+ * else in Postgres.  The segment size can be chosen somewhat arbitrarily;
+ * we make it 32 pages by default, or 256Kb, i.e. 1M transactions for CLOG
+ * or 64K transactions for SUBTRANS.
+ *
+ * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
+ * page numbering also wraps around at 0xFFFFFFFF/xxxx_XACTS_PER_PAGE (where
+ * xxxx is CLOG or SUBTRANS, respectively), and segment numbering at
+ * 0xFFFFFFFF/xxxx_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.  We need
+ * take no explicit notice of that fact in this module, except when comparing
+ * segment and page numbers in SimpleLruTruncate (see PagePrecedes()).
+ */
+
+#define SLRU_PAGES_PER_SEGMENT	32
+
+
+/*----------
+ * Shared-memory data structures for SLRU control
+ *
+ * We use a simple least-recently-used scheme to manage a pool of page
+ * buffers.  Under ordinary circumstances we expect that write
+ * traffic will occur mostly to the latest page (and to the just-prior
+ * page, soon after a page transition).  Read traffic will probably touch
+ * a larger span of pages, but in any case a fairly small number of page
+ * buffers should be sufficient.  So, we just search the buffers using plain
+ * linear search; there's no need for a hashtable or anything fancy.
+ * The management algorithm is straight LRU except that we will never swap
+ * 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.
+ *----------
+ */
+
+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;
+
+/*
+ * Shared-memory state
+ */
+typedef struct SlruSharedData
+{
+	/*
+	 * Info for each buffer slot.  Page number is undefined when status is
+	 * EMPTY.  lru_count is essentially the number of page switches since
+	 * last use of this page; the page with highest lru_count is the best
+	 * candidate to replace.
+	 */
+	char	   *page_buffer[NUM_CLOG_BUFFERS];
+	SlruPageStatus page_status[NUM_CLOG_BUFFERS];
+	int			page_number[NUM_CLOG_BUFFERS];
+	unsigned int page_lru_count[NUM_CLOG_BUFFERS];
+
+	/*
+	 * latest_page_number is the page number of the current end of the
+	 * CLOG; this is not critical data, since we use it only to avoid
+	 * swapping out the latest page.
+	 */
+	int			latest_page_number;
+} SlruSharedData;
+typedef SlruSharedData *SlruShared;
+
+
+#define SlruFileName(ctl, path, seg) \
+	snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir, seg)
+
+/*
+ * Macro to mark a buffer slot "most recently used".
+ */
+#define SlruRecentlyUsed(shared, slotno)	\
+	do { \
+		if ((shared)->page_lru_count[slotno] != 0) { \
+			int		iilru; \
+			for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
+				(shared)->page_lru_count[iilru]++; \
+			(shared)->page_lru_count[slotno] = 0; \
+		} \
+	} while (0)
+
+/* Saved info for SlruReportIOError */
+typedef enum
+{
+	SLRU_OPEN_FAILED,
+	SLRU_CREATE_FAILED,
+	SLRU_SEEK_FAILED,
+	SLRU_READ_FAILED,
+	SLRU_WRITE_FAILED
+} SlruErrorCause;
+static SlruErrorCause slru_errcause;
+static int	slru_errno;
+
+
+static bool SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno);
+static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno);
+static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
+static int	SlruSelectLRUPage(SlruCtl ctl, int pageno);
+static bool SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions);
+
+
+/*
+ * Initialization of shared memory
+ */
+
+int
+SimpleLruShmemSize(void)
+{
+	return MAXALIGN(sizeof(SlruSharedData)) + BLCKSZ * NUM_CLOG_BUFFERS
+#ifdef EXEC_BACKEND
+			+ MAXALIGN(sizeof(SlruLockData))
+#endif
+	;
+}
+
+void
+SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir)
+{
+	bool		found;
+	char	   *ptr;
+	SlruShared	shared;
+	SlruLock	locks;
+
+	ptr = ShmemInitStruct(name, SimpleLruShmemSize(), &found);
+	shared = (SlruShared) ptr;
+
+#ifdef EXEC_BACKEND
+	/*
+	 * Locks are in shared memory
+	 */
+	locks = (SlruLock)(ptr + MAXALIGN(sizeof(SlruSharedData)) +
+	                         BLCKSZ * NUM_CLOG_BUFFERS);
+#else
+	/*
+	 * Locks are in private memory
+	 */
+	Assert(!IsUnderPostmaster);
+	locks = malloc(sizeof(SlruLockData));
+	Assert(locks);
+#endif
+
+
+	if (!IsUnderPostmaster)
+	/* Initialize locks and shared memory area */
+	{
+		char	   *bufptr;
+		int			slotno;
+
+		Assert(!found);
+
+		locks->ControlLock = LWLockAssign();
+
+		memset(shared, 0, sizeof(SlruSharedData));
+
+		bufptr = (char *)shared + MAXALIGN(sizeof(SlruSharedData));
+	
+		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+		{
+			locks->BufferLocks[slotno] = LWLockAssign();
+			shared->page_buffer[slotno] = bufptr;
+			shared->page_status[slotno] = SLRU_PAGE_EMPTY;
+			shared->page_lru_count[slotno] = 1;
+			bufptr += BLCKSZ;
+		}
+
+		/* shared->latest_page_number will be set later */
+	}
+	else
+		Assert(found);
+
+
+	ctl->locks = locks;
+	ctl->shared = shared;
+
+
+	/* Init directory path */
+	snprintf(ctl->Dir, MAXPGPATH, "%s/%s", DataDir, subdir);
+}
+
+/*
+ * Initialize (or reinitialize) a page to zeroes.
+ *
+ * The page is not actually written, just set up in shared memory.
+ * The slot number of the new page is returned.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+int
+SimpleLruZeroPage(SlruCtl ctl, int pageno)
+{
+	int			slotno;
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* 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 */
+	MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
+
+	/* Assume this page is now the latest active page */
+	shared->latest_page_number = pageno;
+
+	return slotno;
+}
+
+/*
+ * Find a page in a shared buffer, reading it in if necessary.
+ * The page number must correspond to an already-initialized page.
+ *
+ * The passed-in xid is used only for error reporting, and may be
+ * InvalidTransactionId if no specific xid is associated with the action.
+ *
+ * Return value is the shared-buffer address of the page.
+ * The buffer's LRU access info is updated.
+ * If forwrite is true, the buffer is marked as dirty.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+char *
+SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite)
+{
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* Outer loop handles restart if we lose the buffer to someone else */
+	for (;;)
+	{
+		int			slotno;
+		bool		ok;
+
+		/* See if page already is in memory; if not, pick victim slot */
+		slotno = SlruSelectLRUPage(ctl, pageno);
+
+		/* Did we find the page in memory? */
+		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);
+				if (forwrite)
+					shared->page_status[slotno] = SLRU_PAGE_DIRTY;
+				return shared->page_buffer[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
+		 * SlruSelectLRUPage from selecting it again for someone else.
+		 */
+		SlruRecentlyUsed(shared, slotno);
+
+		/* Release shared lock, grab per-buffer lock instead */
+		LWLockRelease(ctl->locks->ControlLock);
+		LWLockAcquire(ctl->locks->BufferLocks[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(ctl->locks->BufferLocks[slotno]);
+			LWLockAcquire(ctl->locks->ControlLock, LW_EXCLUSIVE);
+			continue;
+		}
+
+		/* Okay, do the read */
+		ok = SlruPhysicalReadPage(ctl, pageno, slotno);
+
+		/* Re-acquire shared control lock and update page state */
+		LWLockAcquire(ctl->locks->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(ctl->locks->BufferLocks[slotno]);
+
+		/* Now it's okay to elog if we failed */
+		if (!ok)
+			SlruReportIOError(ctl, pageno, xid);
+
+		SlruRecentlyUsed(shared, slotno);
+		if (forwrite)
+			shared->page_status[slotno] = SLRU_PAGE_DIRTY;
+		return shared->page_buffer[slotno];
+	}
+}
+
+/*
+ * Write a page from a shared buffer, if necessary.
+ * Does nothing if the specified slot is not dirty.
+ *
+ * NOTE: only one write attempt is made here.  Hence, it is possible that
+ * the page is still dirty at exit (if someone else re-dirtied it during
+ * the write).	However, we *do* attempt a fresh write even if the page
+ * is already being written; this is for checkpoints.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+void
+SimpleLruWritePage(SlruCtl ctl, int slotno)
+{
+	int			pageno;
+	bool		ok;
+	SlruShared shared = (SlruShared) ctl->shared;
+
+	/* 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(ctl->locks->ControlLock);
+	LWLockAcquire(ctl->locks->BufferLocks[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(ctl->locks->BufferLocks[slotno]);
+		LWLockAcquire(ctl->locks->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);
+
+	/* Re-acquire shared control lock and update page state */
+	LWLockAcquire(ctl->locks->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(ctl->locks->BufferLocks[slotno]);
+
+	/* Now it's okay to elog if we failed */
+	if (!ok)
+		SlruReportIOError(ctl, pageno, InvalidTransactionId);
+}
+
+/*
+ * Physical read of a (previously existing) page into a buffer slot
+ *
+ * On failure, we cannot just elog(ERROR) since caller has put state in
+ * shared memory that must be undone.  So, we return FALSE and save enough
+ * info in static variables to let SlruReportIOError make the report.
+ *
+ * For now, assume it's not worth keeping a file pointer open across
+ * read/write operations.  We could cache one virtual file pointer ...
+ */
+static bool
+SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+	int			fd;
+
+	SlruFileName(ctl, path, segno);
+
+	/*
+	 * In a crash-and-restart situation, it's possible for us to receive
+	 * commands to set the commit status of transactions whose bits are in
+	 * already-truncated segments of the commit log (see notes in
+	 * SlruPhysicalWritePage).	Hence, if we are InRecovery, allow the
+	 * case where the file doesn't exist, and return zeroes instead.
+	 */
+	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
+	if (fd < 0)
+	{
+		if (errno != ENOENT || !InRecovery)
+		{
+			slru_errcause = SLRU_OPEN_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+
+		elog(LOG, "file %s doesn't exist, reading as zeroes", path);
+		MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
+		return true;
+	}
+
+	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
+	{
+		slru_errcause = SLRU_SEEK_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	errno = 0;
+	if (read(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
+	{
+		slru_errcause = SLRU_READ_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	close(fd);
+	return true;
+}
+
+/*
+ * Physical write of a page from a buffer slot
+ *
+ * On failure, we cannot just elog(ERROR) since caller has put state in
+ * shared memory that must be undone.  So, we return FALSE and save enough
+ * info in static variables to let SlruReportIOError make the report.
+ *
+ * For now, assume it's not worth keeping a file pointer open across
+ * read/write operations.  We could cache one virtual file pointer ...
+ */
+static bool
+SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+	int			fd;
+
+	SlruFileName(ctl, path, segno);
+
+	/*
+	 * If the file doesn't already exist, we should create it.  It is
+	 * possible for this to need to happen when writing a page that's not
+	 * first in its segment; we assume the OS can cope with that.  (Note:
+	 * it might seem that it'd be okay to create files only when
+	 * SimpleLruZeroPage is called for the first page of a segment.	However,
+	 * if after a crash and restart the REDO logic elects to replay the
+	 * log from a checkpoint before the latest one, then it's possible
+	 * that we will get commands to set transaction status of transactions
+	 * that have already been truncated from the commit log.  Easiest way
+	 * to deal with that is to accept references to nonexistent files here
+	 * and in SlruPhysicalReadPage.)
+	 */
+	fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
+	if (fd < 0)
+	{
+		if (errno != ENOENT)
+		{
+			slru_errcause = SLRU_OPEN_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+
+		fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
+						   S_IRUSR | S_IWUSR);
+		if (fd < 0)
+		{
+			slru_errcause = SLRU_CREATE_FAILED;
+			slru_errno = errno;
+			return false;
+		}
+	}
+
+	if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
+	{
+		slru_errcause = SLRU_SEEK_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	errno = 0;
+	if (write(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
+	{
+		/* if write didn't set errno, assume problem is no disk space */
+		if (errno == 0)
+			errno = ENOSPC;
+		slru_errcause = SLRU_WRITE_FAILED;
+		slru_errno = errno;
+		return false;
+	}
+
+	close(fd);
+	return true;
+}
+
+/*
+ * Issue the error message after failure of SlruPhysicalReadPage or
+ * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
+ */
+static void
+SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
+{
+	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
+	int			offset = rpageno * BLCKSZ;
+	char		path[MAXPGPATH];
+
+	/* XXX TODO: provide xid as context in error messages */
+
+	SlruFileName(ctl, path, segno);
+	errno = slru_errno;
+	switch (slru_errcause)
+	{
+		case SLRU_OPEN_FAILED:
+			elog(ERROR, "open of %s failed: %m", path);
+			break;
+		case SLRU_CREATE_FAILED:
+			elog(ERROR, "creation of file %s failed: %m", path);
+			break;
+		case SLRU_SEEK_FAILED:
+			elog(ERROR, "lseek of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		case SLRU_READ_FAILED:
+			elog(ERROR, "read of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		case SLRU_WRITE_FAILED:
+			elog(ERROR, "write of file %s, offset %u failed: %m",
+				 path, offset);
+			break;
+		default:
+			/* can't get here, we trust */
+			elog(ERROR, "unknown SimpleLru I/O error");
+			break;
+	}
+}
+
+/*
+ * Select the slot to re-use when we need a free slot.
+ *
+ * The target page number is passed because we need to consider the
+ * possibility that some other process reads in the target page while
+ * we are doing I/O to free a slot.  Hence, check or recheck to see if
+ * any slot already holds the target page, and return that slot if so.
+ * Thus, the returned slot is *either* a slot already holding the pageno
+ * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
+ * or CLEAN).
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+static int
+SlruSelectLRUPage(SlruCtl ctl, int pageno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	/* Outer loop handles restart after I/O */
+	for (;;)
+	{
+		int			slotno;
+		int			bestslot = 0;
+		unsigned int bestcount = 0;
+
+		/* See if page already has a buffer assigned */
+		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+		{
+			if (shared->page_number[slotno] == pageno &&
+				shared->page_status[slotno] != SLRU_PAGE_EMPTY)
+				return slotno;
+		}
+
+		/*
+		 * If we find any EMPTY slot, just select that one. Else locate
+		 * the least-recently-used slot that isn't the latest page.
+		 */
+		for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+		{
+			if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
+				return slotno;
+			if (shared->page_lru_count[slotno] > bestcount &&
+			    shared->page_number[slotno] != shared->latest_page_number)
+			{
+				bestslot = slotno;
+				bestcount = shared->page_lru_count[slotno];
+			}
+		}
+
+		/*
+		 * 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, false);
+		else
+			SimpleLruWritePage(ctl, bestslot);
+
+		/*
+		 * Now loop back and try again.  This is the easiest way of
+		 * dealing with corner cases such as the victim page being
+		 * re-dirtied while we wrote it.
+		 */
+	}
+}
+
+/*
+ * This must be called ONCE during postmaster or standalone-backend startup
+ */
+void
+SimpleLruSetLatestPage(SlruCtl ctl, int pageno)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+
+	shared->latest_page_number = pageno;
+}
+
+/*
+ * This is called during checkpoint and postmaster/standalone-backend shutdown
+ */
+void
+SimpleLruFlush(SlruCtl ctl, bool checkpoint)
+{
+	SlruShared	shared = (SlruShared) ctl->shared;
+	int			slotno;
+
+	LWLockAcquire(ctl->locks->ControlLock, LW_EXCLUSIVE);
+
+	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+	{
+		SimpleLruWritePage(ctl, slotno);
+		/*
+		 * When called during a checkpoint,
+		 * we cannot assert that the slot is clean now, since another
+		 * process might have re-dirtied it already.  That's okay.
+		 */
+		Assert(checkpoint ||
+			   shared->page_status[slotno] == SLRU_PAGE_EMPTY ||
+			   shared->page_status[slotno] == SLRU_PAGE_CLEAN);
+	}
+
+	LWLockRelease(ctl->locks->ControlLock);
+}
+
+/*
+ * Remove all segments before the one holding the passed page number
+ *
+ * When this is called, we know that the database logically contains no
+ * reference to transaction IDs older than oldestXact.	However, we must
+ * not remove any segment until we have performed a checkpoint, to ensure
+ * that no such references remain on disk either; else a crash just after
+ * the truncation might leave us with a problem.  Since CLOG segments hold
+ * a large number of transactions, the opportunity to actually remove a
+ * segment is fairly rare, and so it seems best not to do the checkpoint
+ * unless we have confirmed that there is a removable segment.	Therefore
+ * we issue the checkpoint command here, not in higher-level code as might
+ * seem cleaner.
+ */
+void
+SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
+{
+	int			slotno;
+	SlruShared	shared = (SlruShared) ctl->shared;
+
+	/*
+	 * The cutoff point is the start of the segment containing cutoffPage.
+	 */
+	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;
+
+	if (!SlruScanDirectory(ctl, cutoffPage, false))
+		return;					/* nothing to remove */
+
+	/* Perform a forced CHECKPOINT */
+	CreateCheckPoint(false, true);
+
+	/*
+	 * Scan shared memory and remove any pages preceding the cutoff
+	 * page, to ensure we won't rewrite them later.  (Any dirty pages
+	 * should have been flushed already during the checkpoint, we're just
+	 * being extra careful here.)
+	 */
+	LWLockAcquire(ctl->locks->ControlLock, LW_EXCLUSIVE);
+
+restart:;
+
+	/*
+	 * While we are holding the lock, make an important safety check: the
+	 * planned cutoff point must be <= the current endpoint page.
+	 * Otherwise we have already wrapped around, and proceeding with the
+	 * truncation would risk removing the current segment.
+	 */
+	if (ctl->PagePrecedes(shared->latest_page_number, cutoffPage))
+	{
+		LWLockRelease(ctl->locks->ControlLock);
+		elog(LOG, "unable to truncate %s: apparent wraparound", ctl->Dir);
+		return;
+	}
+
+	for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+	{
+		if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
+			continue;
+		if (!ctl->PagePrecedes(shared->page_number[slotno], cutoffPage))
+			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;
+		}
+
+		/*
+		 * 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, false);
+		else
+			SimpleLruWritePage(ctl, slotno);
+		goto restart;
+	}
+
+	LWLockRelease(ctl->locks->ControlLock);
+
+	/* Now we can remove the old segment(s) */
+	(void) SlruScanDirectory(ctl, cutoffPage, true);
+}
+
+/*
+ * SlruTruncate subroutine: scan directory for removable segments.
+ * Actually remove them iff doDeletions is true.  Return TRUE iff any
+ * removable segments were found.  Note: no locking is needed.
+ */
+static bool
+SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
+{
+	bool		found = false;
+	DIR		   *cldir;
+	struct dirent *clde;
+	int			segno;
+	int			segpage;
+	char		path[MAXPGPATH];
+
+	cldir = opendir(ctl->Dir);
+	if (cldir == NULL)
+		elog(ERROR, "could not open directory (%s): %m", ctl->Dir);
+
+	errno = 0;
+	while ((clde = readdir(cldir)) != NULL)
+	{
+		if (strlen(clde->d_name) == 4 &&
+			strspn(clde->d_name, "0123456789ABCDEF") == 4)
+		{
+			segno = (int) strtol(clde->d_name, NULL, 16);
+			segpage = segno * SLRU_PAGES_PER_SEGMENT;
+			if (ctl->PagePrecedes(segpage, cutoffPage))
+			{
+				found = true;
+				if (doDeletions)
+				{
+					elog(LOG, "removing file %s/%s", ctl->Dir, clde->d_name);
+					snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, clde->d_name);
+					unlink(path);
+				}
+			}
+		}
+		errno = 0;
+	}
+	if (errno)
+		elog(ERROR, "could not read directory (%s): %m", ctl->Dir);
+	closedir(cldir);
+
+	return found;
+}
diff -ruN ../base/src/backend/storage/lmgr/lwlock.c src/backend/storage/lmgr/lwlock.c
--- ../base/src/backend/storage/lmgr/lwlock.c	2002-10-10 19:07:27.000000000 +0200
+++ src/backend/storage/lmgr/lwlock.c	2003-05-05 21:22:54.000000000 +0200
@@ -108,8 +108,8 @@
 	/* bufmgr.c needs two for each shared buffer */
 	numLocks += 2 * NBuffers;
 
-	/* clog.c needs one per CLOG buffer */
-	numLocks += NUM_CLOG_BUFFERS;
+	/* clog.c needs one per CLOG buffer + one control lock */
+	numLocks += NUM_CLOG_BUFFERS + 1;
 
 	/* Perhaps create a few more for use by user-defined modules? */
 
diff -ruN ../base/src/include/access/slru.h src/include/access/slru.h
--- ../base/src/include/access/slru.h	1970-01-01 01:00:00.000000000 +0100
+++ src/include/access/slru.h	2003-05-06 00:02:55.000000000 +0200
@@ -0,0 +1,62 @@
+/*
+ * slru.h
+ *
+ * Simple LRU
+ *
+ * Portions Copyright (c) 2003, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $Id: $
+ */
+#ifndef SLRU_H
+#define SLRU_H
+
+#include "access/xlog.h"
+
+/* exported because lwlock.c needs it */
+#define NUM_CLOG_BUFFERS	8
+
+typedef struct SlruLockData
+{
+	LWLockId ControlLock;
+/*
+ * BufferLocks is set during CLOGShmemInit and does not change thereafter.
+ * The value is automatically inherited by backends via fork, and
+ * doesn't need to be in shared memory.
+ */
+	LWLockId BufferLocks[NUM_CLOG_BUFFERS];		/* Per-buffer I/O locks */
+} SlruLockData;
+typedef SlruLockData *SlruLock;
+
+typedef struct SlruCtlData
+{
+	void *shared;	/* pointer to SlruSharedData */
+	SlruLock locks;
+
+/*
+ * Dir is set during SimpleLruShmemInit and does not change thereafter.
+ * The value is automatically inherited by backends via fork, and
+ * doesn't need to be in shared memory.
+ */
+	char Dir[MAXPGPATH];
+	
+/*
+ * Decide which of two page numbers is "older" for truncation purposes.
+ * We need to use comparison of TransactionIds here in order to do the right
+ * thing with wraparound XID arithmetic.
+ */
+	bool (*PagePrecedes)(int, int);
+
+} SlruCtlData;
+typedef SlruCtlData *SlruCtl;
+
+extern int	SimpleLruShmemSize(void);
+extern void SimpleLruInit(SlruCtl ctl, const char *name, const char *subdir);
+extern int	SimpleLruZeroPage(SlruCtl ctl, int pageno);
+extern char	*SimpleLruReadPage(SlruCtl ctl, int pageno, TransactionId xid, bool forwrite);
+extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
+extern void SimpleLruSetLatestPage(SlruCtl ctl, int pageno);
+extern void SimpleLruFlush(SlruCtl ctl, bool checkpoint);
+extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
+
+#endif   /* SLRU_H */
diff -ruN ../base/src/include/storage/lwlock.h src/include/storage/lwlock.h
--- ../base/src/include/storage/lwlock.h	2002-06-21 02:12:30.000000000 +0200
+++ src/include/storage/lwlock.h	2003-05-05 21:22:55.000000000 +0200
@@ -37,7 +37,6 @@
 	WALWriteLock,
 	ControlFileLock,
 	CheckpointLock,
-	CLogControlLock,
 	RelCacheInitLock,
 
 	NumFixedLWLocks,			/* must be last except for