slru-truncate-callbacks.patch

*** a/src/backend/access/transam/clog.c
--- b/src/backend/access/transam/clog.c
***************
*** 606,612 **** TruncateCLOG(TransactionId oldestXact)
  	cutoffPage = TransactionIdToPage(oldestXact);
  
  	/* Check to see if there's any files that could be removed */
! 	if (!SlruScanDirectory(ClogCtl, cutoffPage, false))
  		return;					/* nothing to remove */
  
  	/* Write XLOG record and flush XLOG to disk */
--- 606,612 ----
  	cutoffPage = TransactionIdToPage(oldestXact);
  
  	/* Check to see if there's any files that could be removed */
! 	if (!SlruScanDirectory(ClogCtl, SlruScanDirCbReportPresence, &cutoffPage))
  		return;					/* nothing to remove */
  
  	/* Write XLOG record and flush XLOG to disk */
*** a/src/backend/access/transam/slru.c
--- b/src/backend/access/transam/slru.c
***************
*** 132,137 **** static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno,
--- 132,139 ----
  static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
  static int	SlruSelectLRUPage(SlruCtl ctl, int pageno);
  
+ static bool SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename,
+ 						  int segpage, void *data);
  
  /*
   * Initialization of shared memory
***************
*** 1137,1169 **** restart:;
  	LWLockRelease(shared->ControlLock);
  
  	/* Now we can remove the old segment(s) */
! 	(void) SlruScanDirectory(ctl, cutoffPage, true);
  }
  
  /*
!  * SimpleLruTruncate 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.
   *
!  * This can be called directly from clog.c, for reasons explained there.
   */
  bool
! SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
  {
- 	bool		found = false;
  	DIR		   *cldir;
  	struct dirent *clde;
  	int			segno;
  	int			segpage;
! 	char		path[MAXPGPATH];
! 
! 	/*
! 	 * The cutoff point is the start of the segment containing cutoffPage.
! 	 * (This is redundant when called from SimpleLruTruncate, but not when
! 	 * called directly from clog.c.)
! 	 */
! 	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;
! 
  	cldir = AllocateDir(ctl->Dir);
  	while ((clde = ReadDir(cldir, ctl->Dir)) != NULL)
  	{
--- 1139,1222 ----
  	LWLockRelease(shared->ControlLock);
  
  	/* Now we can remove the old segment(s) */
! 	(void) SlruScanDirectory(ctl, SlruScanDirCbDeleteCutoff, &cutoffPage);
  }
  
  /*
!  * SlruScanDirectory callback
!  * 		This callback reports true if there's any segment prior to the one
!  * 		containing the page passed as "data".
!  */
! bool
! SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int segpage, void *data)
! {
! 	int		cutoffPage = *(int *) data;
! 
! 	cutoffPage -= cutoffPage % SLRU_PAGES_PER_SEGMENT;
! 
! 	if (ctl->PagePrecedes(segpage, cutoffPage))
! 		return true;	/* found one; don't iterate any more */
! 
! 	return false;	/* keep going */
! }
! 
! /*
!  * SlruScanDirectory callback.
!  *		This callback deletes segments prior to the one passed in as "data".
!  */
! static bool
! SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int segpage, void *data)
! {
! 	char	path[MAXPGPATH];
! 	int		cutoffPage = *(int *) data;
! 
! 	if (ctl->PagePrecedes(segpage, cutoffPage))
! 	{
! 		snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, filename);
! 		ereport(DEBUG2,
! 				(errmsg("removing file \"%s\"", path)));
! 		unlink(path);
! 	}
! 
! 	return false;	/* keep going */
! }
! 
! /*
!  * SlruScanDirectory callback.
!  *		This callback deletes all segments.
!  */
! bool
! SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage, void *data)
! {
! 	char	path[MAXPGPATH];
! 
! 	snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, filename);
! 	ereport(DEBUG2,
! 			(errmsg("removing file \"%s\"", path)));
! 	unlink(path);
! 
! 	return false;	/* keep going */
! }
! 
! /*
!  * Scan the SimpleLRU directory and apply a callback to each file found in it.
!  *
!  * If the callback returns true, the scan is stopped.  The last return value
!  * from the callback is returned.
   *
!  * Note that the ordering in which the directory is scanned is not guaranteed.
!  *
!  * Note that no locking is applied.
   */
  bool
! SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data)
  {
  	DIR		   *cldir;
  	struct dirent *clde;
  	int			segno;
  	int			segpage;
! 	bool		retval;
! 	
  	cldir = AllocateDir(ctl->Dir);
  	while ((clde = ReadDir(cldir, ctl->Dir)) != NULL)
  	{
***************
*** 1172,1191 **** SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions)
  		{
  			segno = (int) strtol(clde->d_name, NULL, 16);
  			segpage = segno * SLRU_PAGES_PER_SEGMENT;
! 			if (ctl->PagePrecedes(segpage, cutoffPage))
! 			{
! 				found = true;
! 				if (doDeletions)
! 				{
! 					snprintf(path, MAXPGPATH, "%s/%s", ctl->Dir, clde->d_name);
! 					ereport(DEBUG2,
! 							(errmsg("removing file \"%s\"", path)));
! 					unlink(path);
! 				}
! 			}
  		}
  	}
  	FreeDir(cldir);
  
! 	return found;
  }
--- 1225,1239 ----
  		{
  			segno = (int) strtol(clde->d_name, NULL, 16);
  			segpage = segno * SLRU_PAGES_PER_SEGMENT;
! 
! 			elog(DEBUG2, "SlruScanDirectory invoking callback on %s/%s",
! 				 ctl->Dir, clde->d_name);
! 			retval = callback(ctl, clde->d_name, segpage, data);
! 			if (retval)
! 				break;
  		}
  	}
  	FreeDir(cldir);
  
! 	return retval;
  }
*** a/src/backend/commands/async.c
--- b/src/backend/commands/async.c
***************
*** 194,200 **** typedef struct QueuePosition
  
  /* choose logically smaller QueuePosition */
  #define QUEUE_POS_MIN(x,y) \
! 	(asyncQueuePagePrecedesLogically((x).page, (y).page) ? (x) : \
  	 (x).page != (y).page ? (y) : \
  	 (x).offset < (y).offset ? (x) : (y))
  
--- 194,200 ----
  
  /* choose logically smaller QueuePosition */
  #define QUEUE_POS_MIN(x,y) \
! 	(asyncQueuePagePrecedes((x).page, (y).page) ? (x) : \
  	 (x).page != (y).page ? (y) : \
  	 (x).offset < (y).offset ? (x) : (y))
  
***************
*** 360,367 **** static bool backendHasExecutedInitialListen = false;
  bool		Trace_notify = false;
  
  /* local function prototypes */
! static bool asyncQueuePagePrecedesPhysically(int p, int q);
! static bool asyncQueuePagePrecedesLogically(int p, int q);
  static void queue_listen(ListenActionKind action, const char *channel);
  static void Async_UnlistenOnExit(int code, Datum arg);
  static void Exec_ListenPreCommit(void);
--- 360,366 ----
  bool		Trace_notify = false;
  
  /* local function prototypes */
! static bool asyncQueuePagePrecedes(int p, int q);
  static void queue_listen(ListenActionKind action, const char *channel);
  static void Async_UnlistenOnExit(int code, Datum arg);
  static void Exec_ListenPreCommit(void);
***************
*** 388,412 **** static void NotifyMyFrontEnd(const char *channel,
  static bool AsyncExistsPendingNotify(const char *channel, const char *payload);
  static void ClearPendingActionsAndNotifies(void);
  
- 
  /*
   * We will work on the page range of 0..QUEUE_MAX_PAGE.
-  *
-  * asyncQueuePagePrecedesPhysically just checks numerically without any magic
-  * if one page precedes another one.  This is wrong for normal operation but
-  * is helpful when clearing pg_notify/ during startup.
-  *
-  * asyncQueuePagePrecedesLogically compares using wraparound logic, as is
-  * required by slru.c.
   */
  static bool
! asyncQueuePagePrecedesPhysically(int p, int q)
! {
! 	return p < q;
! }
! 
! static bool
! asyncQueuePagePrecedesLogically(int p, int q)
  {
  	int			diff;
  
--- 387,397 ----
  static bool AsyncExistsPendingNotify(const char *channel, const char *payload);
  static void ClearPendingActionsAndNotifies(void);
  
  /*
   * We will work on the page range of 0..QUEUE_MAX_PAGE.
   */
  static bool
! asyncQueuePagePrecedes(int p, int q)
  {
  	int			diff;
  
***************
*** 484,490 **** AsyncShmemInit(void)
  	/*
  	 * Set up SLRU management of the pg_notify data.
  	 */
! 	AsyncCtl->PagePrecedes = asyncQueuePagePrecedesLogically;
  	SimpleLruInit(AsyncCtl, "Async Ctl", NUM_ASYNC_BUFFERS, 0,
  				  AsyncCtlLock, "pg_notify");
  	/* Override default assumption that writes should be fsync'd */
--- 469,475 ----
  	/*
  	 * Set up SLRU management of the pg_notify data.
  	 */
! 	AsyncCtl->PagePrecedes = asyncQueuePagePrecedes;
  	SimpleLruInit(AsyncCtl, "Async Ctl", NUM_ASYNC_BUFFERS, 0,
  				  AsyncCtlLock, "pg_notify");
  	/* Override default assumption that writes should be fsync'd */
***************
*** 494,508 **** AsyncShmemInit(void)
  	{
  		/*
  		 * During start or reboot, clean out the pg_notify directory.
- 		 *
- 		 * Since we want to remove every file, we temporarily use
- 		 * asyncQueuePagePrecedesPhysically() and pass INT_MAX as the
- 		 * comparison value; every file in the directory should therefore
- 		 * appear to be less than that.
  		 */
! 		AsyncCtl->PagePrecedes = asyncQueuePagePrecedesPhysically;
! 		(void) SlruScanDirectory(AsyncCtl, INT_MAX, true);
! 		AsyncCtl->PagePrecedes = asyncQueuePagePrecedesLogically;
  
  		/* Now initialize page zero to empty */
  		LWLockAcquire(AsyncCtlLock, LW_EXCLUSIVE);
--- 479,486 ----
  	{
  		/*
  		 * During start or reboot, clean out the pg_notify directory.
  		 */
! 		(void) SlruScanDirectory(AsyncCtl, SlruScanDirCbDeleteAll, NULL);
  
  		/* Now initialize page zero to empty */
  		LWLockAcquire(AsyncCtlLock, LW_EXCLUSIVE);
***************
*** 1223,1229 **** asyncQueueIsFull(void)
  		nexthead = 0;			/* wrap around */
  	boundary = QUEUE_POS_PAGE(QUEUE_TAIL);
  	boundary -= boundary % SLRU_PAGES_PER_SEGMENT;
! 	return asyncQueuePagePrecedesLogically(nexthead, boundary);
  }
  
  /*
--- 1201,1207 ----
  		nexthead = 0;			/* wrap around */
  	boundary = QUEUE_POS_PAGE(QUEUE_TAIL);
  	boundary -= boundary % SLRU_PAGES_PER_SEGMENT;
! 	return asyncQueuePagePrecedes(nexthead, boundary);
  }
  
  /*
***************
*** 2074,2080 **** asyncQueueAdvanceTail(void)
  	 */
  	newtailpage = QUEUE_POS_PAGE(min);
  	boundary = newtailpage - (newtailpage % SLRU_PAGES_PER_SEGMENT);
! 	if (asyncQueuePagePrecedesLogically(oldtailpage, boundary))
  	{
  		/*
  		 * SimpleLruTruncate() will ask for AsyncCtlLock but will also release
--- 2052,2058 ----
  	 */
  	newtailpage = QUEUE_POS_PAGE(min);
  	boundary = newtailpage - (newtailpage % SLRU_PAGES_PER_SEGMENT);
! 	if (asyncQueuePagePrecedes(oldtailpage, boundary))
  	{
  		/*
  		 * SimpleLruTruncate() will ask for AsyncCtlLock but will also release
*** a/src/backend/commands/vacuum.c
--- b/src/backend/commands/vacuum.c
***************
*** 674,680 **** vac_update_datfrozenxid(void)
  	 * Initialize the "min" calculation with GetOldestXmin, which is a
  	 * reasonable approximation to the minimum relfrozenxid for not-yet-
  	 * committed pg_class entries for new tables; see AddNewRelationTuple().
! 	 * Se we cannot produce a wrong minimum by starting with this.
  	 */
  	newFrozenXid = GetOldestXmin(true, true);
  
--- 674,680 ----
  	 * Initialize the "min" calculation with GetOldestXmin, which is a
  	 * reasonable approximation to the minimum relfrozenxid for not-yet-
  	 * committed pg_class entries for new tables; see AddNewRelationTuple().
! 	 * So we cannot produce a wrong minimum by starting with this.
  	 */
  	newFrozenXid = GetOldestXmin(true, true);
  
*** a/src/include/access/slru.h
--- b/src/include/access/slru.h
***************
*** 145,150 **** extern int SimpleLruReadPage_ReadOnly(SlruCtl ctl, int pageno,
  extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
  extern void SimpleLruFlush(SlruCtl ctl, bool checkpoint);
  extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
! extern bool SlruScanDirectory(SlruCtl ctl, int cutoffPage, bool doDeletions);
  
  #endif   /* SLRU_H */
--- 145,159 ----
  extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
  extern void SimpleLruFlush(SlruCtl ctl, bool checkpoint);
  extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
! 
! typedef bool (*SlruScanCallback) (SlruCtl ctl, char *filename, int segpage,
! 					 void *data);
! extern bool SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data);
! 
! /* SlruScanDirectory public callbacks */
! extern bool SlruScanDirCbReportPresence(SlruCtl ctl, char *filename,
! 							int segpage, void *data);
! extern bool SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage,
! 					   void *data);
  
  #endif   /* SLRU_H */