Add SLRU tests for 64-bit page case

4ed8f0913b added 64-bit page numbering for SLRU.  This commit adds tests for
page numbers higher than 2^32.

Author: Maxim Orlov
Reviewed-by: Aleksander Alekseev, Alexander Korotkov
Discussion: https://postgr.es/m/CACG%3DezZe1NQSCnfHOr78AtAZxJZeCvxrts0ygrxYwe%3DpyyjVWA%40mail.gmail.com
Discussion: https://postgr.es/m/CAJ7c6TPDOYBYrnCAeyndkBktO0WG2xSdYduTF0nxq%2BvfkmTF5Q%40mail.gmail.com

doc/src/sgml/config.sgml

@@ -2151,6 +2151,22 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-max-notify-queue-pages" xreflabel="max_notify_queue_pages">
+      <term><varname>max_notify_queue_pages</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>max_notify_queue_pages</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies the maximum amount of allocated pages for
+        <xref linkend="sql-notify"/> / <xref linkend="sql-listen"/> queue.
+        The default value is 1048576. For 8 KB pages it allows to consume
+        up to 8 GB of disk space.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
      </sect2>
 

doc/src/sgml/ref/listen.sgml

@@ -148,6 +148,7 @@ Asynchronous notification "virtual" received from server process with PID 8448.
   <simplelist type="inline">
    <member><xref linkend="sql-notify"/></member>
    <member><xref linkend="sql-unlisten"/></member>
+   <member><xref linkend="guc-max-notify-queue-pages"/></member>
   </simplelist>
  </refsect1>
 </refentry>

doc/src/sgml/ref/notify.sgml

@@ -228,6 +228,7 @@ Asynchronous notification "foo" with payload "payload" received from server proc
   <simplelist type="inline">
    <member><xref linkend="sql-listen"/></member>
    <member><xref linkend="sql-unlisten"/></member>
+   <member><xref linkend="guc-max-notify-queue-pages"/></member>
   </simplelist>
  </refsect1>
 </refentry>

src/backend/access/rmgrdesc/clogdesc.c

@@ -25,18 +25,18 @@ clog_desc(StringInfo buf, XLogReaderState *record)
 
 	if (info == CLOG_ZEROPAGE)
 	{
-		int			pageno;
+		int64		pageno;
 
-		memcpy(&pageno, rec, sizeof(int));
+		memcpy(&pageno, rec, sizeof(pageno));
-		appendStringInfo(buf, "page %d", pageno);
+		appendStringInfo(buf, "page %lld", (long long) pageno);
 	}
 	else if (info == CLOG_TRUNCATE)
 	{
 		xl_clog_truncate xlrec;
 
 		memcpy(&xlrec, rec, sizeof(xl_clog_truncate));
-		appendStringInfo(buf, "page %d; oldestXact %u",
+		appendStringInfo(buf, "page %lld; oldestXact %u",
-						 xlrec.pageno, xlrec.oldestXact);
+						 (long long) xlrec.pageno, xlrec.oldestXact);
 	}
 }
 

src/backend/access/rmgrdesc/committsdesc.c

@@ -26,17 +26,17 @@ commit_ts_desc(StringInfo buf, XLogReaderState *record)
 
 	if (info == COMMIT_TS_ZEROPAGE)
 	{
-		int			pageno;
+		int64		pageno;
 
-		memcpy(&pageno, rec, sizeof(int));
+		memcpy(&pageno, rec, sizeof(pageno));
-		appendStringInfo(buf, "%d", pageno);
+		appendStringInfo(buf, "%lld", (long long) pageno);
 	}
 	else if (info == COMMIT_TS_TRUNCATE)
 	{
 		xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) rec;
 
-		appendStringInfo(buf, "pageno %d, oldestXid %u",
+		appendStringInfo(buf, "pageno %lld, oldestXid %u",
-						 trunc->pageno, trunc->oldestXid);
+						 (long long) trunc->pageno, trunc->oldestXid);
 	}
 }
 

src/backend/access/rmgrdesc/mxactdesc.c

@@ -55,10 +55,10 @@ multixact_desc(StringInfo buf, XLogReaderState *record)
 	if (info == XLOG_MULTIXACT_ZERO_OFF_PAGE ||
 		info == XLOG_MULTIXACT_ZERO_MEM_PAGE)
 	{
-		int			pageno;
+		int64		pageno;
 
-		memcpy(&pageno, rec, sizeof(int));
+		memcpy(&pageno, rec, sizeof(pageno));
-		appendStringInfo(buf, "%d", pageno);
+		appendStringInfo(buf, "%lld", (long long) pageno);
 	}
 	else if (info == XLOG_MULTIXACT_CREATE_ID)
 	{

src/backend/access/transam/clog.c

@@ -62,7 +62,17 @@
 #define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
 #define CLOG_XACT_BITMASK	((1 << CLOG_BITS_PER_XACT) - 1)
 
-#define TransactionIdToPage(xid)	((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
+
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToPage(TransactionId xid)
+{
+	return xid / (int64) CLOG_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToByte(xid)	(TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
 #define TransactionIdToBIndex(xid)	((xid) % (TransactionId) CLOG_XACTS_PER_BYTE)
@@ -89,24 +99,24 @@ static SlruCtlData XactCtlData;
 #define XactCtl (&XactCtlData)
 
 
-static int	ZeroCLOGPage(int pageno, bool writeXlog);
+static int	ZeroCLOGPage(int64 pageno, bool writeXlog);
-static bool CLOGPagePrecedes(int page1, int page2);
+static bool CLOGPagePrecedes(int64 page1, int64 page2);
-static void WriteZeroPageXlogRec(int pageno);
+static void WriteZeroPageXlogRec(int64 pageno);
-static void WriteTruncateXlogRec(int pageno, TransactionId oldestXact,
+static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXact,
 								 Oid oldestXactDb);
 static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
 									   TransactionId *subxids, XidStatus status,
-									   XLogRecPtr lsn, int pageno,
+									   XLogRecPtr lsn, int64 pageno,
 									   bool all_xact_same_page);
 static void TransactionIdSetStatusBit(TransactionId xid, XidStatus status,
 									  XLogRecPtr lsn, int slotno);
 static void set_status_by_pages(int nsubxids, TransactionId *subxids,
 								XidStatus status, XLogRecPtr lsn);
 static bool TransactionGroupUpdateXidStatus(TransactionId xid,
-											XidStatus status, XLogRecPtr lsn, int pageno);
+											XidStatus status, XLogRecPtr lsn, int64 pageno);
 static void TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
 											   TransactionId *subxids, XidStatus status,
-											   XLogRecPtr lsn, int pageno);
+											   XLogRecPtr lsn, int64 pageno);
 
 
 /*
@@ -162,7 +172,7 @@ void
 TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
 						   TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
 {
-	int			pageno = TransactionIdToPage(xid);	/* get page of parent */
+	int64		pageno = TransactionIdToPage(xid);	/* get page of parent */
 	int			i;
 
 	Assert(status == TRANSACTION_STATUS_COMMITTED ||
@@ -236,7 +246,7 @@ static void
 set_status_by_pages(int nsubxids, TransactionId *subxids,
 					XidStatus status, XLogRecPtr lsn)
 {
-	int			pageno = TransactionIdToPage(subxids[0]);
+	int64		pageno = TransactionIdToPage(subxids[0]);
 	int			offset = 0;
 	int			i = 0;
 
@@ -245,7 +255,7 @@ set_status_by_pages(int nsubxids, TransactionId *subxids,
 	while (i < nsubxids)
 	{
 		int			num_on_page = 0;
-		int			nextpageno;
+		int64		nextpageno;
 
 		do
 		{
@@ -271,7 +281,7 @@ set_status_by_pages(int nsubxids, TransactionId *subxids,
 static void
 TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
 						   TransactionId *subxids, XidStatus status,
-						   XLogRecPtr lsn, int pageno,
+						   XLogRecPtr lsn, int64 pageno,
 						   bool all_xact_same_page)
 {
 	/* Can't use group update when PGPROC overflows. */
@@ -337,7 +347,7 @@ TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
 static void
 TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
 								   TransactionId *subxids, XidStatus status,
-								   XLogRecPtr lsn, int pageno)
+								   XLogRecPtr lsn, int64 pageno)
 {
 	int			slotno;
 	int			i;
@@ -411,7 +421,7 @@ TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
  */
 static bool
 TransactionGroupUpdateXidStatus(TransactionId xid, XidStatus status,
-								XLogRecPtr lsn, int pageno)
+								XLogRecPtr lsn, int64 pageno)
 {
 	volatile PROC_HDR *procglobal = ProcGlobal;
 	PGPROC	   *proc = MyProc;
@@ -637,7 +647,7 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
 XidStatus
 TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
 {
-	int			pageno = TransactionIdToPage(xid);
+	int64		pageno = TransactionIdToPage(xid);
 	int			byteno = TransactionIdToByte(xid);
 	int			bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
 	int			slotno;
@@ -697,7 +707,7 @@ CLOGShmemInit(void)
 	XactCtl->PagePrecedes = CLOGPagePrecedes;
 	SimpleLruInit(XactCtl, "Xact", CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE,
 				  XactSLRULock, "pg_xact", LWTRANCHE_XACT_BUFFER,
-				  SYNC_HANDLER_CLOG);
+				  SYNC_HANDLER_CLOG, false);
 	SlruPagePrecedesUnitTests(XactCtl, CLOG_XACTS_PER_PAGE);
 }
 
@@ -734,7 +744,7 @@ BootStrapCLOG(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroCLOGPage(int pageno, bool writeXlog)
+ZeroCLOGPage(int64 pageno, bool writeXlog)
 {
 	int			slotno;
 
@@ -754,7 +764,7 @@ void
 StartupCLOG(void)
 {
 	TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-	int			pageno = TransactionIdToPage(xid);
+	int64		pageno = TransactionIdToPage(xid);
 
 	LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 
@@ -773,7 +783,7 @@ void
 TrimCLOG(void)
 {
 	TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-	int			pageno = TransactionIdToPage(xid);
+	int64		pageno = TransactionIdToPage(xid);
 
 	LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 
@@ -838,7 +848,7 @@ CheckPointCLOG(void)
 void
 ExtendCLOG(TransactionId newestXact)
 {
-	int			pageno;
+	int64		pageno;
 
 	/*
 	 * No work except at first XID of a page.  But beware: just after
@@ -877,7 +887,7 @@ ExtendCLOG(TransactionId newestXact)
 void
 TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid)
 {
-	int			cutoffPage;
+	int64		cutoffPage;
 
 	/*
 	 * The cutoff point is the start of the segment containing oldestXact. We
@@ -930,7 +940,7 @@ TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid)
  * don't optimize that edge case.
  */
 static bool
-CLOGPagePrecedes(int page1, int page2)
+CLOGPagePrecedes(int64 page1, int64 page2)
 {
 	TransactionId xid1;
 	TransactionId xid2;
@@ -949,10 +959,10 @@ CLOGPagePrecedes(int page1, int page2)
  * Write a ZEROPAGE xlog record
  */
 static void
-WriteZeroPageXlogRec(int pageno)
+WriteZeroPageXlogRec(int64 pageno)
 {
 	XLogBeginInsert();
-	XLogRegisterData((char *) (&pageno), sizeof(int));
+	XLogRegisterData((char *) (&pageno), sizeof(pageno));
 	(void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE);
 }
 
@@ -963,7 +973,7 @@ WriteZeroPageXlogRec(int pageno)
  * in TruncateCLOG().
  */
 static void
-WriteTruncateXlogRec(int pageno, TransactionId oldestXact, Oid oldestXactDb)
+WriteTruncateXlogRec(int64 pageno, TransactionId oldestXact, Oid oldestXactDb)
 {
 	XLogRecPtr	recptr;
 	xl_clog_truncate xlrec;
@@ -991,10 +1001,10 @@ clog_redo(XLogReaderState *record)
 
 	if (info == CLOG_ZEROPAGE)
 	{
-		int			pageno;
+		int64		pageno;
 		int			slotno;
 
-		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
 		LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 

src/backend/access/transam/commit_ts.c

@@ -65,8 +65,17 @@ typedef struct CommitTimestampEntry
 #define COMMIT_TS_XACTS_PER_PAGE \
 	(BLCKSZ / SizeOfCommitTimestampEntry)
 
-#define TransactionIdToCTsPage(xid) \
+
-	((xid) / (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToCTsPage(TransactionId xid)
+{
+	return xid / (int64) COMMIT_TS_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToCTsEntry(xid)	\
 	((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
 
@@ -103,16 +112,16 @@ bool		track_commit_timestamp;
 
 static void SetXidCommitTsInPage(TransactionId xid, int nsubxids,
 								 TransactionId *subxids, TimestampTz ts,
-								 RepOriginId nodeid, int pageno);
+								 RepOriginId nodeid, int64 pageno);
 static void TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts,
 									 RepOriginId nodeid, int slotno);
 static void error_commit_ts_disabled(void);
-static int	ZeroCommitTsPage(int pageno, bool writeXlog);
+static int	ZeroCommitTsPage(int64 pageno, bool writeXlog);
-static bool CommitTsPagePrecedes(int page1, int page2);
+static bool CommitTsPagePrecedes(int64 page1, int64 page2);
 static void ActivateCommitTs(void);
 static void DeactivateCommitTs(void);
-static void WriteZeroPageXlogRec(int pageno);
+static void WriteZeroPageXlogRec(int64 pageno);
-static void WriteTruncateXlogRec(int pageno, TransactionId oldestXid);
+static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid);
 
 /*
  * TransactionTreeSetCommitTsData
@@ -170,7 +179,7 @@ TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids,
 	i = 0;
 	for (;;)
 	{
-		int			pageno = TransactionIdToCTsPage(headxid);
+		int64		pageno = TransactionIdToCTsPage(headxid);
 		int			j;
 
 		for (j = i; j < nsubxids; j++)
@@ -214,7 +223,7 @@ TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids,
 static void
 SetXidCommitTsInPage(TransactionId xid, int nsubxids,
 					 TransactionId *subxids, TimestampTz ts,
-					 RepOriginId nodeid, int pageno)
+					 RepOriginId nodeid, int64 pageno)
 {
 	int			slotno;
 	int			i;
@@ -266,7 +275,7 @@ bool
 TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts,
 							 RepOriginId *nodeid)
 {
-	int			pageno = TransactionIdToCTsPage(xid);
+	int64		pageno = TransactionIdToCTsPage(xid);
 	int			entryno = TransactionIdToCTsEntry(xid);
 	int			slotno;
 	CommitTimestampEntry entry;
@@ -523,7 +532,8 @@ CommitTsShmemInit(void)
 	SimpleLruInit(CommitTsCtl, "CommitTs", CommitTsShmemBuffers(), 0,
 				  CommitTsSLRULock, "pg_commit_ts",
 				  LWTRANCHE_COMMITTS_BUFFER,
-				  SYNC_HANDLER_COMMIT_TS);
+				  SYNC_HANDLER_COMMIT_TS,
+				  false);
 	SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);
 
 	commitTsShared = ShmemInitStruct("CommitTs shared",
@@ -569,7 +579,7 @@ BootStrapCommitTs(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroCommitTsPage(int pageno, bool writeXlog)
+ZeroCommitTsPage(int64 pageno, bool writeXlog)
 {
 	int			slotno;
 
@@ -662,7 +672,7 @@ static void
 ActivateCommitTs(void)
 {
 	TransactionId xid;
-	int			pageno;
+	int64		pageno;
 
 	/* If we've done this already, there's nothing to do */
 	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
@@ -795,7 +805,7 @@ CheckPointCommitTs(void)
 void
 ExtendCommitTs(TransactionId newestXact)
 {
-	int			pageno;
+	int64		pageno;
 
 	/*
 	 * Nothing to do if module not enabled.  Note we do an unlocked read of
@@ -833,7 +843,7 @@ ExtendCommitTs(TransactionId newestXact)
 void
 TruncateCommitTs(TransactionId oldestXact)
 {
-	int			cutoffPage;
+	int64		cutoffPage;
 
 	/*
 	 * The cutoff point is the start of the segment containing oldestXact. We
@@ -918,7 +928,7 @@ AdvanceOldestCommitTsXid(TransactionId oldestXact)
  * oldestXact=N+2.1, it would be precious at oldestXact=N+2.9.
  */
 static bool
-CommitTsPagePrecedes(int page1, int page2)
+CommitTsPagePrecedes(int64 page1, int64 page2)
 {
 	TransactionId xid1;
 	TransactionId xid2;
@@ -937,10 +947,10 @@ CommitTsPagePrecedes(int page1, int page2)
  * Write a ZEROPAGE xlog record
  */
 static void
-WriteZeroPageXlogRec(int pageno)
+WriteZeroPageXlogRec(int64 pageno)
 {
 	XLogBeginInsert();
-	XLogRegisterData((char *) (&pageno), sizeof(int));
+	XLogRegisterData((char *) (&pageno), sizeof(pageno));
 	(void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
 }
 
@@ -948,7 +958,7 @@ WriteZeroPageXlogRec(int pageno)
  * Write a TRUNCATE xlog record
  */
 static void
-WriteTruncateXlogRec(int pageno, TransactionId oldestXid)
+WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid)
 {
 	xl_commit_ts_truncate xlrec;
 
@@ -973,10 +983,10 @@ commit_ts_redo(XLogReaderState *record)
 
 	if (info == COMMIT_TS_ZEROPAGE)
 	{
-		int			pageno;
+		int64		pageno;
 		int			slotno;
 
-		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
 		LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
 

src/backend/access/transam/multixact.c

@@ -354,10 +354,10 @@ static void mXactCachePut(MultiXactId multi, int nmembers,
 static char *mxstatus_to_string(MultiXactStatus status);
 
 /* management of SLRU infrastructure */
-static int	ZeroMultiXactOffsetPage(int pageno, bool writeXlog);
+static int	ZeroMultiXactOffsetPage(int64 pageno, bool writeXlog);
-static int	ZeroMultiXactMemberPage(int pageno, bool writeXlog);
+static int	ZeroMultiXactMemberPage(int64 pageno, bool writeXlog);
-static bool MultiXactOffsetPagePrecedes(int page1, int page2);
+static bool MultiXactOffsetPagePrecedes(int64 page1, int64 page2);
-static bool MultiXactMemberPagePrecedes(int page1, int page2);
+static bool MultiXactMemberPagePrecedes(int64 page1, int64 page2);
 static bool MultiXactOffsetPrecedes(MultiXactOffset offset1,
 									MultiXactOffset offset2);
 static void ExtendMultiXactOffset(MultiXactId multi);
@@ -366,7 +366,7 @@ static bool MultiXactOffsetWouldWrap(MultiXactOffset boundary,
 									 MultiXactOffset start, uint32 distance);
 static bool SetOffsetVacuumLimit(bool is_startup);
 static bool find_multixact_start(MultiXactId multi, MultiXactOffset *result);
-static void WriteMZeroPageXlogRec(int pageno, uint8 info);
+static void WriteMZeroPageXlogRec(int64 pageno, uint8 info);
 static void WriteMTruncateXlogRec(Oid oldestMultiDB,
 								  MultiXactId startTruncOff,
 								  MultiXactId endTruncOff,
@@ -864,8 +864,8 @@ static void
 RecordNewMultiXact(MultiXactId multi, MultiXactOffset offset,
 				   int nmembers, MultiXactMember *members)
 {
-	int			pageno;
+	int64		pageno;
-	int			prev_pageno;
+	int64		prev_pageno;
 	int			entryno;
 	int			slotno;
 	MultiXactOffset *offptr;
@@ -1225,8 +1225,8 @@ int
 GetMultiXactIdMembers(MultiXactId multi, MultiXactMember **members,
 					  bool from_pgupgrade, bool isLockOnly)
 {
-	int			pageno;
+	int64		pageno;
-	int			prev_pageno;
+	int64		prev_pageno;
 	int			entryno;
 	int			slotno;
 	MultiXactOffset *offptr;
@@ -1854,13 +1854,15 @@ MultiXactShmemInit(void)
 				  "MultiXactOffset", NUM_MULTIXACTOFFSET_BUFFERS, 0,
 				  MultiXactOffsetSLRULock, "pg_multixact/offsets",
 				  LWTRANCHE_MULTIXACTOFFSET_BUFFER,
-				  SYNC_HANDLER_MULTIXACT_OFFSET);
+				  SYNC_HANDLER_MULTIXACT_OFFSET,
+				  false);
 	SlruPagePrecedesUnitTests(MultiXactOffsetCtl, MULTIXACT_OFFSETS_PER_PAGE);
 	SimpleLruInit(MultiXactMemberCtl,
 				  "MultiXactMember", NUM_MULTIXACTMEMBER_BUFFERS, 0,
 				  MultiXactMemberSLRULock, "pg_multixact/members",
 				  LWTRANCHE_MULTIXACTMEMBER_BUFFER,
-				  SYNC_HANDLER_MULTIXACT_MEMBER);
+				  SYNC_HANDLER_MULTIXACT_MEMBER,
+				  false);
 	/* doesn't call SimpleLruTruncate() or meet criteria for unit tests */
 
 	/* Initialize our shared state struct */
@@ -1928,7 +1930,7 @@ BootStrapMultiXact(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroMultiXactOffsetPage(int pageno, bool writeXlog)
+ZeroMultiXactOffsetPage(int64 pageno, bool writeXlog)
 {
 	int			slotno;
 
@@ -1944,7 +1946,7 @@ ZeroMultiXactOffsetPage(int pageno, bool writeXlog)
  * Ditto, for MultiXactMember
  */
 static int
-ZeroMultiXactMemberPage(int pageno, bool writeXlog)
+ZeroMultiXactMemberPage(int64 pageno, bool writeXlog)
 {
 	int			slotno;
 
@@ -1974,7 +1976,7 @@ ZeroMultiXactMemberPage(int pageno, bool writeXlog)
 static void
 MaybeExtendOffsetSlru(void)
 {
-	int			pageno;
+	int64		pageno;
 
 	pageno = MultiXactIdToOffsetPage(MultiXactState->nextMXact);
 
@@ -2009,7 +2011,7 @@ StartupMultiXact(void)
 {
 	MultiXactId multi = MultiXactState->nextMXact;
 	MultiXactOffset offset = MultiXactState->nextOffset;
-	int			pageno;
+	int64		pageno;
 
 	/*
 	 * Initialize offset's idea of the latest page number.
@@ -2034,7 +2036,7 @@ TrimMultiXact(void)
 	MultiXactOffset offset;
 	MultiXactId oldestMXact;
 	Oid			oldestMXactDB;
-	int			pageno;
+	int64		pageno;
 	int			entryno;
 	int			flagsoff;
 
@@ -2403,7 +2405,7 @@ MultiXactAdvanceOldest(MultiXactId oldestMulti, Oid oldestMultiDB)
 static void
 ExtendMultiXactOffset(MultiXactId multi)
 {
-	int			pageno;
+	int64		pageno;
 
 	/*
 	 * No work except at first MultiXactId of a page.  But beware: just after
@@ -2452,7 +2454,7 @@ ExtendMultiXactMember(MultiXactOffset offset, int nmembers)
 		flagsbit = MXOffsetToFlagsBitShift(offset);
 		if (flagsoff == 0 && flagsbit == 0)
 		{
-			int			pageno;
+			int64		pageno;
 
 			pageno = MXOffsetToMemberPage(offset);
 
@@ -2735,7 +2737,7 @@ static bool
 find_multixact_start(MultiXactId multi, MultiXactOffset *result)
 {
 	MultiXactOffset offset;
-	int			pageno;
+	int64		pageno;
 	int			entryno;
 	int			slotno;
 	MultiXactOffset *offptr;
@@ -2854,7 +2856,7 @@ MultiXactMemberFreezeThreshold(void)
 
 typedef struct mxtruncinfo
 {
-	int			earliestExistingPage;
+	int64		earliestExistingPage;
 } mxtruncinfo;
 
 /*
@@ -2862,7 +2864,7 @@ typedef struct mxtruncinfo
  *		This callback determines the earliest existing page number.
  */
 static bool
-SlruScanDirCbFindEarliest(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbFindEarliest(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
 	mxtruncinfo *trunc = (mxtruncinfo *) data;
 
@@ -3113,7 +3115,7 @@ TruncateMultiXact(MultiXactId newOldestMulti, Oid newOldestMultiDB)
  * translational symmetry.
  */
 static bool
-MultiXactOffsetPagePrecedes(int page1, int page2)
+MultiXactOffsetPagePrecedes(int64 page1, int64 page2)
 {
 	MultiXactId multi1;
 	MultiXactId multi2;
@@ -3133,7 +3135,7 @@ MultiXactOffsetPagePrecedes(int page1, int page2)
  * purposes.  There is no "invalid offset number" so use the numbers verbatim.
  */
 static bool
-MultiXactMemberPagePrecedes(int page1, int page2)
+MultiXactMemberPagePrecedes(int64 page1, int64 page2)
 {
 	MultiXactOffset offset1;
 	MultiXactOffset offset2;
@@ -3191,10 +3193,10 @@ MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2)
  * OFFSETs page (info shows which)
  */
 static void
-WriteMZeroPageXlogRec(int pageno, uint8 info)
+WriteMZeroPageXlogRec(int64 pageno, uint8 info)
 {
 	XLogBeginInsert();
-	XLogRegisterData((char *) (&pageno), sizeof(int));
+	XLogRegisterData((char *) (&pageno), sizeof(pageno));
 	(void) XLogInsert(RM_MULTIXACT_ID, info);
 }
 
@@ -3239,10 +3241,10 @@ multixact_redo(XLogReaderState *record)
 
 	if (info == XLOG_MULTIXACT_ZERO_OFF_PAGE)
 	{
-		int			pageno;
+		int64		pageno;
 		int			slotno;
 
-		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
 		LWLockAcquire(MultiXactOffsetSLRULock, LW_EXCLUSIVE);
 
@@ -3254,10 +3256,10 @@ multixact_redo(XLogReaderState *record)
 	}
 	else if (info == XLOG_MULTIXACT_ZERO_MEM_PAGE)
 	{
-		int			pageno;
+		int64		pageno;
 		int			slotno;
 
-		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
 		LWLockAcquire(MultiXactMemberSLRULock, LW_EXCLUSIVE);
 
@@ -3299,7 +3301,7 @@ multixact_redo(XLogReaderState *record)
 	else if (info == XLOG_MULTIXACT_TRUNCATE_ID)
 	{
 		xl_multixact_truncate xlrec;
-		int			pageno;
+		int64		pageno;
 
 		memcpy(&xlrec, XLogRecGetData(record),
 			   SizeOfMultiXactTruncate);

src/backend/access/transam/slru.c

@@ -60,8 +60,33 @@
 #include "storage/fd.h"
 #include "storage/shmem.h"
 
-#define SlruFileName(ctl, path, seg) \
+static int	inline
-	snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir, seg)
+SlruFileName(SlruCtl ctl, char *path, int64 segno)
+{
+	if (ctl->long_segment_names)
+	{
+		/*
+		 * We could use 16 characters here but the disadvantage would be that
+		 * the SLRU segments will be hard to distinguish from WAL segments.
+		 *
+		 * For this reason we use 15 characters. It is enough but also means
+		 * that in the future we can't decrease SLRU_PAGES_PER_SEGMENT easily.
+		 */
+		Assert(segno >= 0 && segno <= INT64CONST(0xFFFFFFFFFFFFFFF));
+		return snprintf(path, MAXPGPATH, "%s/%015llX", ctl->Dir,
+						(long long) segno);
+	}
+	else
+	{
+		/*
+		 * Despite the fact that %04X format string is used up to 24 bit
+		 * integers are allowed. See SlruCorrectSegmentFilenameLength()
+		 */
+		Assert(segno >= 0 && segno <= INT64CONST(0xFFFFFF));
+		return snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir,
+						(unsigned int) segno);
+	}
+}
 
 /*
  * During SimpleLruWriteAll(), we will usually not need to write more than one
@@ -75,7 +100,7 @@ typedef struct SlruWriteAllData
 {
 	int			num_files;		/* # files actually open */
 	int			fd[MAX_WRITEALL_BUFFERS];	/* their FD's */
-	int			segno[MAX_WRITEALL_BUFFERS];	/* their log seg#s */
+	int64		segno[MAX_WRITEALL_BUFFERS];	/* their log seg#s */
 } SlruWriteAllData;
 
 typedef struct SlruWriteAllData *SlruWriteAll;
@@ -138,15 +163,16 @@ static int	slru_errno;
 static void SimpleLruZeroLSNs(SlruCtl ctl, int slotno);
 static void SimpleLruWaitIO(SlruCtl ctl, int slotno);
 static void SlruInternalWritePage(SlruCtl ctl, int slotno, SlruWriteAll fdata);
-static bool SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno);
+static bool SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno);
-static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno,
+static bool SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno,
 								  SlruWriteAll fdata);
-static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
+static void SlruReportIOError(SlruCtl ctl, int64 pageno, TransactionId xid);
-static int	SlruSelectLRUPage(SlruCtl ctl, int pageno);
+static int	SlruSelectLRUPage(SlruCtl ctl, int64 pageno);
 
 static bool SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename,
-									  int segpage, void *data);
+									  int64 segpage, void *data);
-static void SlruInternalDeleteSegment(SlruCtl ctl, int segno);
+static void SlruInternalDeleteSegment(SlruCtl ctl, int64 segno);
+
 
 /*
  * Initialization of shared memory
@@ -162,7 +188,7 @@ SimpleLruShmemSize(int nslots, int nlsns)
 	sz += MAXALIGN(nslots * sizeof(char *));	/* page_buffer[] */
 	sz += MAXALIGN(nslots * sizeof(SlruPageStatus));	/* page_status[] */
 	sz += MAXALIGN(nslots * sizeof(bool));	/* page_dirty[] */
-	sz += MAXALIGN(nslots * sizeof(int));	/* page_number[] */
+	sz += MAXALIGN(nslots * sizeof(int64)); /* page_number[] */
 	sz += MAXALIGN(nslots * sizeof(int));	/* page_lru_count[] */
 	sz += MAXALIGN(nslots * sizeof(LWLockPadded));	/* buffer_locks[] */
 
@@ -187,7 +213,7 @@ SimpleLruShmemSize(int nslots, int nlsns)
 void
 SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
 			  LWLock *ctllock, const char *subdir, int tranche_id,
-			  SyncRequestHandler sync_handler)
+			  SyncRequestHandler sync_handler, bool long_segment_names)
 {
 	SlruShared	shared;
 	bool		found;
@@ -226,8 +252,8 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
 		offset += MAXALIGN(nslots * sizeof(SlruPageStatus));
 		shared->page_dirty = (bool *) (ptr + offset);
 		offset += MAXALIGN(nslots * sizeof(bool));
-		shared->page_number = (int *) (ptr + offset);
+		shared->page_number = (int64 *) (ptr + offset);
-		offset += MAXALIGN(nslots * sizeof(int));
+		offset += MAXALIGN(nslots * sizeof(int64));
 		shared->page_lru_count = (int *) (ptr + offset);
 		offset += MAXALIGN(nslots * sizeof(int));
 
@@ -266,6 +292,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
 	 */
 	ctl->shared = shared;
 	ctl->sync_handler = sync_handler;
+	ctl->long_segment_names = long_segment_names;
 	strlcpy(ctl->Dir, subdir, sizeof(ctl->Dir));
 }
 
@@ -278,7 +305,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
  * Control lock must be held at entry, and will be held at exit.
  */
 int
-SimpleLruZeroPage(SlruCtl ctl, int pageno)
+SimpleLruZeroPage(SlruCtl ctl, int64 pageno)
 {
 	SlruShared	shared = ctl->shared;
 	int			slotno;
@@ -393,7 +420,7 @@ SimpleLruWaitIO(SlruCtl ctl, int slotno)
  * Control lock must be held at entry, and will be held at exit.
  */
 int
-SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
+SimpleLruReadPage(SlruCtl ctl, int64 pageno, bool write_ok,
 				  TransactionId xid)
 {
 	SlruShared	shared = ctl->shared;
@@ -493,7 +520,7 @@ SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
  * It is unspecified whether the lock will be shared or exclusive.
  */
 int
-SimpleLruReadPage_ReadOnly(SlruCtl ctl, int pageno, TransactionId xid)
+SimpleLruReadPage_ReadOnly(SlruCtl ctl, int64 pageno, TransactionId xid)
 {
 	SlruShared	shared = ctl->shared;
 	int			slotno;
@@ -540,7 +567,7 @@ static void
 SlruInternalWritePage(SlruCtl ctl, int slotno, SlruWriteAll fdata)
 {
 	SlruShared	shared = ctl->shared;
-	int			pageno = shared->page_number[slotno];
+	int64		pageno = shared->page_number[slotno];
 	bool		ok;
 
 	/* If a write is in progress, wait for it to finish */
@@ -624,9 +651,9 @@ SimpleLruWritePage(SlruCtl ctl, int slotno)
  * large enough to contain the given page.
  */
 bool
-SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno)
+SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int64 pageno)
 {
-	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int64		segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
 	int			offset = rpageno * BLCKSZ;
 	char		path[MAXPGPATH];
@@ -682,10 +709,10 @@ SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno)
  * read/write operations.  We could cache one virtual file pointer ...
  */
 static bool
-SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
+SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno)
 {
 	SlruShared	shared = ctl->shared;
-	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int64		segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
 	off_t		offset = rpageno * BLCKSZ;
 	char		path[MAXPGPATH];
@@ -754,10 +781,10 @@ SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
  * SimpleLruWriteAll.
  */
 static bool
-SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno, SlruWriteAll fdata)
+SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
 {
 	SlruShared	shared = ctl->shared;
-	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int64		segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
 	off_t		offset = rpageno * BLCKSZ;
 	char		path[MAXPGPATH];
@@ -929,9 +956,9 @@ SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno, SlruWriteAll fdata)
  * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
  */
 static void
-SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
+SlruReportIOError(SlruCtl ctl, int64 pageno, TransactionId xid)
 {
-	int			segno = pageno / SLRU_PAGES_PER_SEGMENT;
+	int64		segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	int			rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
 	int			offset = rpageno * BLCKSZ;
 	char		path[MAXPGPATH];
@@ -1014,7 +1041,7 @@ SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-SlruSelectLRUPage(SlruCtl ctl, int pageno)
+SlruSelectLRUPage(SlruCtl ctl, int64 pageno)
 {
 	SlruShared	shared = ctl->shared;
 
@@ -1025,10 +1052,10 @@ SlruSelectLRUPage(SlruCtl ctl, int pageno)
 		int			cur_count;
 		int			bestvalidslot = 0;	/* keep compiler quiet */
 		int			best_valid_delta = -1;
-		int			best_valid_page_number = 0; /* keep compiler quiet */
+		int64		best_valid_page_number = 0; /* keep compiler quiet */
 		int			bestinvalidslot = 0;	/* keep compiler quiet */
 		int			best_invalid_delta = -1;
-		int			best_invalid_page_number = 0;	/* keep compiler quiet */
+		int64		best_invalid_page_number = 0;	/* keep compiler quiet */
 
 		/* See if page already has a buffer assigned */
 		for (slotno = 0; slotno < shared->num_slots; slotno++)
@@ -1069,7 +1096,7 @@ SlruSelectLRUPage(SlruCtl ctl, int pageno)
 		for (slotno = 0; slotno < shared->num_slots; slotno++)
 		{
 			int			this_delta;
-			int			this_page_number;
+			int64		this_page_number;
 
 			if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
 				return slotno;
@@ -1159,7 +1186,7 @@ SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied)
 	SlruShared	shared = ctl->shared;
 	SlruWriteAllData fdata;
 	int			slotno;
-	int			pageno = 0;
+	int64		pageno = 0;
 	int			i;
 	bool		ok;
 
@@ -1224,7 +1251,7 @@ SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied)
  * after it has accrued freshly-written data.
  */
 void
-SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
+SimpleLruTruncate(SlruCtl ctl, int64 cutoffPage)
 {
 	SlruShared	shared = ctl->shared;
 	int			slotno;
@@ -1302,7 +1329,7 @@ restart:
  * they either can't yet contain anything, or have already been cleaned out.
  */
 static void
-SlruInternalDeleteSegment(SlruCtl ctl, int segno)
+SlruInternalDeleteSegment(SlruCtl ctl, int64 segno)
 {
 	char		path[MAXPGPATH];
 
@@ -1325,7 +1352,7 @@ SlruInternalDeleteSegment(SlruCtl ctl, int segno)
  * Delete an individual SLRU segment, identified by the segment number.
  */
 void
-SlruDeleteSegment(SlruCtl ctl, int segno)
+SlruDeleteSegment(SlruCtl ctl, int64 segno)
 {
 	SlruShared	shared = ctl->shared;
 	int			slotno;
@@ -1389,9 +1416,9 @@ restart:
  * first>=cutoff && last>=cutoff: no; every page of this segment is too young
  */
 static bool
-SlruMayDeleteSegment(SlruCtl ctl, int segpage, int cutoffPage)
+SlruMayDeleteSegment(SlruCtl ctl, int64 segpage, int64 cutoffPage)
 {
-	int			seg_last_page = segpage + SLRU_PAGES_PER_SEGMENT - 1;
+	int64		seg_last_page = segpage + SLRU_PAGES_PER_SEGMENT - 1;
 
 	Assert(segpage % SLRU_PAGES_PER_SEGMENT == 0);
 
@@ -1405,7 +1432,7 @@ SlruPagePrecedesTestOffset(SlruCtl ctl, int per_page, uint32 offset)
 {
 	TransactionId lhs,
 				rhs;
-	int			newestPage,
+	int64		newestPage,
 				oldestPage;
 	TransactionId newestXact,
 				oldestXact;
@@ -1498,9 +1525,10 @@ SlruPagePrecedesUnitTests(SlruCtl ctl, int per_page)
  *		one containing the page passed as "data".
  */
 bool
-SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int64 segpage,
+							void *data)
 {
-	int			cutoffPage = *(int *) data;
+	int64		cutoffPage = *(int64 *) data;
 
 	if (SlruMayDeleteSegment(ctl, segpage, cutoffPage))
 		return true;			/* found one; don't iterate any more */
@@ -1513,9 +1541,10 @@ SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int segpage, void *data
  *		This callback deletes segments prior to the one passed in as "data".
  */
 static bool
-SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int64 segpage,
+						  void *data)
 {
-	int			cutoffPage = *(int *) data;
+	int64		cutoffPage = *(int64 *) data;
 
 	if (SlruMayDeleteSegment(ctl, segpage, cutoffPage))
 		SlruInternalDeleteSegment(ctl, segpage / SLRU_PAGES_PER_SEGMENT);
@@ -1528,13 +1557,37 @@ SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int segpage, void *data)
  *		This callback deletes all segments.
  */
 bool
-SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
 	SlruInternalDeleteSegment(ctl, segpage / SLRU_PAGES_PER_SEGMENT);
 
 	return false;				/* keep going */
 }
 
+/*
+ * An internal function used by SlruScanDirectory().
+ *
+ * Returns true if a file with a name of a given length may be a correct
+ * SLRU segment.
+ */
+static inline bool
+SlruCorrectSegmentFilenameLength(SlruCtl ctl, size_t len)
+{
+	if (ctl->long_segment_names)
+		return (len == 15);		/* see SlruFileName() */
+	else
+
+		/*
+		 * Commit 638cf09e76d allowed 5-character lengths. Later commit
+		 * 73c986adde5 allowed 6-character length.
+		 *
+		 * Note: There is an ongoing plan to migrate all SLRUs to 64-bit page
+		 * numbers, and the corresponding 15-character file names, which may
+		 * eventually deprecate the support for 4, 5, and 6-character names.
+		 */
+		return (len == 4 || len == 5 || len == 6);
+}
+
 /*
  * Scan the SimpleLru directory and apply a callback to each file found in it.
  *
@@ -1556,8 +1609,8 @@ SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data)
 	bool		retval = false;
 	DIR		   *cldir;
 	struct dirent *clde;
-	int			segno;
+	int64		segno;
-	int			segpage;
+	int64		segpage;
 
 	cldir = AllocateDir(ctl->Dir);
 	while ((clde = ReadDir(cldir, ctl->Dir)) != NULL)
@@ -1566,10 +1619,10 @@ SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data)
 
 		len = strlen(clde->d_name);
 
-		if ((len == 4 || len == 5 || len == 6) &&
+		if (SlruCorrectSegmentFilenameLength(ctl, len) &&
 			strspn(clde->d_name, "0123456789ABCDEF") == len)
 		{
-			segno = (int) strtol(clde->d_name, NULL, 16);
+			segno = strtoi64(clde->d_name, NULL, 16);
 			segpage = segno * SLRU_PAGES_PER_SEGMENT;
 
 			elog(DEBUG2, "SlruScanDirectory invoking callback on %s/%s",

src/backend/access/transam/subtrans.c

@@ -51,7 +51,16 @@
 /* We need four bytes per xact */
 #define SUBTRANS_XACTS_PER_PAGE (BLCKSZ / sizeof(TransactionId))
 
-#define TransactionIdToPage(xid) ((xid) / (TransactionId) SUBTRANS_XACTS_PER_PAGE)
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/SUBTRANS_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToPage(TransactionId xid)
+{
+	return xid / (int64) SUBTRANS_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToEntry(xid) ((xid) % (TransactionId) SUBTRANS_XACTS_PER_PAGE)
 
 
@@ -63,8 +72,8 @@ static SlruCtlData SubTransCtlData;
 #define SubTransCtl  (&SubTransCtlData)
 
 
-static int	ZeroSUBTRANSPage(int pageno);
+static int	ZeroSUBTRANSPage(int64 pageno);
-static bool SubTransPagePrecedes(int page1, int page2);
+static bool SubTransPagePrecedes(int64 page1, int64 page2);
 
 
 /*
@@ -73,7 +82,7 @@ static bool SubTransPagePrecedes(int page1, int page2);
 void
 SubTransSetParent(TransactionId xid, TransactionId parent)
 {
-	int			pageno = TransactionIdToPage(xid);
+	int64		pageno = TransactionIdToPage(xid);
 	int			entryno = TransactionIdToEntry(xid);
 	int			slotno;
 	TransactionId *ptr;
@@ -108,7 +117,7 @@ SubTransSetParent(TransactionId xid, TransactionId parent)
 TransactionId
 SubTransGetParent(TransactionId xid)
 {
-	int			pageno = TransactionIdToPage(xid);
+	int64		pageno = TransactionIdToPage(xid);
 	int			entryno = TransactionIdToEntry(xid);
 	int			slotno;
 	TransactionId *ptr;
@@ -193,7 +202,8 @@ SUBTRANSShmemInit(void)
 	SubTransCtl->PagePrecedes = SubTransPagePrecedes;
 	SimpleLruInit(SubTransCtl, "Subtrans", NUM_SUBTRANS_BUFFERS, 0,
 				  SubtransSLRULock, "pg_subtrans",
-				  LWTRANCHE_SUBTRANS_BUFFER, SYNC_HANDLER_NONE);
+				  LWTRANCHE_SUBTRANS_BUFFER, SYNC_HANDLER_NONE,
+				  false);
 	SlruPagePrecedesUnitTests(SubTransCtl, SUBTRANS_XACTS_PER_PAGE);
 }
 
@@ -233,7 +243,7 @@ BootStrapSUBTRANS(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroSUBTRANSPage(int pageno)
+ZeroSUBTRANSPage(int64 pageno)
 {
 	return SimpleLruZeroPage(SubTransCtl, pageno);
 }
@@ -249,8 +259,8 @@ void
 StartupSUBTRANS(TransactionId oldestActiveXID)
 {
 	FullTransactionId nextXid;
-	int			startPage;
+	int64		startPage;
-	int			endPage;
+	int64		endPage;
 
 	/*
 	 * Since we don't expect pg_subtrans to be valid across crashes, we
@@ -307,7 +317,7 @@ CheckPointSUBTRANS(void)
 void
 ExtendSUBTRANS(TransactionId newestXact)
 {
-	int			pageno;
+	int64		pageno;
 
 	/*
 	 * No work except at first XID of a page.  But beware: just after
@@ -337,7 +347,7 @@ ExtendSUBTRANS(TransactionId newestXact)
 void
 TruncateSUBTRANS(TransactionId oldestXact)
 {
-	int			cutoffPage;
+	int64		cutoffPage;
 
 	/*
 	 * The cutoff point is the start of the segment containing oldestXact. We
@@ -359,7 +369,7 @@ TruncateSUBTRANS(TransactionId oldestXact)
  * Analogous to CLOGPagePrecedes().
  */
 static bool
-SubTransPagePrecedes(int page1, int page2)
+SubTransPagePrecedes(int64 page1, int64 page2)
 {
 	TransactionId xid1;
 	TransactionId xid2;

src/backend/access/transam/twophase.c

@@ -942,8 +942,46 @@ TwoPhaseGetDummyProc(TransactionId xid, bool lock_held)
 /* State file support													*/
 /************************************************************************/
 
-#define TwoPhaseFilePath(path, xid) \
+/*
-	snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)
+ * Compute the FullTransactionId for the given TransactionId.
+ *
+ * The wrap logic is safe here because the span of active xids cannot exceed one
+ * epoch at any given time.
+ */
+static inline FullTransactionId
+AdjustToFullTransactionId(TransactionId xid)
+{
+	FullTransactionId nextFullXid;
+	TransactionId nextXid;
+	uint32		epoch;
+
+	Assert(TransactionIdIsValid(xid));
+
+	LWLockAcquire(XidGenLock, LW_SHARED);
+	nextFullXid = ShmemVariableCache->nextXid;
+	LWLockRelease(XidGenLock);
+
+	nextXid = XidFromFullTransactionId(nextFullXid);
+	epoch = EpochFromFullTransactionId(nextFullXid);
+	if (unlikely(xid > nextXid))
+	{
+		/* Wraparound occured, must be from a prev epoch. */
+		Assert(epoch > 0);
+		epoch--;
+	}
+
+	return FullTransactionIdFromEpochAndXid(epoch, xid);
+}
+
+static inline int
+TwoPhaseFilePath(char *path, TransactionId xid)
+{
+	FullTransactionId fxid = AdjustToFullTransactionId(xid);
+
+	return snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X%08X",
+					EpochFromFullTransactionId(fxid),
+					XidFromFullTransactionId(fxid));
+}
 
 /*
  * 2PC state file format:
@@ -1882,13 +1920,15 @@ restoreTwoPhaseData(void)
 	cldir = AllocateDir(TWOPHASE_DIR);
 	while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
 	{
-		if (strlen(clde->d_name) == 8 &&
+		if (strlen(clde->d_name) == 16 &&
-			strspn(clde->d_name, "0123456789ABCDEF") == 8)
+			strspn(clde->d_name, "0123456789ABCDEF") == 16)
 		{
 			TransactionId xid;
+			FullTransactionId fxid;
 			char	   *buf;
 
-			xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
+			fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16));
+			xid = XidFromFullTransactionId(fxid);
 
 			buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr,
 										true, false, false);

src/backend/commands/async.c

@@ -103,12 +103,11 @@
  *	  until we reach either a notification from an uncommitted transaction or
  *	  the head pointer's position.
  *
- * 6. To avoid SLRU wraparound and limit disk space consumption, the tail
+ * 6. To limit disk space consumption, the tail pointer needs to be advanced
- *	  pointer needs to be advanced so that old pages can be truncated.
+ *	  so that old pages can be truncated. This is relatively expensive
- *	  This is relatively expensive (notably, it requires an exclusive lock),
+ *	  (notably, it requires an exclusive lock), so we don't want to do it
- *	  so we don't want to do it often.  We make sending backends do this work
+ *	  often. We make sending backends do this work if they advanced the queue
- *	  if they advanced the queue head into a new page, but only once every
+ *	  head into a new page, but only once every QUEUE_CLEANUP_DELAY pages.
- *	  QUEUE_CLEANUP_DELAY pages.
  *
  * An application that listens on the same channel it notifies will get
  * NOTIFY messages for its own NOTIFYs.  These can be ignored, if not useful,
@@ -120,7 +119,7 @@
  * The amount of shared memory used for notify management (NUM_NOTIFY_BUFFERS)
  * can be varied without affecting anything but performance.  The maximum
  * amount of notification data that can be queued at one time is determined
- * by slru.c's wraparound limit; see QUEUE_MAX_PAGE below.
+ * by max_notify_queue_pages GUC.
  *-------------------------------------------------------------------------
  */
 
@@ -196,7 +195,7 @@ typedef struct AsyncQueueEntry
  */
 typedef struct QueuePosition
 {
-	int			page;			/* SLRU page number */
+	int64		page;			/* SLRU page number */
 	int			offset;			/* byte offset within page */
 } QueuePosition;
 
@@ -312,23 +311,8 @@ static SlruCtlData NotifyCtlData;
 
 #define NotifyCtl					(&NotifyCtlData)
 #define QUEUE_PAGESIZE				BLCKSZ
-#define QUEUE_FULL_WARN_INTERVAL	5000	/* warn at most once every 5s */
 
-/*
+#define QUEUE_FULL_WARN_INTERVAL	5000	/* warn at most once every 5s */
- * Use segments 0000 through FFFF.  Each contains SLRU_PAGES_PER_SEGMENT pages
- * which gives us the pages from 0 to SLRU_PAGES_PER_SEGMENT * 0x10000 - 1.
- * We could use as many segments as SlruScanDirectory() allows, but this gives
- * us so much space already that it doesn't seem worth the trouble.
- *
- * The most data we can have in the queue at a time is QUEUE_MAX_PAGE/2
- * pages, because more than that would confuse slru.c into thinking there
- * was a wraparound condition.  With the default BLCKSZ this means there
- * can be up to 8GB of queued-and-not-read data.
- *
- * Note: it's possible to redefine QUEUE_MAX_PAGE with a smaller multiple of
- * SLRU_PAGES_PER_SEGMENT, for easier testing of queue-full behaviour.
- */
-#define QUEUE_MAX_PAGE			(SLRU_PAGES_PER_SEGMENT * 0x10000 - 1)
 
 /*
  * listenChannels identifies the channels we are actually listening to
@@ -439,12 +423,15 @@ static bool amRegisteredListener = false;
 /* have we advanced to a page that's a multiple of QUEUE_CLEANUP_DELAY? */
 static bool tryAdvanceTail = false;
 
-/* GUC parameter */
+/* GUC parameters */
 bool		Trace_notify = false;
 
+/* For 8 KB pages this gives 8 GB of disk space */
+int			max_notify_queue_pages = 1048576;
+
 /* local function prototypes */
-static int	asyncQueuePageDiff(int p, int q);
+static inline int64 asyncQueuePageDiff(int64 p, int64 q);
-static bool asyncQueuePagePrecedes(int p, int q);
+static inline bool asyncQueuePagePrecedes(int64 p, int64 q);
 static void queue_listen(ListenActionKind action, const char *channel);
 static void Async_UnlistenOnExit(int code, Datum arg);
 static void Exec_ListenPreCommit(void);
@@ -474,39 +461,23 @@ static int	notification_match(const void *key1, const void *key2, Size keysize);
 static void ClearPendingActionsAndNotifies(void);
 
 /*
- * Compute the difference between two queue page numbers (i.e., p - q),
+ * Compute the difference between two queue page numbers.
- * accounting for wraparound.
+ * Previously this function accounted for a wraparound.
  */
-static int
+static inline int64
-asyncQueuePageDiff(int p, int q)
+asyncQueuePageDiff(int64 p, int64 q)
 {
-	int			diff;
+	return p - q;
-
-	/*
-	 * We have to compare modulo (QUEUE_MAX_PAGE+1)/2.  Both inputs should be
-	 * in the range 0..QUEUE_MAX_PAGE.
-	 */
-	Assert(p >= 0 && p <= QUEUE_MAX_PAGE);
-	Assert(q >= 0 && q <= QUEUE_MAX_PAGE);
-
-	diff = p - q;
-	if (diff >= ((QUEUE_MAX_PAGE + 1) / 2))
-		diff -= QUEUE_MAX_PAGE + 1;
-	else if (diff < -((QUEUE_MAX_PAGE + 1) / 2))
-		diff += QUEUE_MAX_PAGE + 1;
-	return diff;
 }
 
 /*
- * Is p < q, accounting for wraparound?
+ * Determines whether p precedes q.
- *
+ * Previously this function accounted for a wraparound.
- * Since asyncQueueIsFull() blocks creation of a page that could precede any
- * extant page, we need not assess entries within a page.
  */
-static bool
+static inline bool
-asyncQueuePagePrecedes(int p, int q)
+asyncQueuePagePrecedes(int64 p, int64 q)
 {
-	return asyncQueuePageDiff(p, q) < 0;
+	return p < q;
 }
 
 /*
@@ -566,12 +537,13 @@ AsyncShmemInit(void)
 	}
 
 	/*
-	 * Set up SLRU management of the pg_notify data.
+	 * Set up SLRU management of the pg_notify data. Note that long segment
+	 * names are used in order to avoid wraparound.
 	 */
 	NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
 	SimpleLruInit(NotifyCtl, "Notify", NUM_NOTIFY_BUFFERS, 0,
 				  NotifySLRULock, "pg_notify", LWTRANCHE_NOTIFY_BUFFER,
-				  SYNC_HANDLER_NONE);
+				  SYNC_HANDLER_NONE, true);
 
 	if (!found)
 	{
@@ -1305,27 +1277,11 @@ asyncQueueUnregister(void)
 static bool
 asyncQueueIsFull(void)
 {
-	int			nexthead;
+	int			headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
-	int			boundary;
+	int			tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
+	int			occupied = headPage - tailPage;
 
-	/*
+	return occupied >= max_notify_queue_pages;
-	 * The queue is full if creating a new head page would create a page that
-	 * logically precedes the current global tail pointer, ie, the head
-	 * pointer would wrap around compared to the tail.  We cannot create such
-	 * a head page for fear of confusing slru.c.  For safety we round the tail
-	 * pointer back to a segment boundary (truncation logic in
-	 * asyncQueueAdvanceTail does not do this, so doing it here is optional).
-	 *
-	 * Note that this test is *not* dependent on how much space there is on
-	 * the current head page.  This is necessary because asyncQueueAddEntries
-	 * might try to create the next head page in any case.
-	 */
-	nexthead = QUEUE_POS_PAGE(QUEUE_HEAD) + 1;
-	if (nexthead > QUEUE_MAX_PAGE)
-		nexthead = 0;			/* wrap around */
-	boundary = QUEUE_STOP_PAGE;
-	boundary -= boundary % SLRU_PAGES_PER_SEGMENT;
-	return asyncQueuePagePrecedes(nexthead, boundary);
 }
 
 /*
@@ -1336,7 +1292,7 @@ asyncQueueIsFull(void)
 static bool
 asyncQueueAdvance(volatile QueuePosition *position, int entryLength)
 {
-	int			pageno = QUEUE_POS_PAGE(*position);
+	int64		pageno = QUEUE_POS_PAGE(*position);
 	int			offset = QUEUE_POS_OFFSET(*position);
 	bool		pageJump = false;
 
@@ -1355,8 +1311,6 @@ asyncQueueAdvance(volatile QueuePosition *position, int entryLength)
 	if (offset + QUEUEALIGN(AsyncQueueEntryEmptySize) > QUEUE_PAGESIZE)
 	{
 		pageno++;
-		if (pageno > QUEUE_MAX_PAGE)
-			pageno = 0;			/* wrap around */
 		offset = 0;
 		pageJump = true;
 	}
@@ -1409,7 +1363,7 @@ asyncQueueAddEntries(ListCell *nextNotify)
 {
 	AsyncQueueEntry qe;
 	QueuePosition queue_head;
-	int			pageno;
+	int64		pageno;
 	int			offset;
 	int			slotno;
 
@@ -1433,9 +1387,6 @@ asyncQueueAddEntries(ListCell *nextNotify)
 	 * If this is the first write since the postmaster started, we need to
 	 * initialize the first page of the async SLRU.  Otherwise, the current
 	 * page should be initialized already, so just fetch it.
-	 *
-	 * (We could also take the first path when the SLRU position has just
-	 * wrapped around, but re-zeroing the page is harmless in that case.)
 	 */
 	pageno = QUEUE_POS_PAGE(queue_head);
 	if (QUEUE_POS_IS_ZERO(queue_head))
@@ -1548,20 +1499,12 @@ asyncQueueUsage(void)
 {
 	int			headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
 	int			tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
-	int			occupied;
+	int			occupied = headPage - tailPage;
-
-	occupied = headPage - tailPage;
 
 	if (occupied == 0)
 		return (double) 0;		/* fast exit for common case */
 
-	if (occupied < 0)
+	return (double) occupied / (double) max_notify_queue_pages;
-	{
-		/* head has wrapped around, tail not yet */
-		occupied += QUEUE_MAX_PAGE + 1;
-	}
-
-	return (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2);
 }
 
 /*
@@ -2209,11 +2152,6 @@ asyncQueueAdvanceTail(void)
 		 */
 		SimpleLruTruncate(NotifyCtl, newtailpage);
 
-		/*
-		 * Update QUEUE_STOP_PAGE.  This changes asyncQueueIsFull()'s verdict
-		 * for the segment immediately prior to the old tail, allowing fresh
-		 * data into that segment.
-		 */
 		LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
 		QUEUE_STOP_PAGE = newtailpage;
 		LWLockRelease(NotifyQueueLock);

src/backend/storage/lmgr/predicate.c

@@ -437,7 +437,7 @@ static void SetPossibleUnsafeConflict(SERIALIZABLEXACT *roXact, SERIALIZABLEXACT
 static void ReleaseRWConflict(RWConflict conflict);
 static void FlagSxactUnsafe(SERIALIZABLEXACT *sxact);
 
-static bool SerialPagePrecedesLogically(int page1, int page2);
+static bool SerialPagePrecedesLogically(int64 page1, int64 page2);
 static void SerialInit(void);
 static void SerialAdd(TransactionId xid, SerCommitSeqNo minConflictCommitSeqNo);
 static SerCommitSeqNo SerialGetMinConflictCommitSeqNo(TransactionId xid);
@@ -724,7 +724,7 @@ FlagSxactUnsafe(SERIALIZABLEXACT *sxact)
  * Analogous to CLOGPagePrecedes().
  */
 static bool
-SerialPagePrecedesLogically(int page1, int page2)
+SerialPagePrecedesLogically(int64 page1, int64 page2)
 {
 	TransactionId xid1;
 	TransactionId xid2;
@@ -744,7 +744,7 @@ SerialPagePrecedesLogicallyUnitTests(void)
 {
 	int			per_page = SERIAL_ENTRIESPERPAGE,
 				offset = per_page / 2;
-	int			newestPage,
+	int64		newestPage,
 				oldestPage,
 				headPage,
 				targetPage;
@@ -809,7 +809,8 @@ SerialInit(void)
 	SerialSlruCtl->PagePrecedes = SerialPagePrecedesLogically;
 	SimpleLruInit(SerialSlruCtl, "Serial",
 				  NUM_SERIAL_BUFFERS, 0, SerialSLRULock, "pg_serial",
-				  LWTRANCHE_SERIAL_BUFFER, SYNC_HANDLER_NONE);
+				  LWTRANCHE_SERIAL_BUFFER, SYNC_HANDLER_NONE,
+				  false);
 #ifdef USE_ASSERT_CHECKING
 	SerialPagePrecedesLogicallyUnitTests();
 #endif
@@ -842,9 +843,9 @@ static void
 SerialAdd(TransactionId xid, SerCommitSeqNo minConflictCommitSeqNo)
 {
 	TransactionId tailXid;
-	int			targetPage;
+	int64		targetPage;
 	int			slotno;
-	int			firstZeroPage;
+	int64		firstZeroPage;
 	bool		isNewPage;
 
 	Assert(TransactionIdIsValid(xid));

src/backend/utils/misc/guc_tables.c

@@ -2687,6 +2687,16 @@ struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_notify_queue_pages", PGC_POSTMASTER, RESOURCES_DISK,
+			gettext_noop("Sets the maximum number of allocated pages for NOTIFY / LISTEN queue."),
+			NULL,
+		},
+		&max_notify_queue_pages,
+		1048576, 64, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"wal_decode_buffer_size", PGC_POSTMASTER, WAL_RECOVERY,
 			gettext_noop("Buffer size for reading ahead in the WAL during recovery."),

src/backend/utils/misc/postgresql.conf.sample

@@ -166,6 +166,9 @@
 #temp_file_limit = -1			# limits per-process temp file space
 					# in kilobytes, or -1 for no limit
 
+#max_notify_queue_pages = 1048576	# limits the number of SLRU pages allocated
+									# for NOTIFY / LISTEN queue
+
 # - Kernel Resources -
 
 #max_files_per_process = 1000		# min 64

src/include/access/clog.h

@@ -31,7 +31,7 @@ typedef int XidStatus;
 
 typedef struct xl_clog_truncate
 {
-	int			pageno;
+	int64		pageno;
 	TransactionId oldestXact;
 	Oid			oldestXactDb;
 } xl_clog_truncate;

src/include/access/commit_ts.h

@@ -60,7 +60,7 @@ typedef struct xl_commit_ts_set
 
 typedef struct xl_commit_ts_truncate
 {
-	int			pageno;
+	int64		pageno;
 	TransactionId oldestXid;
 } xl_commit_ts_truncate;
 

src/include/access/slru.h

@@ -64,7 +64,7 @@ typedef struct SlruSharedData
 	char	  **page_buffer;
 	SlruPageStatus *page_status;
 	bool	   *page_dirty;
-	int		   *page_number;
+	int64	   *page_number;
 	int		   *page_lru_count;
 	LWLockPadded *buffer_locks;
 
@@ -95,7 +95,7 @@ typedef struct SlruSharedData
 	 * this is not critical data, since we use it only to avoid swapping out
 	 * the latest page.
 	 */
-	int			latest_page_number;
+	int64		latest_page_number;
 
 	/* SLRU's index for statistics purposes (might not be unique) */
 	int			slru_stats_idx;
@@ -127,7 +127,15 @@ typedef struct SlruCtlData
 	 * the behavior of this callback has no functional implications.)  Use
 	 * SlruPagePrecedesUnitTests() in SLRUs meeting its criteria.
 	 */
-	bool		(*PagePrecedes) (int, int);
+	bool		(*PagePrecedes) (int64, int64);
+
+	/*
+	 * If true, use long segment filenames formed from lower 48 bits of the
+	 * segment number, e.g. pg_xact/000000001234. Otherwise, use short
+	 * filenames formed from lower 16 bits of the segment number e.g.
+	 * pg_xact/1234.
+	 */
+	bool		long_segment_names;
 
 	/*
 	 * Dir is set during SimpleLruInit and does not change thereafter. Since
@@ -142,11 +150,12 @@ typedef SlruCtlData *SlruCtl;
 extern Size SimpleLruShmemSize(int nslots, int nlsns);
 extern void SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
 						  LWLock *ctllock, const char *subdir, int tranche_id,
-						  SyncRequestHandler sync_handler);
+						  SyncRequestHandler sync_handler,
-extern int	SimpleLruZeroPage(SlruCtl ctl, int pageno);
+						  bool long_segment_names);
-extern int	SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
+extern int	SimpleLruZeroPage(SlruCtl ctl, int64 pageno);
+extern int	SimpleLruReadPage(SlruCtl ctl, int64 pageno, bool write_ok,
 							  TransactionId xid);
-extern int	SimpleLruReadPage_ReadOnly(SlruCtl ctl, int pageno,
+extern int	SimpleLruReadPage_ReadOnly(SlruCtl ctl, int64 pageno,
 									   TransactionId xid);
 extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
 extern void SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied);
@@ -155,20 +164,20 @@ extern void SlruPagePrecedesUnitTests(SlruCtl ctl, int per_page);
 #else
 #define SlruPagePrecedesUnitTests(ctl, per_page) do {} while (0)
 #endif
-extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
+extern void SimpleLruTruncate(SlruCtl ctl, int64 cutoffPage);
-extern bool SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno);
+extern bool SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int64 pageno);
 
-typedef bool (*SlruScanCallback) (SlruCtl ctl, char *filename, int segpage,
+typedef bool (*SlruScanCallback) (SlruCtl ctl, char *filename, int64 segpage,
 								  void *data);
 extern bool SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data);
-extern void SlruDeleteSegment(SlruCtl ctl, int segno);
+extern void SlruDeleteSegment(SlruCtl ctl, int64 segno);
 
 extern int	SlruSyncFileTag(SlruCtl ctl, const FileTag *ftag, char *path);
 
 /* SlruScanDirectory public callbacks */
 extern bool SlruScanDirCbReportPresence(SlruCtl ctl, char *filename,
-										int segpage, void *data);
+										int64 segpage, void *data);
-extern bool SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage,
+extern bool SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int64 segpage,
 								   void *data);
 
 #endif							/* SLRU_H */

src/include/commands/async.h

@@ -21,6 +21,7 @@
 #define NUM_NOTIFY_BUFFERS	8
 
 extern PGDLLIMPORT bool Trace_notify;
+extern PGDLLIMPORT int max_notify_queue_pages;
 extern PGDLLIMPORT volatile sig_atomic_t notifyInterruptPending;
 
 extern Size AsyncShmemSize(void);

src/include/storage/proc.h

@@ -281,7 +281,7 @@ struct PGPROC
 	TransactionId clogGroupMemberXid;	/* transaction id of clog group member */
 	XidStatus	clogGroupMemberXidStatus;	/* transaction status of clog
 											 * group member */
-	int			clogGroupMemberPage;	/* clog page corresponding to
+	int64		clogGroupMemberPage;	/* clog page corresponding to
 										 * transaction id of clog group member */
 	XLogRecPtr	clogGroupMemberLsn; /* WAL location of commit record for clog
 									 * group member */

src/include/storage/sync.h

@@ -52,7 +52,7 @@ typedef struct FileTag
 	int16		handler;		/* SyncRequestHandler value, saving space */
 	int16		forknum;		/* ForkNumber, saving space */
 	RelFileLocator rlocator;
-	uint32		segno;
+	uint64		segno;
 } FileTag;
 
 extern void InitSync(void);

src/test/modules/test_slru/expected/test_slru.out

@@ -61,7 +61,7 @@ SELECT test_slru_page_writeall();
 
 -- Flush the last page written out.
 SELECT test_slru_page_sync(12393);
-NOTICE:  Called SlruSyncFileTag() for segment 387 on path pg_test_slru/0183
+NOTICE:  Called SlruSyncFileTag() for segment 387 on path pg_test_slru/000000000000183
  test_slru_page_sync 
 ---------------------
  
@@ -132,4 +132,139 @@ SELECT test_slru_page_exists(12393);
  f
 (1 row)
 
+--
+-- Test 64-bit pages
+--
+SELECT test_slru_page_exists(0x1234500000000);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+SELECT test_slru_page_write(0x1234500000000, 'Test SLRU 64-bit');
+ test_slru_page_write 
+----------------------
+ 
+(1 row)
+
+SELECT test_slru_page_read(0x1234500000000);
+ test_slru_page_read 
+---------------------
+ Test SLRU 64-bit
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000000);
+ test_slru_page_exists 
+-----------------------
+ t
+(1 row)
+
+-- 48 extra pages
+SELECT count(test_slru_page_write(a, 'Test SLRU 64-bit'))
+  FROM generate_series(0x1234500000001, 0x1234500000030, 1) as a;
+ count 
+-------
+    48
+(1 row)
+
+-- Reading page in buffer for read and write
+SELECT test_slru_page_read(0x1234500000020, true);
+ test_slru_page_read 
+---------------------
+ Test SLRU 64-bit
+(1 row)
+
+-- Reading page in buffer for read-only
+SELECT test_slru_page_readonly(0x1234500000020);
+ test_slru_page_readonly 
+-------------------------
+ Test SLRU 64-bit
+(1 row)
+
+-- Reading page not in buffer with read-only
+SELECT test_slru_page_readonly(0x1234500000001);
+ test_slru_page_readonly 
+-------------------------
+ Test SLRU 64-bit
+(1 row)
+
+-- Write all the pages in buffers
+SELECT test_slru_page_writeall();
+ test_slru_page_writeall 
+-------------------------
+ 
+(1 row)
+
+-- Flush the last page written out.
+SELECT test_slru_page_sync(0x1234500000030);
+NOTICE:  Called SlruSyncFileTag() for segment 10007944888321 on path pg_test_slru/000091A28000001
+ test_slru_page_sync 
+---------------------
+ 
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000030);
+ test_slru_page_exists 
+-----------------------
+ t
+(1 row)
+
+-- Segment deletion
+SELECT test_slru_page_delete(0x1234500000030);
+NOTICE:  Called SlruDeleteSegment() for segment 10007944888321
+ test_slru_page_delete 
+-----------------------
+ 
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000030);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+-- Page truncation
+SELECT test_slru_page_exists(0x1234500000020);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+SELECT test_slru_page_truncate(0x1234500000020);
+ test_slru_page_truncate 
+-------------------------
+ 
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000020);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+-- Full deletion
+SELECT test_slru_delete_all();
+ test_slru_delete_all 
+----------------------
+ 
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000000);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000020);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
+SELECT test_slru_page_exists(0x1234500000030);
+ test_slru_page_exists 
+-----------------------
+ f
+(1 row)
+
 DROP EXTENSION test_slru;

src/test/modules/test_slru/sql/test_slru.sql

@@ -35,4 +35,42 @@ SELECT test_slru_page_exists(12345);
 SELECT test_slru_page_exists(12377);
 SELECT test_slru_page_exists(12393);
 
+--
+-- Test 64-bit pages
+--
+SELECT test_slru_page_exists(0x1234500000000);
+SELECT test_slru_page_write(0x1234500000000, 'Test SLRU 64-bit');
+SELECT test_slru_page_read(0x1234500000000);
+SELECT test_slru_page_exists(0x1234500000000);
+
+-- 48 extra pages
+SELECT count(test_slru_page_write(a, 'Test SLRU 64-bit'))
+  FROM generate_series(0x1234500000001, 0x1234500000030, 1) as a;
+
+-- Reading page in buffer for read and write
+SELECT test_slru_page_read(0x1234500000020, true);
+-- Reading page in buffer for read-only
+SELECT test_slru_page_readonly(0x1234500000020);
+-- Reading page not in buffer with read-only
+SELECT test_slru_page_readonly(0x1234500000001);
+
+-- Write all the pages in buffers
+SELECT test_slru_page_writeall();
+-- Flush the last page written out.
+SELECT test_slru_page_sync(0x1234500000030);
+SELECT test_slru_page_exists(0x1234500000030);
+-- Segment deletion
+SELECT test_slru_page_delete(0x1234500000030);
+SELECT test_slru_page_exists(0x1234500000030);
+-- Page truncation
+SELECT test_slru_page_exists(0x1234500000020);
+SELECT test_slru_page_truncate(0x1234500000020);
+SELECT test_slru_page_exists(0x1234500000020);
+
+-- Full deletion
+SELECT test_slru_delete_all();
+SELECT test_slru_page_exists(0x1234500000000);
+SELECT test_slru_page_exists(0x1234500000020);
+SELECT test_slru_page_exists(0x1234500000030);
+
 DROP EXTENSION test_slru;

src/test/modules/test_slru/test_slru--1.0.sql

@@ -1,21 +1,21 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION test_slru" to load this file. \quit
 
-CREATE OR REPLACE FUNCTION test_slru_page_write(int, text) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_write(bigint, text) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_write' LANGUAGE C;
 CREATE OR REPLACE FUNCTION test_slru_page_writeall() RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_writeall' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_sync(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_sync(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_sync' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_read(int, bool DEFAULT true) RETURNS text
+CREATE OR REPLACE FUNCTION test_slru_page_read(bigint, bool DEFAULT true) RETURNS text
   AS 'MODULE_PATHNAME', 'test_slru_page_read' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_readonly(int) RETURNS text
+CREATE OR REPLACE FUNCTION test_slru_page_readonly(bigint) RETURNS text
   AS 'MODULE_PATHNAME', 'test_slru_page_readonly' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_exists(int) RETURNS bool
+CREATE OR REPLACE FUNCTION test_slru_page_exists(bigint) RETURNS bool
   AS 'MODULE_PATHNAME', 'test_slru_page_exists' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_delete(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_delete(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_delete' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_truncate(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_truncate(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_truncate' LANGUAGE C;
 CREATE OR REPLACE FUNCTION test_slru_delete_all() RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_delete_all' LANGUAGE C;

src/test/modules/test_slru/test_slru.c

@@ -51,7 +51,7 @@ static shmem_request_hook_type prev_shmem_request_hook = NULL;
 static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
 
 static bool
-test_slru_scan_cb(SlruCtl ctl, char *filename, int segpage, void *data)
+test_slru_scan_cb(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
 	elog(NOTICE, "Calling test_slru_scan_cb()");
 	return SlruScanDirCbDeleteAll(ctl, filename, segpage, data);
@@ -60,7 +60,7 @@ test_slru_scan_cb(SlruCtl ctl, char *filename, int segpage, void *data)
 Datum
 test_slru_page_write(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	char	   *data = text_to_cstring(PG_GETARG_TEXT_PP(1));
 	int			slotno;
 
@@ -95,7 +95,7 @@ test_slru_page_writeall(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_read(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	bool		write_ok = PG_GETARG_BOOL(1);
 	char	   *data = NULL;
 	int			slotno;
@@ -113,7 +113,7 @@ test_slru_page_read(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_readonly(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	char	   *data = NULL;
 	int			slotno;
 
@@ -131,7 +131,7 @@ test_slru_page_readonly(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_exists(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	bool		found;
 
 	LWLockAcquire(TestSLRULock, LW_EXCLUSIVE);
@@ -144,7 +144,7 @@ test_slru_page_exists(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_sync(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	FileTag		ftag;
 	char		path[MAXPGPATH];
 
@@ -152,8 +152,8 @@ test_slru_page_sync(PG_FUNCTION_ARGS)
 	ftag.segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	SlruSyncFileTag(TestSlruCtl, &ftag, path);
 
-	elog(NOTICE, "Called SlruSyncFileTag() for segment %u on path %s",
+	elog(NOTICE, "Called SlruSyncFileTag() for segment %lld on path %s",
-		 ftag.segno, path);
+		 (long long) ftag.segno, path);
 
 	PG_RETURN_VOID();
 }
@@ -161,13 +161,14 @@ test_slru_page_sync(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_delete(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 	FileTag		ftag;
 
 	ftag.segno = pageno / SLRU_PAGES_PER_SEGMENT;
 	SlruDeleteSegment(TestSlruCtl, ftag.segno);
 
-	elog(NOTICE, "Called SlruDeleteSegment() for segment %u", ftag.segno);
+	elog(NOTICE, "Called SlruDeleteSegment() for segment %lld",
+		 (long long) ftag.segno);
 
 	PG_RETURN_VOID();
 }
@@ -175,7 +176,7 @@ test_slru_page_delete(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_truncate(PG_FUNCTION_ARGS)
 {
-	int			pageno = PG_GETARG_INT32(0);
+	int64		pageno = PG_GETARG_INT64(0);
 
 	SimpleLruTruncate(TestSlruCtl, pageno);
 	PG_RETURN_VOID();
@@ -205,7 +206,7 @@ test_slru_shmem_request(void)
 }
 
 static bool
-test_slru_page_precedes_logically(int page1, int page2)
+test_slru_page_precedes_logically(int64 page1, int64 page2)
 {
 	return page1 < page2;
 }
@@ -213,6 +214,11 @@ test_slru_page_precedes_logically(int page1, int page2)
 static void
 test_slru_shmem_startup(void)
 {
+	/*
+	 * Short segments names are well tested elsewhere so in this test we are
+	 * focusing on long names.
+	 */
+	const bool	long_segment_names = true;
 	const char	slru_dir_name[] = "pg_test_slru";
 	int			test_tranche_id;
 
@@ -233,7 +239,7 @@ test_slru_shmem_startup(void)
 	TestSlruCtl->PagePrecedes = test_slru_page_precedes_logically;
 	SimpleLruInit(TestSlruCtl, "TestSLRU",
 				  NUM_TEST_BUFFERS, 0, TestSLRULock, slru_dir_name,
-				  test_tranche_id, SYNC_HANDLER_NONE);
+				  test_tranche_id, SYNC_HANDLER_NONE, long_segment_names);
 }
 
 void