public inbox for [email protected]
help / color / mirror / Atom feedFrom: Li, Yong <[email protected]>
To: Aleksander Alekseev <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Bagga, Rishu <[email protected]>
Cc: Robert Haas <[email protected]>
Cc: Debnath, Shawn <[email protected]>
Cc: Andrey Borodin <[email protected]>
Cc: Shyrabokau, Anton <[email protected]>
Subject: Re: Proposal to add page headers to SLRU pages
Date: Wed, 6 Mar 2024 12:01:46 +0000
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <CAJ7c6TMUY=H2Hn=YbdYimk_NfVGq1QR9UhaPsE2Zd+8rgzdZMg@mail.gmail.com>
<IA1PR07MB982925731C4C0B38DC8F2F7FB98BA@IA1PR07MB9829.namprd07.prod.outlook.com>
<[email protected]>
<[email protected]>
<[email protected]>
<CA+TgmoZ2i5smJuGDRqsUbEKHga_xOc27f6=59ci9aLjA0c2_KA@mail.gmail.com>
<DM4PR07MB9812DA12DA729E8C709DEDD9B98AA@DM4PR07MB9812.namprd07.prod.outlook.com>
<[email protected]>
<[email protected]>
<CAJ7c6TMrFrUF1W3nO=w+3mJu_G8UaQkMJh2aV9WhP20Xs0M_6g@mail.gmail.com>
<[email protected]>
<[email protected]>
Rebase the patch against the latest HEAD.
Regards,
Yong
Attachments:
[application/octet-stream] slru_page_header_v5.patch (55.4K, ../[email protected]/3-slru_page_header_v5.patch)
download | inline diff:
src/backend/access/transam/clog.c | 52 +++++-----
src/backend/access/transam/commit_ts.c | 26 +++--
src/backend/access/transam/multixact.c | 63 +++++++-----
src/backend/access/transam/slru.c | 112 +++++++--------------
src/backend/access/transam/subtrans.c | 12 +--
src/backend/commands/async.c | 27 +++---
src/backend/storage/lmgr/predicate.c | 16 +--
src/backend/storage/page/bufpage.c | 25 +++++
src/bin/pg_checksums/pg_checksums.c | 9 ++
src/bin/pg_resetwal/t/001_basic.pl | 6 +-
src/bin/pg_upgrade/file.c | 172 +++++++++++++++++++++++++++++++++
src/bin/pg_upgrade/pg_upgrade.c | 23 +++--
src/bin/pg_upgrade/pg_upgrade.h | 6 ++
src/include/access/slru.h | 15 +--
src/include/catalog/catversion.h | 2 +-
src/include/storage/bufpage.h | 7 ++
src/test/modules/test_slru/test_slru.c | 13 +--
17 files changed, 395 insertions(+), 191 deletions(-)
diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c
index 44c253246b..4ed4c82f05 100644
--- a/src/backend/access/transam/clog.c
+++ b/src/backend/access/transam/clog.c
@@ -42,6 +42,7 @@
#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
+#include "storage/bufpage.h"
#include "storage/proc.h"
#include "storage/sync.h"
#include "utils/guc_hooks.h"
@@ -61,7 +62,7 @@
/* 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 (BLCKSZ * CLOG_XACTS_PER_BYTE)
+#define CLOG_XACTS_PER_PAGE (SizeOfPageContents * CLOG_XACTS_PER_BYTE)
#define CLOG_XACT_BITMASK ((1 << CLOG_BITS_PER_XACT) - 1)
/*
@@ -88,13 +89,6 @@ TransactionIdToPage(TransactionId xid)
#define TransactionIdToByte(xid) (TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
#define TransactionIdToBIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_BYTE)
-/* We store the latest async LSN for each group of transactions */
-#define CLOG_XACTS_PER_LSN_GROUP 32 /* keep this a power of 2 */
-#define CLOG_LSNS_PER_PAGE (CLOG_XACTS_PER_PAGE / CLOG_XACTS_PER_LSN_GROUP)
-
-#define GetLSNIndex(slotno, xid) ((slotno) * CLOG_LSNS_PER_PAGE + \
- ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE) / CLOG_XACTS_PER_LSN_GROUP)
-
/*
* The number of subtransactions below which we consider to apply clog group
* update optimization. Testing reveals that the number higher than this can
@@ -112,7 +106,7 @@ static SlruCtlData XactCtlData;
static int ZeroCLOGPage(int64 pageno, bool writeXlog);
static bool CLOGPagePrecedes(int64 page1, int64 page2);
-static void WriteZeroPageXlogRec(int64 pageno);
+static XLogRecPtr WriteZeroPageXlogRec(int64 pageno);
static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXact,
Oid oldestXactDb);
static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
@@ -670,8 +664,9 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
Assert(LWLockHeldByMeInMode(SimpleLruGetBankLock(XactCtl,
XactCtl->shared->page_number[slotno]),
LW_EXCLUSIVE));
+ Page page = XactCtl->shared->page_buffer[slotno];
- byteptr = XactCtl->shared->page_buffer[slotno] + byteno;
+ byteptr = PageGetContents(page) + byteno;
curval = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
/*
@@ -700,7 +695,7 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
*byteptr = byteval;
/*
- * Update the group LSN if the transaction completion LSN is higher.
+ * Update the page LSN if the transaction completion LSN is higher.
*
* Note: lsn will be invalid when supplied during InRecovery processing,
* so we don't need to do anything special to avoid LSN updates during
@@ -709,10 +704,8 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
*/
if (!XLogRecPtrIsInvalid(lsn))
{
- int lsnindex = GetLSNIndex(slotno, xid);
-
- if (XactCtl->shared->group_lsn[lsnindex] < lsn)
- XactCtl->shared->group_lsn[lsnindex] = lsn;
+ if (PageGetLSN(page) < lsn)
+ PageSetLSN(page, lsn);
}
}
@@ -738,19 +731,19 @@ TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
int byteno = TransactionIdToByte(xid);
int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
int slotno;
- int lsnindex;
+ Page page;
char *byteptr;
XidStatus status;
/* lock is acquired by SimpleLruReadPage_ReadOnly */
slotno = SimpleLruReadPage_ReadOnly(XactCtl, pageno, xid);
- byteptr = XactCtl->shared->page_buffer[slotno] + byteno;
+ page = XactCtl->shared->page_buffer[slotno];
+ byteptr = PageGetContents(page) + byteno;
status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
- lsnindex = GetLSNIndex(slotno, xid);
- *lsn = XactCtl->shared->group_lsn[lsnindex];
+ *lsn = PageGetLSN(page);
LWLockRelease(SimpleLruGetBankLock(XactCtl, pageno));
@@ -780,7 +773,7 @@ CLOGShmemBuffers(void)
Size
CLOGShmemSize(void)
{
- return SimpleLruShmemSize(CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE);
+ return SimpleLruShmemSize(CLOGShmemBuffers());
}
void
@@ -808,7 +801,7 @@ CLOGShmemInit(void)
Assert(transaction_buffers != 0);
XactCtl->PagePrecedes = CLOGPagePrecedes;
- SimpleLruInit(XactCtl, "transaction", CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE,
+ SimpleLruInit(XactCtl, "transaction", CLOGShmemBuffers(),
"pg_xact", LWTRANCHE_XACT_BUFFER,
LWTRANCHE_XACT_SLRU, SYNC_HANDLER_CLOG, false);
SlruPagePrecedesUnitTests(XactCtl, CLOG_XACTS_PER_PAGE);
@@ -860,11 +853,17 @@ static int
ZeroCLOGPage(int64 pageno, bool writeXlog)
{
int slotno;
+ Page page;
+ XLogRecPtr lsn = 0;
slotno = SimpleLruZeroPage(XactCtl, pageno);
+ page = XactCtl->shared->page_buffer[slotno];
if (writeXlog)
- WriteZeroPageXlogRec(pageno);
+ {
+ lsn = WriteZeroPageXlogRec(pageno);
+ PageSetLSN(page, lsn);
+ }
return slotno;
}
@@ -917,12 +916,12 @@ TrimCLOG(void)
char *byteptr;
slotno = SimpleLruReadPage(XactCtl, pageno, false, xid);
- byteptr = XactCtl->shared->page_buffer[slotno] + byteno;
+ byteptr = PageGetContents(XactCtl->shared->page_buffer[slotno]) + byteno;
/* Zero so-far-unused positions in the current byte */
*byteptr &= (1 << bshift) - 1;
/* Zero the rest of the page */
- MemSet(byteptr + 1, 0, BLCKSZ - byteno - 1);
+ MemSet(byteptr + 1, 0, SizeOfPageContents - byteno - 1);
XactCtl->shared->page_dirty[slotno] = true;
}
@@ -946,7 +945,6 @@ CheckPointCLOG(void)
TRACE_POSTGRESQL_CLOG_CHECKPOINT_DONE(true);
}
-
/*
* Make sure that CLOG has room for a newly-allocated XID.
*
@@ -1070,12 +1068,12 @@ CLOGPagePrecedes(int64 page1, int64 page2)
/*
* Write a ZEROPAGE xlog record
*/
-static void
+static XLogRecPtr
WriteZeroPageXlogRec(int64 pageno)
{
XLogBeginInsert();
XLogRegisterData((char *) (&pageno), sizeof(pageno));
- (void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE);
+ return XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE);
}
/*
diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c
index f221494687..dce06d6cf6 100644
--- a/src/backend/access/transam/commit_ts.c
+++ b/src/backend/access/transam/commit_ts.c
@@ -29,6 +29,7 @@
#include "access/xlogutils.h"
#include "funcapi.h"
#include "miscadmin.h"
+#include "storage/bufpage.h"
#include "storage/shmem.h"
#include "utils/fmgrprotos.h"
#include "utils/guc_hooks.h"
@@ -61,7 +62,7 @@ typedef struct CommitTimestampEntry
sizeof(RepOriginId))
#define COMMIT_TS_XACTS_PER_PAGE \
- (BLCKSZ / SizeOfCommitTimestampEntry)
+ (SizeOfPageContents / SizeOfCommitTimestampEntry)
/*
@@ -118,7 +119,7 @@ static int ZeroCommitTsPage(int64 pageno, bool writeXlog);
static bool CommitTsPagePrecedes(int64 page1, int64 page2);
static void ActivateCommitTs(void);
static void DeactivateCommitTs(void);
-static void WriteZeroPageXlogRec(int64 pageno);
+static XLogRecPtr WriteZeroPageXlogRec(int64 pageno);
static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid);
/*
@@ -253,11 +254,12 @@ TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts,
CommitTimestampEntry entry;
Assert(TransactionIdIsNormal(xid));
+ Assert(xid == slotno * COMMIT_TS_XACTS_PER_PAGE + entryno);
entry.time = ts;
entry.nodeid = nodeid;
- memcpy(CommitTsCtl->shared->page_buffer[slotno] +
+ memcpy(PageGetContents(CommitTsCtl->shared->page_buffer[slotno]) +
SizeOfCommitTimestampEntry * entryno,
&entry, SizeOfCommitTimestampEntry);
}
@@ -336,7 +338,7 @@ TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts,
/* lock is acquired by SimpleLruReadPage_ReadOnly */
slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, xid);
memcpy(&entry,
- CommitTsCtl->shared->page_buffer[slotno] +
+ PageGetContents(CommitTsCtl->shared->page_buffer[slotno]) +
SizeOfCommitTimestampEntry * entryno,
SizeOfCommitTimestampEntry);
@@ -518,7 +520,7 @@ CommitTsShmemBuffers(void)
Size
CommitTsShmemSize(void)
{
- return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) +
+ return SimpleLruShmemSize(CommitTsShmemBuffers()) +
sizeof(CommitTimestampShared);
}
@@ -553,7 +555,7 @@ CommitTsShmemInit(void)
Assert(commit_timestamp_buffers != 0);
CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
- SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0,
+ SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(),
"pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER,
LWTRANCHE_COMMITTS_SLRU,
SYNC_HANDLER_COMMIT_TS,
@@ -615,11 +617,17 @@ static int
ZeroCommitTsPage(int64 pageno, bool writeXlog)
{
int slotno;
+ Page page;
+ XLogRecPtr lsn = 0;
slotno = SimpleLruZeroPage(CommitTsCtl, pageno);
+ page = CommitTsCtl->shared->page_buffer[slotno];
if (writeXlog)
- WriteZeroPageXlogRec(pageno);
+ {
+ lsn = WriteZeroPageXlogRec(pageno);
+ PageSetLSN(page, lsn);
+ }
return slotno;
}
@@ -985,12 +993,12 @@ CommitTsPagePrecedes(int64 page1, int64 page2)
/*
* Write a ZEROPAGE xlog record
*/
-static void
+static XLogRecPtr
WriteZeroPageXlogRec(int64 pageno)
{
XLogBeginInsert();
XLogRegisterData((char *) (&pageno), sizeof(pageno));
- (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
+ return XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
}
/*
diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c
index 83b578dced..8d934c7bb3 100644
--- a/src/backend/access/transam/multixact.c
+++ b/src/backend/access/transam/multixact.c
@@ -83,6 +83,7 @@
#include "miscadmin.h"
#include "pg_trace.h"
#include "postmaster/autovacuum.h"
+#include "storage/bufpage.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/procarray.h"
@@ -105,7 +106,7 @@
*/
/* We need four bytes per offset */
-#define MULTIXACT_OFFSETS_PER_PAGE (BLCKSZ / sizeof(MultiXactOffset))
+#define MULTIXACT_OFFSETS_PER_PAGE (SizeOfPageContents / sizeof(MultiXactOffset))
#define MultiXactIdToOffsetPage(xid) \
((xid) / (MultiXactOffset) MULTIXACT_OFFSETS_PER_PAGE)
@@ -118,8 +119,8 @@
* additional flag bits for each TransactionId. To do this without getting
* into alignment issues, we store four bytes of flags, and then the
* corresponding 4 Xids. Each such 5-word (20-byte) set we call a "group", and
- * are stored as a whole in pages. Thus, with 8kB BLCKSZ, we keep 409 groups
- * per page. This wastes 12 bytes per page, but that's OK -- simplicity (and
+ * are stored as a whole in pages. Thus, with 8kB BLCKSZ, we keep 408 groups
+ * per page. This wastes 8 bytes per page, but that's OK -- simplicity (and
* performance) trumps space efficiency here.
*
* Note that the "offset" macros work with byte offset, not array indexes, so
@@ -137,7 +138,7 @@
/* size in bytes of a complete group */
#define MULTIXACT_MEMBERGROUP_SIZE \
(sizeof(TransactionId) * MULTIXACT_MEMBERS_PER_MEMBERGROUP + MULTIXACT_FLAGBYTES_PER_GROUP)
-#define MULTIXACT_MEMBERGROUPS_PER_PAGE (BLCKSZ / MULTIXACT_MEMBERGROUP_SIZE)
+#define MULTIXACT_MEMBERGROUPS_PER_PAGE (SizeOfPageContents / MULTIXACT_MEMBERGROUP_SIZE)
#define MULTIXACT_MEMBERS_PER_PAGE \
(MULTIXACT_MEMBERGROUPS_PER_PAGE * MULTIXACT_MEMBERS_PER_MEMBERGROUP)
@@ -364,7 +365,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(int64 pageno, uint8 info);
+static XLogRecPtr WriteMZeroPageXlogRec(int64 pageno, uint8 info);
static void WriteMTruncateXlogRec(Oid oldestMultiDB,
MultiXactId startTruncOff,
MultiXactId endTruncOff,
@@ -885,7 +886,7 @@ RecordNewMultiXact(MultiXactId multi, MultiXactOffset offset,
* take the trouble to generalize the slru.c error reporting code.
*/
slotno = SimpleLruReadPage(MultiXactOffsetCtl, pageno, true, multi);
- offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
+ offptr = (MultiXactOffset *) PageGetContents(MultiXactOffsetCtl->shared->page_buffer[slotno]);
offptr += entryno;
*offptr = offset;
@@ -934,12 +935,12 @@ RecordNewMultiXact(MultiXactId multi, MultiXactOffset offset,
}
memberptr = (TransactionId *)
- (MultiXactMemberCtl->shared->page_buffer[slotno] + memberoff);
+ (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + memberoff);
*memberptr = members[i].xid;
flagsptr = (uint32 *)
- (MultiXactMemberCtl->shared->page_buffer[slotno] + flagsoff);
+ (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + flagsoff);
flagsval = *flagsptr;
flagsval &= ~(((1 << MXACT_MEMBER_BITS_PER_XACT) - 1) << bshift);
@@ -1364,7 +1365,7 @@ retry:
LWLockAcquire(lock, LW_EXCLUSIVE);
slotno = SimpleLruReadPage(MultiXactOffsetCtl, pageno, true, multi);
- offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
+ offptr = (MultiXactOffset *) PageGetContents(MultiXactOffsetCtl->shared->page_buffer[slotno]);
offptr += entryno;
offset = *offptr;
@@ -1413,7 +1414,7 @@ retry:
slotno = SimpleLruReadPage(MultiXactOffsetCtl, pageno, true, tmpMXact);
}
- offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
+ offptr = (MultiXactOffset *) PageGetContents(MultiXactOffsetCtl->shared->page_buffer[slotno]);
offptr += entryno;
nextMXOffset = *offptr;
@@ -1470,7 +1471,7 @@ retry:
}
xactptr = (TransactionId *)
- (MultiXactMemberCtl->shared->page_buffer[slotno] + memberoff);
+ (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + memberoff);
if (!TransactionIdIsValid(*xactptr))
{
@@ -1481,7 +1482,7 @@ retry:
flagsoff = MXOffsetToFlagsOffset(offset);
bshift = MXOffsetToFlagsBitShift(offset);
- flagsptr = (uint32 *) (MultiXactMemberCtl->shared->page_buffer[slotno] + flagsoff);
+ flagsptr = (uint32 *) (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + flagsoff);
ptr[truelength].xid = *xactptr;
ptr[truelength].status = (*flagsptr >> bshift) & MXACT_MEMBER_XACT_BITMASK;
@@ -1880,8 +1881,8 @@ MultiXactShmemSize(void)
mul_size(sizeof(MultiXactId) * 2, MaxOldestSlot))
size = SHARED_MULTIXACT_STATE_SIZE;
- size = add_size(size, SimpleLruShmemSize(multixact_offset_buffers, 0));
- size = add_size(size, SimpleLruShmemSize(multixact_member_buffers, 0));
+ size = add_size(size, SimpleLruShmemSize(multixact_offset_buffers));
+ size = add_size(size, SimpleLruShmemSize(multixact_member_buffers));
return size;
}
@@ -1897,14 +1898,14 @@ MultiXactShmemInit(void)
MultiXactMemberCtl->PagePrecedes = MultiXactMemberPagePrecedes;
SimpleLruInit(MultiXactOffsetCtl,
- "multixact_offset", multixact_offset_buffers, 0,
+ "multixact_offset", multixact_offset_buffers,
"pg_multixact/offsets", LWTRANCHE_MULTIXACTOFFSET_BUFFER,
LWTRANCHE_MULTIXACTOFFSET_SLRU,
SYNC_HANDLER_MULTIXACT_OFFSET,
false);
SlruPagePrecedesUnitTests(MultiXactOffsetCtl, MULTIXACT_OFFSETS_PER_PAGE);
SimpleLruInit(MultiXactMemberCtl,
- "multixact_member", multixact_member_buffers, 0,
+ "multixact_member", multixact_member_buffers,
"pg_multixact/members", LWTRANCHE_MULTIXACTMEMBER_BUFFER,
LWTRANCHE_MULTIXACTMEMBER_SLRU,
SYNC_HANDLER_MULTIXACT_MEMBER,
@@ -1999,11 +2000,17 @@ static int
ZeroMultiXactOffsetPage(int64 pageno, bool writeXlog)
{
int slotno;
+ Page page;
+ XLogRecPtr lsn = 0;
slotno = SimpleLruZeroPage(MultiXactOffsetCtl, pageno);
+ page = MultiXactOffsetCtl->shared->page_buffer[slotno];
if (writeXlog)
- WriteMZeroPageXlogRec(pageno, XLOG_MULTIXACT_ZERO_OFF_PAGE);
+ {
+ lsn = WriteMZeroPageXlogRec(pageno, XLOG_MULTIXACT_ZERO_OFF_PAGE);
+ PageSetLSN(page, lsn);
+ }
return slotno;
}
@@ -2015,11 +2022,17 @@ static int
ZeroMultiXactMemberPage(int64 pageno, bool writeXlog)
{
int slotno;
+ Page page;
+ XLogRecPtr lsn = 0;
slotno = SimpleLruZeroPage(MultiXactMemberCtl, pageno);
+ page = MultiXactMemberCtl->shared->page_buffer[slotno];
if (writeXlog)
- WriteMZeroPageXlogRec(pageno, XLOG_MULTIXACT_ZERO_MEM_PAGE);
+ {
+ lsn = WriteMZeroPageXlogRec(pageno, XLOG_MULTIXACT_ZERO_MEM_PAGE);
+ PageSetLSN(page, lsn);
+ }
return slotno;
}
@@ -2143,10 +2156,10 @@ TrimMultiXact(void)
LWLockAcquire(lock, LW_EXCLUSIVE);
slotno = SimpleLruReadPage(MultiXactOffsetCtl, pageno, true, nextMXact);
- offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
+ offptr = (MultiXactOffset *) PageGetContents(MultiXactOffsetCtl->shared->page_buffer[slotno]);
offptr += entryno;
- MemSet(offptr, 0, BLCKSZ - (entryno * sizeof(MultiXactOffset)));
+ MemSet(offptr, 0, SizeOfPageContents - (entryno * sizeof(MultiXactOffset)));
MultiXactOffsetCtl->shared->page_dirty[slotno] = true;
LWLockRelease(lock);
@@ -2177,9 +2190,9 @@ TrimMultiXact(void)
memberoff = MXOffsetToMemberOffset(offset);
slotno = SimpleLruReadPage(MultiXactMemberCtl, pageno, true, offset);
xidptr = (TransactionId *)
- (MultiXactMemberCtl->shared->page_buffer[slotno] + memberoff);
+ (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + memberoff);
- MemSet(xidptr, 0, BLCKSZ - memberoff);
+ MemSet(xidptr, 0, SizeOfPageContents - memberoff);
/*
* Note: we don't need to zero out the flag bits in the remaining
@@ -2834,7 +2847,7 @@ find_multixact_start(MultiXactId multi, MultiXactOffset *result)
/* lock is acquired by SimpleLruReadPage_ReadOnly */
slotno = SimpleLruReadPage_ReadOnly(MultiXactOffsetCtl, pageno, multi);
- offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno];
+ offptr = (MultiXactOffset *) PageGetContents(MultiXactOffsetCtl->shared->page_buffer[slotno]);
offptr += entryno;
offset = *offptr;
LWLockRelease(SimpleLruGetBankLock(MultiXactOffsetCtl, pageno));
@@ -3268,12 +3281,12 @@ MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2)
* Write an xlog record reflecting the zeroing of either a MEMBERs or
* OFFSETs page (info shows which)
*/
-static void
+static XLogRecPtr
WriteMZeroPageXlogRec(int64 pageno, uint8 info)
{
XLogBeginInsert();
XLogRegisterData((char *) (&pageno), sizeof(pageno));
- (void) XLogInsert(RM_MULTIXACT_ID, info);
+ return XLogInsert(RM_MULTIXACT_ID, info);
}
/*
diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c
index f99ec38a4a..aded4d9708 100644
--- a/src/backend/access/transam/slru.c
+++ b/src/backend/access/transam/slru.c
@@ -68,6 +68,7 @@
#include "access/xlogutils.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "storage/bufpage.h"
#include "storage/fd.h"
#include "storage/shmem.h"
#include "utils/guc_hooks.h"
@@ -155,13 +156,13 @@ typedef enum
SLRU_WRITE_FAILED,
SLRU_FSYNC_FAILED,
SLRU_CLOSE_FAILED,
+ SLRU_DATA_CORRUPTED,
} SlruErrorCause;
static SlruErrorCause slru_errcause;
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, int64 pageno, int slotno);
@@ -181,7 +182,7 @@ static inline void SlruRecentlyUsed(SlruShared shared, int slotno);
*/
Size
-SimpleLruShmemSize(int nslots, int nlsns)
+SimpleLruShmemSize(int nslots)
{
int nbanks = nslots / SLRU_BANK_SIZE;
Size sz;
@@ -200,9 +201,6 @@ SimpleLruShmemSize(int nslots, int nlsns)
sz += MAXALIGN(nbanks * sizeof(LWLockPadded)); /* bank_locks[] */
sz += MAXALIGN(nbanks * sizeof(int)); /* bank_cur_lru_count[] */
- if (nlsns > 0)
- sz += MAXALIGN(nslots * nlsns * sizeof(XLogRecPtr)); /* group_lsn[] */
-
return BUFFERALIGN(sz) + BLCKSZ * nslots;
}
@@ -227,7 +225,6 @@ SimpleLruAutotuneBuffers(int divisor, int max)
* ctl: address of local (unshared) control structure.
* name: name of SLRU. (This is user-visible, pick with care!)
* nslots: number of page slots to use.
- * nlsns: number of LSN groups per page (set to zero if not relevant).
* ctllock: LWLock to use to control access to the shared control structure.
* subdir: PGDATA-relative subdirectory that will contain the files.
* buffer_tranche_id: tranche ID to use for the SLRU's per-buffer LWLocks.
@@ -235,7 +232,7 @@ SimpleLruAutotuneBuffers(int divisor, int max)
* sync_handler: which set of functions to use to handle sync requests
*/
void
-SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
+SimpleLruInit(SlruCtl ctl, const char *name, int nslots,
const char *subdir, int buffer_tranche_id, int bank_tranche_id,
SyncRequestHandler sync_handler, bool long_segment_names)
{
@@ -246,7 +243,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
Assert(nslots <= SLRU_MAX_ALLOWED_BUFFERS);
shared = (SlruShared) ShmemInitStruct(name,
- SimpleLruShmemSize(nslots, nlsns),
+ SimpleLruShmemSize(nslots),
&found);
if (!IsUnderPostmaster)
@@ -260,7 +257,6 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
memset(shared, 0, sizeof(SlruSharedData));
shared->num_slots = nslots;
- shared->lsn_groups_per_page = nlsns;
pg_atomic_init_u64(&shared->latest_page_number, 0);
@@ -287,12 +283,6 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
shared->bank_cur_lru_count = (int *) (ptr + offset);
offset += MAXALIGN(nbanks * sizeof(int));
- if (nlsns > 0)
- {
- shared->group_lsn = (XLogRecPtr *) (ptr + offset);
- offset += MAXALIGN(nslots * nlsns * sizeof(XLogRecPtr));
- }
-
ptr += BUFFERALIGN(offset);
for (int slotno = 0; slotno < nslots; slotno++)
{
@@ -314,7 +304,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
}
/* Should fit to estimated shmem size */
- Assert(ptr - (char *) shared <= SimpleLruShmemSize(nslots, nlsns));
+ Assert(ptr - (char *) shared <= SimpleLruShmemSize(nslots));
}
else
{
@@ -378,11 +368,8 @@ SimpleLruZeroPage(SlruCtl ctl, int64 pageno)
shared->page_dirty[slotno] = true;
SlruRecentlyUsed(shared, slotno);
- /* Set the buffer to zeroes */
- MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
-
- /* Set the LSNs for this new page to zero */
- SimpleLruZeroLSNs(ctl, slotno);
+ /* Initialize the page. */
+ PageInitSLRU(shared->page_buffer[slotno], BLCKSZ, 0);
/*
* Assume this page is now the latest active page.
@@ -400,26 +387,6 @@ SimpleLruZeroPage(SlruCtl ctl, int64 pageno)
return slotno;
}
-/*
- * Zero all the LSNs we store for this slru page.
- *
- * This should be called each time we create a new page, and each time we read
- * in a page from disk into an existing buffer. (Such an old page cannot
- * have any interesting LSNs, since we'd have flushed them before writing
- * the page in the first place.)
- *
- * This assumes that InvalidXLogRecPtr is bitwise-all-0.
- */
-static void
-SimpleLruZeroLSNs(SlruCtl ctl, int slotno)
-{
- SlruShared shared = ctl->shared;
-
- if (shared->lsn_groups_per_page > 0)
- MemSet(&shared->group_lsn[slotno * shared->lsn_groups_per_page], 0,
- shared->lsn_groups_per_page * sizeof(XLogRecPtr));
-}
-
/*
* Wait for any active I/O on a page slot to finish. (This does not
* guarantee that new I/O hasn't been started before we return, though.
@@ -546,9 +513,6 @@ SimpleLruReadPage(SlruCtl ctl, int64 pageno, bool write_ok,
/* Do the read */
ok = SlruPhysicalReadPage(ctl, pageno, slotno);
- /* Set the LSNs for this newly read-in page to zero */
- SimpleLruZeroLSNs(ctl, slotno);
-
/* Re-acquire bank control lock and update page state */
LWLockAcquire(banklock, LW_EXCLUSIVE);
@@ -815,7 +779,7 @@ SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno)
ereport(LOG,
(errmsg("file \"%s\" doesn't exist, reading as zeroes",
path)));
- MemSet(shared->page_buffer[slotno], 0, BLCKSZ);
+ PageInitSLRU(shared->page_buffer[slotno], BLCKSZ, 0);
return true;
}
@@ -838,6 +802,13 @@ SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno)
return false;
}
+ if (!PageIsVerifiedExtended(shared->page_buffer[slotno], pageno, PIV_REPORT_STAT))
+ {
+ slru_errcause = SLRU_DATA_CORRUPTED;
+ slru_errno = 0;
+ return false;
+ }
+
return true;
}
@@ -864,6 +835,8 @@ SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
off_t offset = rpageno * BLCKSZ;
char path[MAXPGPATH];
int fd = -1;
+ Page page = shared->page_buffer[slotno];
+ XLogRecPtr lsn;
/* update the stats counter of written pages */
pgstat_count_slru_page_written(shared->slru_stats_idx);
@@ -873,40 +846,18 @@ SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
* write out data before associated WAL records. This is the same action
* performed during FlushBuffer() in the main buffer manager.
*/
- if (shared->group_lsn != NULL)
+ lsn = PageGetLSN(page);
+ if (!XLogRecPtrIsInvalid(lsn))
{
/*
- * We must determine the largest async-commit LSN for the page. This
- * is a bit tedious, but since this entire function is a slow path
- * anyway, it seems better to do this here than to maintain a per-page
- * LSN variable (which'd need an extra comparison in the
- * transaction-commit path).
+ * As noted above, elog(ERROR) is not acceptable here, so if
+ * XLogFlush were to fail, we must PANIC. This isn't much of a
+ * restriction because XLogFlush is just about all critical
+ * section anyway, but let's make sure.
*/
- XLogRecPtr max_lsn;
- int lsnindex;
-
- lsnindex = slotno * shared->lsn_groups_per_page;
- max_lsn = shared->group_lsn[lsnindex++];
- for (int lsnoff = 1; lsnoff < shared->lsn_groups_per_page; lsnoff++)
- {
- XLogRecPtr this_lsn = shared->group_lsn[lsnindex++];
-
- if (max_lsn < this_lsn)
- max_lsn = this_lsn;
- }
-
- if (!XLogRecPtrIsInvalid(max_lsn))
- {
- /*
- * As noted above, elog(ERROR) is not acceptable here, so if
- * XLogFlush were to fail, we must PANIC. This isn't much of a
- * restriction because XLogFlush is just about all critical
- * section anyway, but let's make sure.
- */
- START_CRIT_SECTION();
- XLogFlush(max_lsn);
- END_CRIT_SECTION();
- }
+ START_CRIT_SECTION();
+ XLogFlush(lsn);
+ END_CRIT_SECTION();
}
/*
@@ -971,6 +922,8 @@ SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
}
}
+ PageSetChecksumInplace(shared->page_buffer[slotno], pageno);
+
errno = 0;
pgstat_report_wait_start(WAIT_EVENT_SLRU_WRITE);
if (pg_pwrite(fd, shared->page_buffer[slotno], BLCKSZ, offset) != BLCKSZ)
@@ -1091,6 +1044,13 @@ SlruReportIOError(SlruCtl ctl, int64 pageno, TransactionId xid)
errdetail("Could not close file \"%s\": %m.",
path)));
break;
+ case SLRU_DATA_CORRUPTED:
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("could not access status of transaction %u", xid),
+ errdetail("Invalid page from file \"%s\" at offset %d.",
+ path, offset)));
+ break;
default:
/* can't get here, we trust */
elog(ERROR, "unrecognized SimpleLru error cause: %d",
diff --git a/src/backend/access/transam/subtrans.c b/src/backend/access/transam/subtrans.c
index 50bb1d8cfc..5549cac3a6 100644
--- a/src/backend/access/transam/subtrans.c
+++ b/src/backend/access/transam/subtrans.c
@@ -34,6 +34,7 @@
#include "miscadmin.h"
#include "pg_trace.h"
#include "utils/guc_hooks.h"
+#include "storage/bufpage.h"
#include "utils/snapmgr.h"
@@ -51,7 +52,7 @@
*/
/* We need four bytes per xact */
-#define SUBTRANS_XACTS_PER_PAGE (BLCKSZ / sizeof(TransactionId))
+#define SUBTRANS_XACTS_PER_PAGE (SizeOfPageContents / sizeof(TransactionId))
/*
* Although we return an int64 the actual value can't currently exceed
@@ -97,7 +98,7 @@ SubTransSetParent(TransactionId xid, TransactionId parent)
LWLockAcquire(lock, LW_EXCLUSIVE);
slotno = SimpleLruReadPage(SubTransCtl, pageno, true, xid);
- ptr = (TransactionId *) SubTransCtl->shared->page_buffer[slotno];
+ ptr = (TransactionId *) PageGetContents(SubTransCtl->shared->page_buffer[slotno]);
ptr += entryno;
/*
@@ -137,7 +138,7 @@ SubTransGetParent(TransactionId xid)
/* lock is acquired by SimpleLruReadPage_ReadOnly */
slotno = SimpleLruReadPage_ReadOnly(SubTransCtl, pageno, xid);
- ptr = (TransactionId *) SubTransCtl->shared->page_buffer[slotno];
+ ptr = (TransactionId *) PageGetContents(SubTransCtl->shared->page_buffer[slotno]);
ptr += entryno;
parent = *ptr;
@@ -213,7 +214,7 @@ SUBTRANSShmemBuffers(void)
Size
SUBTRANSShmemSize(void)
{
- return SimpleLruShmemSize(SUBTRANSShmemBuffers(), 0);
+ return SimpleLruShmemSize(SUBTRANSShmemBuffers());
}
void
@@ -241,7 +242,7 @@ SUBTRANSShmemInit(void)
Assert(subtransaction_buffers != 0);
SubTransCtl->PagePrecedes = SubTransPagePrecedes;
- SimpleLruInit(SubTransCtl, "subtransaction", SUBTRANSShmemBuffers(), 0,
+ SimpleLruInit(SubTransCtl, "subtransaction", SUBTRANSShmemBuffers(),
"pg_subtrans", LWTRANCHE_SUBTRANS_BUFFER,
LWTRANCHE_SUBTRANS_SLRU, SYNC_HANDLER_NONE, false);
SlruPagePrecedesUnitTests(SubTransCtl, SUBTRANS_XACTS_PER_PAGE);
@@ -366,7 +367,6 @@ CheckPointSUBTRANS(void)
TRACE_POSTGRESQL_SUBTRANS_CHECKPOINT_DONE(true);
}
-
/*
* Make sure that SUBTRANS has room for a newly-allocated XID.
*
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index d0891e3f0e..a400c54c03 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -140,6 +140,7 @@
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
+#include "storage/bufpage.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/procsignal.h"
@@ -160,7 +161,7 @@
* than that, so changes in that data structure won't affect user-visible
* restrictions.
*/
-#define NOTIFY_PAYLOAD_MAX_LENGTH (BLCKSZ - NAMEDATALEN - 128)
+#define NOTIFY_PAYLOAD_MAX_LENGTH (SizeOfPageContents - NAMEDATALEN - 128)
/*
* Struct representing an entry in the global notify queue
@@ -309,7 +310,7 @@ static SlruCtlData NotifyCtlData;
#define NotifyCtl (&NotifyCtlData)
#define QUEUE_PAGESIZE BLCKSZ
-
+#define QUEUE_PAGE_CAPACITY (QUEUE_PAGESIZE - MAXALIGN(SizeOfPageHeaderData))
#define QUEUE_FULL_WARN_INTERVAL 5000 /* warn at most once every 5s */
/*
@@ -490,7 +491,7 @@ AsyncShmemSize(void)
size = mul_size(MaxBackends, sizeof(QueueBackendStatus));
size = add_size(size, offsetof(AsyncQueueControl, backend));
- size = add_size(size, SimpleLruShmemSize(notify_buffers, 0));
+ size = add_size(size, SimpleLruShmemSize(notify_buffers));
return size;
}
@@ -535,7 +536,7 @@ AsyncShmemInit(void)
* names are used in order to avoid wraparound.
*/
NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
- SimpleLruInit(NotifyCtl, "notify", notify_buffers, 0,
+ SimpleLruInit(NotifyCtl, "notify", notify_buffers,
"pg_notify", LWTRANCHE_NOTIFY_BUFFER, LWTRANCHE_NOTIFY_SLRU,
SYNC_HANDLER_NONE, true);
@@ -1295,14 +1296,14 @@ asyncQueueAdvance(volatile QueuePosition *position, int entryLength)
* written or read.
*/
offset += entryLength;
- Assert(offset <= QUEUE_PAGESIZE);
+ Assert(offset <= QUEUE_PAGE_CAPACITY);
/*
* In a second step check if another entry can possibly be written to the
* page. If so, stay here, we have reached the next position. If not, then
* we need to move on to the next page.
*/
- if (offset + QUEUEALIGN(AsyncQueueEntryEmptySize) > QUEUE_PAGESIZE)
+ if (offset + QUEUEALIGN(AsyncQueueEntryEmptySize) > QUEUE_PAGE_CAPACITY)
{
pageno++;
offset = 0;
@@ -1405,7 +1406,7 @@ asyncQueueAddEntries(ListCell *nextNotify)
offset = QUEUE_POS_OFFSET(queue_head);
/* Check whether the entry really fits on the current page */
- if (offset + qe.length <= QUEUE_PAGESIZE)
+ if (offset + qe.length <= QUEUE_PAGE_CAPACITY)
{
/* OK, so advance nextNotify past this item */
nextNotify = lnext(pendingNotifies->events, nextNotify);
@@ -1417,14 +1418,14 @@ asyncQueueAddEntries(ListCell *nextNotify)
* only check dboid and since it won't match any reader's database
* OID, they will ignore this entry and move on.
*/
- qe.length = QUEUE_PAGESIZE - offset;
+ qe.length = QUEUE_PAGE_CAPACITY - offset;
qe.dboid = InvalidOid;
qe.data[0] = '\0'; /* empty channel */
qe.data[1] = '\0'; /* empty payload */
}
/* Now copy qe into the shared buffer page */
- memcpy(NotifyCtl->shared->page_buffer[slotno] + offset,
+ memcpy(PageGetContents(NotifyCtl->shared->page_buffer[slotno]) + offset,
&qe,
qe.length);
@@ -1955,10 +1956,10 @@ asyncQueueReadAllNotifications(void)
else
{
/* fetch all the rest of the page */
- copysize = QUEUE_PAGESIZE - curoffset;
+ copysize = QUEUE_PAGE_CAPACITY - curoffset;
}
- memcpy(page_buffer.buf + curoffset,
- NotifyCtl->shared->page_buffer[slotno] + curoffset,
+ memcpy(PageGetContents(page_buffer.buf) + curoffset,
+ PageGetContents(NotifyCtl->shared->page_buffer[slotno]) + curoffset,
copysize);
/* Release lock that we got from SimpleLruReadPage_ReadOnly() */
LWLockRelease(SimpleLruGetBankLock(NotifyCtl, curpage));
@@ -2029,7 +2030,7 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
if (QUEUE_POS_EQUAL(thisentry, stop))
break;
- qe = (AsyncQueueEntry *) (page_buffer + QUEUE_POS_OFFSET(thisentry));
+ qe = (AsyncQueueEntry *) (PageGetContents(page_buffer) + QUEUE_POS_OFFSET(thisentry));
/*
* Advance *current over this message, possibly to the next page. As
diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c
index 3f378c0099..c26ae8ddd2 100644
--- a/src/backend/storage/lmgr/predicate.c
+++ b/src/backend/storage/lmgr/predicate.c
@@ -207,6 +207,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "port/pg_lfind.h"
+#include "storage/bufpage.h"
#include "storage/predicate.h"
#include "storage/predicate_internals.h"
#include "storage/proc.h"
@@ -326,8 +327,8 @@ static SlruCtlData SerialSlruCtlData;
#define SerialSlruCtl (&SerialSlruCtlData)
#define SERIAL_PAGESIZE BLCKSZ
-#define SERIAL_ENTRYSIZE sizeof(SerCommitSeqNo)
-#define SERIAL_ENTRIESPERPAGE (SERIAL_PAGESIZE / SERIAL_ENTRYSIZE)
+#define SERIAL_ENTRYSIZE sizeof(SerCommitSeqNo)
+#define SERIAL_ENTRIESPERPAGE (SERIAL_PAGESIZE - MAXALIGN(SizeOfPageHeaderData) / SERIAL_ENTRYSIZE)
/*
* Set maximum pages based on the number needed to track all transactions.
@@ -337,7 +338,7 @@ static SlruCtlData SerialSlruCtlData;
#define SerialNextPage(page) (((page) >= SERIAL_MAX_PAGE) ? 0 : (page) + 1)
#define SerialValue(slotno, xid) (*((SerCommitSeqNo *) \
- (SerialSlruCtl->shared->page_buffer[slotno] + \
+ (PageGetContents(SerialSlruCtl->shared->page_buffer[slotno]) + \
((((uint32) (xid)) % SERIAL_ENTRIESPERPAGE) * SERIAL_ENTRYSIZE))))
#define SerialPage(xid) (((uint32) (xid)) / SERIAL_ENTRIESPERPAGE)
@@ -789,10 +790,13 @@ SerialPagePrecedesLogicallyUnitTests(void)
* requires burning ~2B XIDs in single-user mode, a negligible
* possibility. Moreover, if it does happen, the consequence would be
* mild, namely a new transaction failing in SimpleLruReadPage().
+ *
+ * NOTE: After adding the page header, the defect affects two pages.
+ * We now assert correct treatment of its second to prior page.
*/
headPage = oldestPage;
targetPage = newestPage;
- Assert(SerialPagePrecedesLogically(headPage, targetPage - 1));
+ Assert(SerialPagePrecedesLogically(headPage, targetPage - 2));
#if 0
Assert(SerialPagePrecedesLogically(headPage, targetPage));
#endif
@@ -812,7 +816,7 @@ SerialInit(void)
*/
SerialSlruCtl->PagePrecedes = SerialPagePrecedesLogically;
SimpleLruInit(SerialSlruCtl, "serializable",
- serializable_buffers, 0, "pg_serial",
+ serializable_buffers, "pg_serial",
LWTRANCHE_SERIAL_BUFFER, LWTRANCHE_SERIAL_SLRU,
SYNC_HANDLER_NONE, false);
#ifdef USE_ASSERT_CHECKING
@@ -1377,7 +1381,7 @@ PredicateLockShmemSize(void)
/* Shared memory structures for SLRU tracking of old committed xids. */
size = add_size(size, sizeof(SerialControlData));
- size = add_size(size, SimpleLruShmemSize(serializable_buffers, 0));
+ size = add_size(size, SimpleLruShmemSize(serializable_buffers));
return size;
}
diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index be6f1f62d2..e8193d7f56 100644
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -59,6 +59,31 @@ PageInit(Page page, Size pageSize, Size specialSize)
/* p->pd_prune_xid = InvalidTransactionId; done by above MemSet */
}
+/*
+ * PageInitSLRU
+ * Initializes the contents of an SLRU page.
+ * Note that we don't calculate an initial checksum here; that's not done
+ * until it's time to write.
+ */
+void
+PageInitSLRU(Page page, Size pageSize, Size specialSize)
+{
+ PageHeader p = (PageHeader) page;
+
+ specialSize = MAXALIGN(specialSize);
+
+ Assert(pageSize == BLCKSZ);
+ Assert(pageSize > specialSize + SizeOfPageHeaderData);
+
+ /* Make sure all fields of page are zero, as well as unused space */
+ MemSet(p, 0, pageSize);
+
+ p->pd_flags = 0;
+ p->pd_lower = SizeOfPageHeaderData;
+ p->pd_upper = pageSize - specialSize;
+ p->pd_special = pageSize - specialSize;
+ PageSetPageSizeAndVersion(page, pageSize, PG_SLRU_PAGE_LAYOUT_VERSION);
+}
/*
* PageIsVerifiedExtended
diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c
index 9e6fd435f6..06f14f1d2d 100644
--- a/src/bin/pg_checksums/pg_checksums.c
+++ b/src/bin/pg_checksums/pg_checksums.c
@@ -16,6 +16,7 @@
#include <dirent.h>
#include <limits.h>
+#include <stdbool.h>
#include <time.h>
#include <sys/stat.h>
#include <unistd.h>
@@ -593,12 +594,20 @@ main(int argc, char *argv[])
{
total_size = scan_directory(DataDir, "global", true);
total_size += scan_directory(DataDir, "base", true);
+ total_size += scan_directory(DataDir, "pg_commit_ts", true);
+ total_size += scan_directory(DataDir, "pg_multixact", true);
+ total_size += scan_directory(DataDir, "pg_serial", true);
total_size += scan_directory(DataDir, "pg_tblspc", true);
+ total_size += scan_directory(DataDir, "pg_xact", true);
}
(void) scan_directory(DataDir, "global", false);
(void) scan_directory(DataDir, "base", false);
+ (void) scan_directory(DataDir, "pg_commit_ts", false);
+ (void) scan_directory(DataDir, "pg_multixact", false);
+ (void) scan_directory(DataDir, "pg_serial", false);
(void) scan_directory(DataDir, "pg_tblspc", false);
+ (void) scan_directory(DataDir, "pg_xact", false);
if (showprogress)
progress_report(true);
diff --git a/src/bin/pg_resetwal/t/001_basic.pl b/src/bin/pg_resetwal/t/001_basic.pl
index 9829e48106..7b9e034e19 100644
--- a/src/bin/pg_resetwal/t/001_basic.pl
+++ b/src/bin/pg_resetwal/t/001_basic.pl
@@ -206,7 +206,7 @@ push @cmd,
sprintf("%d,%d", hex($files[0]) == 0 ? 3 : hex($files[0]), hex($files[-1]));
@files = get_slru_files('pg_multixact/offsets');
-$mult = 32 * $blcksz / 4;
+$mult = 32 * ($blcksz - 24) / 4;
# -m argument is "new,old"
push @cmd, '-m',
sprintf("%d,%d",
@@ -214,11 +214,11 @@ push @cmd, '-m',
hex($files[0]) == 0 ? 1 : hex($files[0] * $mult));
@files = get_slru_files('pg_multixact/members');
-$mult = 32 * int($blcksz / 20) * 4;
+$mult = 32 * int(($blcksz - 24) / 20) * 4;
push @cmd, '-O', (hex($files[-1]) + 1) * $mult;
@files = get_slru_files('pg_xact');
-$mult = 32 * $blcksz * 4;
+$mult = 32 * ($blcksz - 24) * 4;
push @cmd,
'-u', (hex($files[0]) == 0 ? 3 : hex($files[0]) * $mult),
'-x', ((hex($files[-1]) + 1) * $mult);
diff --git a/src/bin/pg_upgrade/file.c b/src/bin/pg_upgrade/file.c
index beba376f2e..6978410f17 100644
--- a/src/bin/pg_upgrade/file.c
+++ b/src/bin/pg_upgrade/file.c
@@ -9,6 +9,7 @@
#include "postgres_fe.h"
+#include <dirent.h>
#include <sys/stat.h>
#include <limits.h>
#include <fcntl.h>
@@ -453,3 +454,174 @@ check_hard_link(void)
unlink(new_link_file);
}
+
+
+/*
+ * Copy SLRU_PAGES_PER_SEGMENT from access/slru.h to avoid including it.
+ */
+#define SLRU_PAGES_PER_SEGMENT 32
+
+#define SEGMENT_SIZE (BLCKSZ * SLRU_PAGES_PER_SEGMENT)
+
+/*
+ * Copy PageInitSLRU from storage/bufpage.c to avoid linking to the backend.
+ */
+void
+PageInitSLRU(Page page, Size pageSize, Size specialSize)
+{
+ PageHeader p = (PageHeader) page;
+
+ specialSize = MAXALIGN(specialSize);
+
+ Assert(pageSize == BLCKSZ);
+ Assert(pageSize > specialSize + SizeOfPageHeaderData);
+
+ /* Make sure all fields of page are zero, as well as unused space */
+ MemSet(p, 0, pageSize);
+
+ p->pd_flags = 0;
+ p->pd_lower = SizeOfPageHeaderData;
+ p->pd_upper = pageSize - specialSize;
+ p->pd_special = pageSize - specialSize;
+ PageSetPageSizeAndVersion(page, pageSize, PG_SLRU_PAGE_LAYOUT_VERSION);
+}
+
+/*
+ * Filter function for scandir(3) to select only segment files.
+ */
+static int
+segment_file_filter(const struct dirent *dirent)
+{
+ return strspn(dirent->d_name, "0123456789ABCDEF") == strlen(dirent->d_name);
+}
+
+static void
+upgrade_file(const char *src_dir, const char *src_file, const char *dst_dir)
+{
+ char src[MAXPGPATH];
+ char dst[MAXPGPATH];
+
+ int seg_name_len;
+ int src_segno;
+ int64 src_pageno;
+ int dst_segno;
+ int64 dst_pageno;
+ int dst_offset;
+
+ int src_fd;
+ int dst_fd;
+
+ char *src_buf;
+ ssize_t src_len;
+ ssize_t src_buf_offset;
+ PGAlignedBlock dst_block;
+ Page page = dst_block.data;
+ int len_to_copy;
+
+ seg_name_len = strlen(src_file);
+ src_segno = (int) strtol(src_file, NULL, 16);
+ src_pageno = src_segno * SLRU_PAGES_PER_SEGMENT;
+
+ dst_pageno = src_pageno * BLCKSZ / SizeOfPageContents;
+ dst_offset = src_pageno * BLCKSZ - dst_pageno * SizeOfPageContents;
+ dst_segno = dst_pageno / SLRU_PAGES_PER_SEGMENT;
+
+ snprintf(src, sizeof(src), "%s/%s", src_dir, src_file);
+ snprintf(dst, sizeof(dst), "%s/%0*X", dst_dir, seg_name_len, dst_segno);
+
+ src_buf = pg_malloc(SEGMENT_SIZE);
+ if ((src_fd = open(src, O_RDONLY | PG_BINARY, 0)) == -1)
+ pg_fatal("could not open file \"%s\": %s", src, strerror(errno));
+ if ((src_len = read(src_fd, src_buf, SEGMENT_SIZE)) == -1)
+ pg_fatal("could not read file \"%s\": %s", src, strerror(errno));
+
+ if ((dst_fd = open(dst, O_RDWR | O_CREAT | PG_BINARY, S_IRUSR | S_IWUSR)) == -1)
+ pg_fatal("could not open file \"%s\": %s", dst, strerror(errno));
+ if (ftruncate(dst_fd, SEGMENT_SIZE) == -1)
+ pg_fatal("could not truncate file \"%s\": %s", dst, strerror(errno));
+
+ /*
+ * Read the destination page at dst_pageno into the buffer. The page may contain
+ * data from the previous source segment. Initialize the page if the page is new.
+ */
+ if (lseek(dst_fd, (dst_pageno % SLRU_PAGES_PER_SEGMENT) * BLCKSZ, SEEK_SET) == -1)
+ pg_fatal("could not seek in file \"%s\": %s", dst, strerror(errno));
+ if (read(dst_fd, page, BLCKSZ) == -1)
+ pg_fatal("could not read file \"%s\": %s", dst, strerror(errno));
+ if (PageIsNew(page))
+ PageInitSLRU(page, BLCKSZ, 0);
+
+ /*
+ * Rewind the file position, so the first write will overwrite the page.
+ */
+ if (lseek(dst_fd, (dst_pageno % SLRU_PAGES_PER_SEGMENT) * BLCKSZ, SEEK_SET) == -1)
+ pg_fatal("could not seek in file \"%s\": %s", dst, strerror(errno));
+
+ src_buf_offset = 0;
+ while (src_buf_offset < src_len)
+ {
+ len_to_copy = Min(src_len - src_buf_offset, SizeOfPageContents - dst_offset);
+ memcpy(PageGetContents(page) + dst_offset, src_buf + src_buf_offset, len_to_copy);
+ src_buf_offset += len_to_copy;
+
+ if (new_cluster.controldata.data_checksum_version > 0)
+ ((PageHeader) page)->pd_checksum = pg_checksum_page(page, dst_pageno);
+ if (write(dst_fd, page, BLCKSZ) == -1)
+ pg_fatal("could not write file \"%s\": %s", dst, strerror(errno));
+
+ dst_pageno++;
+ dst_offset = 0;
+ PageInitSLRU(page, BLCKSZ, 0);
+
+ /*
+ * Switch segments if we reached the end of the current segment.
+ */
+ if (dst_pageno % SLRU_PAGES_PER_SEGMENT == 0)
+ {
+ if (fsync(dst_fd) == -1)
+ pg_fatal("could not fsync file \"%s\": %s", dst, strerror(errno));
+ if (close(dst_fd) == -1)
+ pg_fatal("could not close file \"%s\": %s", dst, strerror(errno));
+
+ dst_segno++;
+ snprintf(dst, sizeof(dst), "%s/%0*X", dst_dir, seg_name_len, dst_segno);
+ if ((dst_fd = open(dst, O_RDWR | O_CREAT | PG_BINARY, S_IRUSR | S_IWUSR)) == -1)
+ pg_fatal("could not open file \"%s\": %s", dst, strerror(errno));
+ if (ftruncate(dst_fd, SEGMENT_SIZE) == -1)
+ pg_fatal("could not truncate file \"%s\": %s", dst, strerror(errno));
+ }
+ }
+
+ if (fsync(dst_fd) == -1)
+ pg_fatal("could not fsync file \"%s\": %s", dst, strerror(errno));
+ if (close(dst_fd) == -1)
+ pg_fatal("could not close file \"%s\": %s", dst, strerror(errno));
+
+ pg_free(src_buf);
+ close(src_fd);
+}
+
+void
+upgrade_xact_cache(const char *src_subdir, const char *dst_subdir)
+{
+ char src_dir[MAXPGPATH];
+ char dst_dir[MAXPGPATH];
+
+ DIR *src_dirp;
+ struct dirent *src_dirent;
+
+ snprintf(src_dir, sizeof(src_dir), "%s/%s", old_cluster.pgdata, src_subdir);
+ snprintf(dst_dir, sizeof(dst_dir), "%s/%s", new_cluster.pgdata, dst_subdir);
+
+ if ((src_dirp = opendir(src_dir)) == NULL)
+ pg_fatal("could not open directory \"%s\": %s", src_dir, strerror(errno));
+
+ while (errno = 0, (src_dirent = readdir(src_dirp)) != NULL)
+ {
+ if (segment_file_filter(src_dirent))
+ upgrade_file(src_dir, src_dirent->d_name, dst_dir);
+ }
+
+ if (closedir(src_dirp) != 0)
+ pg_fatal("could not close directory \"%s\": %s", src_dir, strerror(errno));
+}
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 10c94a6c1f..ed4a0f5c43 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -660,14 +660,23 @@ copy_subdir_files(const char *old_subdir, const char *new_subdir)
static void
copy_xact_xlog_xid(void)
{
+ bool slru_header_changed = false;
+
/*
* Copy old commit logs to new data dir. pg_clog has been renamed to
* pg_xact in post-10 clusters.
*/
- copy_subdir_files(GET_MAJOR_VERSION(old_cluster.major_version) <= 906 ?
- "pg_clog" : "pg_xact",
- GET_MAJOR_VERSION(new_cluster.major_version) <= 906 ?
- "pg_clog" : "pg_xact");
+ char *xact_old_dir = GET_MAJOR_VERSION(old_cluster.major_version) <= 906 ? "pg_clog" : "pg_xact";
+ char *xact_new_dir = GET_MAJOR_VERSION(new_cluster.major_version) <= 906 ? "pg_clog" : "pg_xact";
+
+ if (new_cluster.controldata.cat_ver >= SLRU_PAGE_HEADER_CAT_VER &&
+ old_cluster.controldata.cat_ver < SLRU_PAGE_HEADER_CAT_VER)
+ slru_header_changed = true;
+
+ if (slru_header_changed)
+ upgrade_xact_cache(xact_old_dir, xact_new_dir);
+ else
+ copy_subdir_files(xact_old_dir, xact_new_dir);
prep_status("Setting oldest XID for new cluster");
exec_prog(UTILITY_LOG_FILE, NULL, true, true,
@@ -702,7 +711,8 @@ copy_xact_xlog_xid(void)
* server doesn't attempt to read multis older than the cutoff value.
*/
if (old_cluster.controldata.cat_ver >= MULTIXACT_FORMATCHANGE_CAT_VER &&
- new_cluster.controldata.cat_ver >= MULTIXACT_FORMATCHANGE_CAT_VER)
+ new_cluster.controldata.cat_ver >= MULTIXACT_FORMATCHANGE_CAT_VER &&
+ !slru_header_changed)
{
copy_subdir_files("pg_multixact/offsets", "pg_multixact/offsets");
copy_subdir_files("pg_multixact/members", "pg_multixact/members");
@@ -722,7 +732,8 @@ copy_xact_xlog_xid(void)
new_cluster.pgdata);
check_ok();
}
- else if (new_cluster.controldata.cat_ver >= MULTIXACT_FORMATCHANGE_CAT_VER)
+ else if (new_cluster.controldata.cat_ver >= MULTIXACT_FORMATCHANGE_CAT_VER ||
+ slru_header_changed)
{
/*
* Remove offsets/0000 file created by initdb that no longer matches
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index 857d715049..dc81020b62 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -115,6 +115,11 @@ extern char *output_files[];
*/
#define MULTIXACT_FORMATCHANGE_CAT_VER 201301231
+/*
+ * A page header was added to each SLRU page in 17.0.
+ */
+#define SLRU_PAGE_HEADER_CAT_VER 202403061
+
/*
* large object chunk size added to pg_controldata,
* commit 5f93c37805e7485488480916b4585e098d3cc883
@@ -412,6 +417,7 @@ void rewriteVisibilityMap(const char *fromfile, const char *tofile,
void check_file_clone(void);
void check_copy_file_range(void);
void check_hard_link(void);
+void upgrade_xact_cache(const char *src_subdir, const char *dst_subdir);
/* fopen_priv() is no longer different from fopen() */
#define fopen_priv(path, mode) fopen(path, mode)
diff --git a/src/include/access/slru.h b/src/include/access/slru.h
index 8a8d191873..5d5bb13cb0 100644
--- a/src/include/access/slru.h
+++ b/src/include/access/slru.h
@@ -96,17 +96,6 @@ typedef struct SlruSharedData
*/
int *bank_cur_lru_count;
- /*
- * Optional array of WAL flush LSNs associated with entries in the SLRU
- * pages. If not zero/NULL, we must flush WAL before writing pages (true
- * for pg_xact, false for everything else). group_lsn[] has
- * lsn_groups_per_page entries per buffer slot, each containing the
- * highest LSN known for a contiguous group of SLRU entries on that slot's
- * page.
- */
- XLogRecPtr *group_lsn;
- int lsn_groups_per_page;
-
/*
* latest_page_number is the page number of the current end of the log;
* this is not critical data, since we use it only to avoid swapping out
@@ -184,9 +173,9 @@ SimpleLruGetBankLock(SlruCtl ctl, int64 pageno)
return &(ctl->shared->bank_locks[bankno].lock);
}
-extern Size SimpleLruShmemSize(int nslots, int nlsns);
+extern Size SimpleLruShmemSize(int nslots);
extern int SimpleLruAutotuneBuffers(int divisor, int max);
-extern void SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
+extern void SimpleLruInit(SlruCtl ctl, const char *name, int nslots,
const char *subdir, int buffer_tranche_id,
int bank_tranche_id, SyncRequestHandler sync_handler,
bool long_segment_names);
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index bedc2a0d72..854fb01281 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -57,6 +57,6 @@
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 202403052
+#define CATALOG_VERSION_NO 202403061
#endif
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index d0df02d39c..2dc83451a6 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -201,6 +201,7 @@ typedef PageHeaderData *PageHeader;
* handling pages.
*/
#define PG_PAGE_LAYOUT_VERSION 4
+#define PG_SLRU_PAGE_LAYOUT_VERSION 1
#define PG_DATA_CHECKSUM_VERSION 1
/* ----------------------------------------------------------------
@@ -257,6 +258,11 @@ PageGetContents(Page page)
return (char *) page + MAXALIGN(SizeOfPageHeaderData);
}
+/*
+ * Space available for storing page contents.
+ */
+#define SizeOfPageContents (BLCKSZ - MAXALIGN(SizeOfPageHeaderData))
+
/* ----------------
* functions to access page size info
* ----------------
@@ -486,6 +492,7 @@ StaticAssertDecl(BLCKSZ == ((BLCKSZ / sizeof(size_t)) * sizeof(size_t)),
"BLCKSZ has to be a multiple of sizeof(size_t)");
extern void PageInit(Page page, Size pageSize, Size specialSize);
+extern void PageInitSLRU(Page page, Size pageSize, Size specialSize);
extern bool PageIsVerifiedExtended(Page page, BlockNumber blkno, int flags);
extern OffsetNumber PageAddItemExtended(Page page, Item item, Size size,
OffsetNumber offsetNumber, int flags);
diff --git a/src/test/modules/test_slru/test_slru.c b/src/test/modules/test_slru/test_slru.c
index 068a21f125..026aa79912 100644
--- a/src/test/modules/test_slru/test_slru.c
+++ b/src/test/modules/test_slru/test_slru.c
@@ -17,6 +17,7 @@
#include "access/slru.h"
#include "access/transam.h"
#include "miscadmin.h"
+#include "storage/bufpage.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/shmem.h"
@@ -72,8 +73,8 @@ test_slru_page_write(PG_FUNCTION_ARGS)
TestSlruCtl->shared->page_status[slotno] = SLRU_PAGE_VALID;
/* write given data to the page, up to the limit of the page */
- strncpy(TestSlruCtl->shared->page_buffer[slotno], data,
- BLCKSZ - 1);
+ strncpy(PageGetContents(TestSlruCtl->shared->page_buffer[slotno]), data,
+ SizeOfPageContents - 1);
SimpleLruWritePage(TestSlruCtl, slotno);
LWLockRelease(lock);
@@ -101,7 +102,7 @@ test_slru_page_read(PG_FUNCTION_ARGS)
LWLockAcquire(lock, LW_EXCLUSIVE);
slotno = SimpleLruReadPage(TestSlruCtl, pageno,
write_ok, InvalidTransactionId);
- data = (char *) TestSlruCtl->shared->page_buffer[slotno];
+ data = (char *) PageGetContents(TestSlruCtl->shared->page_buffer[slotno]);
LWLockRelease(lock);
PG_RETURN_TEXT_P(cstring_to_text(data));
@@ -120,7 +121,7 @@ test_slru_page_readonly(PG_FUNCTION_ARGS)
pageno,
InvalidTransactionId);
Assert(LWLockHeldByMe(lock));
- data = (char *) TestSlruCtl->shared->page_buffer[slotno];
+ data = (char *) PageGetContents(TestSlruCtl->shared->page_buffer[slotno]);
LWLockRelease(lock);
PG_RETURN_TEXT_P(cstring_to_text(data));
@@ -201,7 +202,7 @@ test_slru_shmem_request(void)
prev_shmem_request_hook();
/* reserve shared memory for the test SLRU */
- RequestAddinShmemSpace(SimpleLruShmemSize(NUM_TEST_BUFFERS, 0));
+ RequestAddinShmemSpace(SimpleLruShmemSize(NUM_TEST_BUFFERS));
}
static bool
@@ -240,7 +241,7 @@ test_slru_shmem_startup(void)
TestSlruCtl->PagePrecedes = test_slru_page_precedes_logically;
SimpleLruInit(TestSlruCtl, "TestSLRU",
- NUM_TEST_BUFFERS, 0, slru_dir_name,
+ NUM_TEST_BUFFERS, slru_dir_name,
test_buffer_tranche_id, test_tranche_id, SYNC_HANDLER_NONE,
long_segment_names);
}
view thread (22+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Proposal to add page headers to SLRU pages
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox