From 35e76b7255b0d621a2c55fadfa3996dccbf7ff17 Mon Sep 17 00:00:00 2001 From: Rishu Bagga Date: Thu, 26 Jun 2025 19:03:10 -0700 Subject: [PATCH] Add page headers to SLRU pages --- src/backend/access/transam/clog.c | 45 +- src/backend/access/transam/commit_ts.c | 1139 +++++++++++------------- src/backend/access/transam/multixact.c | 55 +- src/backend/access/transam/slru.c | 71 +- src/backend/access/transam/subtrans.c | 8 +- src/backend/commands/async.c | 23 +- src/backend/storage/lmgr/predicate.c | 12 +- src/backend/storage/page/bufpage.c | 26 + src/bin/pg_checksums/pg_checksums.c | 9 + src/bin/pg_resetwal/t/001_basic.pl | 6 +- src/bin/pg_upgrade/file.c | 178 ++++ src/bin/pg_upgrade/pg_upgrade.c | 28 +- src/bin/pg_upgrade/pg_upgrade.h | 6 + src/include/catalog/catversion.h | 2 +- src/include/storage/bufpage.h | 7 + src/test/modules/test_slru/test_slru.c | 9 +- 16 files changed, 915 insertions(+), 709 deletions(-) diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index 48f10bec91e..9b4ba0861a9 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) /* @@ -90,7 +91,13 @@ TransactionIdToPage(TransactionId xid) /* 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) + +/* + * Use BLCKSZ instead of SizeOfPageContents so that CLOG_LSNS_PER_PAGE is + * a power of 2. Using BLCKSZ wastes the last 4 LSN groups per page, but + * this is acceptable given that each page has 1,024 LSN groups. + */ +#define CLOG_LSNS_PER_PAGE ((BLCKSZ * CLOG_XACTS_PER_BYTE) / 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) @@ -112,7 +119,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, @@ -665,13 +672,15 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i char *byteptr; char byteval; char curval; + Page page; Assert(XactCtl->shared->page_number[slotno] == TransactionIdToPage(xid)); Assert(LWLockHeldByMeInMode(SimpleLruGetBankLock(XactCtl, XactCtl->shared->page_number[slotno]), LW_EXCLUSIVE)); - byteptr = XactCtl->shared->page_buffer[slotno] + byteno; + page = XactCtl->shared->page_buffer[slotno]; + byteptr = PageGetContents(page) + byteno; curval = (*byteptr >> bshift) & CLOG_XACT_BITMASK; /* @@ -700,7 +709,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 & group 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 +718,13 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i */ if (!XLogRecPtrIsInvalid(lsn)) { - int lsnindex = GetLSNIndex(slotno, xid); + 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); } } @@ -739,13 +751,15 @@ TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn) 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; @@ -860,11 +874,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 +937,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 +966,6 @@ CheckPointCLOG(void) TRACE_POSTGRESQL_CLOG_CHECKPOINT_DONE(true); } - /* * Make sure that CLOG has room for a newly-allocated XID. * @@ -1070,12 +1089,12 @@ CLOGPagePrecedes(int64 page1, int64 page2) /* * Write a ZEROPAGE xlog record */ -static void +static XLogRecPtr WriteZeroPageXlogRec(int64 pageno) { XLogBeginInsert(); XLogRegisterData(&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 113fae1437a..4fe73e3be19 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" @@ -51,31 +52,27 @@ * the largest possible file name is more than 5 chars long; see * SlruScanDirectory. */ -typedef struct CommitTimestampEntry -{ - TimestampTz time; - RepOriginId nodeid; +typedef struct CommitTimestampEntry { + TimestampTz time; + RepOriginId nodeid; } CommitTimestampEntry; -#define SizeOfCommitTimestampEntry (offsetof(CommitTimestampEntry, nodeid) + \ - sizeof(RepOriginId)) - -#define COMMIT_TS_XACTS_PER_PAGE \ - (BLCKSZ / SizeOfCommitTimestampEntry) +#define SizeOfCommitTimestampEntry \ + (offsetof(CommitTimestampEntry, nodeid) + sizeof(RepOriginId)) +#define COMMIT_TS_XACTS_PER_PAGE \ + (SizeOfPageContents / SizeOfCommitTimestampEntry) /* * 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; +static inline int64 TransactionIdToCTsPage(TransactionId xid) { + return xid / (int64)COMMIT_TS_XACTS_PER_PAGE; } -#define TransactionIdToCTsEntry(xid) \ - ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE) +#define TransactionIdToCTsEntry(xid) \ + ((xid) % (TransactionId)COMMIT_TS_XACTS_PER_PAGE) /* * Link to shared-memory data structures for CommitTs control @@ -95,30 +92,28 @@ static SlruCtlData CommitTsCtlData; * without acquiring the lock; where this happens, a comment explains the * rationale for it. */ -typedef struct CommitTimestampShared -{ - TransactionId xidLastCommit; - CommitTimestampEntry dataLastCommit; - bool commitTsActive; +typedef struct CommitTimestampShared { + TransactionId xidLastCommit; + CommitTimestampEntry dataLastCommit; + bool commitTsActive; } CommitTimestampShared; static CommitTimestampShared *commitTsShared; - /* GUC variable */ -bool track_commit_timestamp; +bool track_commit_timestamp; static void SetXidCommitTsInPage(TransactionId xid, int nsubxids, - TransactionId *subxids, TimestampTz ts, - RepOriginId nodeid, int64 pageno); + TransactionId *subxids, TimestampTz ts, + RepOriginId nodeid, int64 pageno); static void TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, - RepOriginId nodeid, int slotno); + RepOriginId nodeid, int slotno); static void error_commit_ts_disabled(void); -static int ZeroCommitTsPage(int64 pageno, bool writeXlog); +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); /* @@ -137,107 +132,101 @@ static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid); * subtrans implementation changes in the future, we might want to revisit the * decision of storing timestamp info for each subxid. */ -void -TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids, - TransactionId *subxids, TimestampTz timestamp, - RepOriginId nodeid) -{ - int i; - TransactionId headxid; - TransactionId newestXact; - - /* - * No-op if the module is not active. - * - * An unlocked read here is fine, because in a standby (the only place - * where the flag can change in flight) this routine is only called by the - * recovery process, which is also the only process which can change the - * flag. - */ - if (!commitTsShared->commitTsActive) - return; - - /* - * Figure out the latest Xid in this batch: either the last subxid if - * there's any, otherwise the parent xid. - */ - if (nsubxids > 0) - newestXact = subxids[nsubxids - 1]; - else - newestXact = xid; - - /* - * We split the xids to set the timestamp to in groups belonging to the - * same SLRU page; the first element in each such set is its head. The - * first group has the main XID as the head; subsequent sets use the first - * subxid not on the previous page as head. This way, we only have to - * lock/modify each SLRU page once. - */ - headxid = xid; - i = 0; - for (;;) - { - int64 pageno = TransactionIdToCTsPage(headxid); - int j; - - for (j = i; j < nsubxids; j++) - { - if (TransactionIdToCTsPage(subxids[j]) != pageno) - break; - } - /* subxids[i..j] are on the same page as the head */ - - SetXidCommitTsInPage(headxid, j - i, subxids + i, timestamp, nodeid, - pageno); - - /* if we wrote out all subxids, we're done. */ - if (j >= nsubxids) - break; - - /* - * Set the new head and skip over it, as well as over the subxids we - * just wrote. - */ - headxid = subxids[j]; - i = j + 1; - } - - /* update the cached value in shared memory */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - commitTsShared->xidLastCommit = xid; - commitTsShared->dataLastCommit.time = timestamp; - commitTsShared->dataLastCommit.nodeid = nodeid; - - /* and move forwards our endpoint, if needed */ - if (TransactionIdPrecedes(TransamVariables->newestCommitTsXid, newestXact)) - TransamVariables->newestCommitTsXid = newestXact; - LWLockRelease(CommitTsLock); +void TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids, + TransactionId *subxids, + TimestampTz timestamp, RepOriginId nodeid) { + int i; + TransactionId headxid; + TransactionId newestXact; + + /* + * No-op if the module is not active. + * + * An unlocked read here is fine, because in a standby (the only place + * where the flag can change in flight) this routine is only called by the + * recovery process, which is also the only process which can change the + * flag. + */ + if (!commitTsShared->commitTsActive) + return; + + /* + * Figure out the latest Xid in this batch: either the last subxid if + * there's any, otherwise the parent xid. + */ + if (nsubxids > 0) + newestXact = subxids[nsubxids - 1]; + else + newestXact = xid; + + /* + * We split the xids to set the timestamp to in groups belonging to the + * same SLRU page; the first element in each such set is its head. The + * first group has the main XID as the head; subsequent sets use the first + * subxid not on the previous page as head. This way, we only have to + * lock/modify each SLRU page once. + */ + headxid = xid; + i = 0; + for (;;) { + int64 pageno = TransactionIdToCTsPage(headxid); + int j; + + for (j = i; j < nsubxids; j++) { + if (TransactionIdToCTsPage(subxids[j]) != pageno) + break; + } + /* subxids[i..j] are on the same page as the head */ + + SetXidCommitTsInPage(headxid, j - i, subxids + i, timestamp, nodeid, + pageno); + + /* if we wrote out all subxids, we're done. */ + if (j >= nsubxids) + break; + + /* + * Set the new head and skip over it, as well as over the subxids we + * just wrote. + */ + headxid = subxids[j]; + i = j + 1; + } + + /* update the cached value in shared memory */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + commitTsShared->xidLastCommit = xid; + commitTsShared->dataLastCommit.time = timestamp; + commitTsShared->dataLastCommit.nodeid = nodeid; + + /* and move forwards our endpoint, if needed */ + if (TransactionIdPrecedes(TransamVariables->newestCommitTsXid, newestXact)) + TransamVariables->newestCommitTsXid = newestXact; + LWLockRelease(CommitTsLock); } /* * Record the commit timestamp of transaction entries in the commit log for all * entries on a single page. Atomic only on this page. */ -static void -SetXidCommitTsInPage(TransactionId xid, int nsubxids, - TransactionId *subxids, TimestampTz ts, - RepOriginId nodeid, int64 pageno) -{ - LWLock *lock = SimpleLruGetBankLock(CommitTsCtl, pageno); - int slotno; - int i; +static void SetXidCommitTsInPage(TransactionId xid, int nsubxids, + TransactionId *subxids, TimestampTz ts, + RepOriginId nodeid, int64 pageno) { + LWLock *lock = SimpleLruGetBankLock(CommitTsCtl, pageno); + int slotno; + int i; - LWLockAcquire(lock, LW_EXCLUSIVE); + LWLockAcquire(lock, LW_EXCLUSIVE); - slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, xid); + slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, xid); - TransactionIdSetCommitTs(xid, ts, nodeid, slotno); - for (i = 0; i < nsubxids; i++) - TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno); + TransactionIdSetCommitTs(xid, ts, nodeid, slotno); + for (i = 0; i < nsubxids; i++) + TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno); - CommitTsCtl->shared->page_dirty[slotno] = true; + CommitTsCtl->shared->page_dirty[slotno] = true; - LWLockRelease(lock); + LWLockRelease(lock); } /* @@ -245,21 +234,19 @@ SetXidCommitTsInPage(TransactionId xid, int nsubxids, * * Caller must hold the correct SLRU bank lock, will be held at exit */ -static void -TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, - RepOriginId nodeid, int slotno) -{ - int entryno = TransactionIdToCTsEntry(xid); - CommitTimestampEntry entry; +static void TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, + RepOriginId nodeid, int slotno) { + int entryno = TransactionIdToCTsEntry(xid); + CommitTimestampEntry entry; - Assert(TransactionIdIsNormal(xid)); + Assert(TransactionIdIsNormal(xid)); - entry.time = ts; - entry.nodeid = nodeid; + entry.time = ts; + entry.nodeid = nodeid; - memcpy(CommitTsCtl->shared->page_buffer[slotno] + - SizeOfCommitTimestampEntry * entryno, - &entry, SizeOfCommitTimestampEntry); + memcpy(PageGetContents(CommitTsCtl->shared->page_buffer[slotno]) + + SizeOfCommitTimestampEntry * entryno, + &entry, SizeOfCommitTimestampEntry); } /* @@ -270,82 +257,79 @@ TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, * null), and the origin node for the Xid is returned in *nodeid, if it's not * null. */ -bool -TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts, - RepOriginId *nodeid) -{ - int64 pageno = TransactionIdToCTsPage(xid); - int entryno = TransactionIdToCTsEntry(xid); - int slotno; - CommitTimestampEntry entry; - TransactionId oldestCommitTsXid; - TransactionId newestCommitTsXid; - - if (!TransactionIdIsValid(xid)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("cannot retrieve commit timestamp for transaction %u", xid))); - else if (!TransactionIdIsNormal(xid)) - { - /* frozen and bootstrap xids are always committed far in the past */ - *ts = 0; - if (nodeid) - *nodeid = 0; - return false; - } - - LWLockAcquire(CommitTsLock, LW_SHARED); - - /* Error if module not enabled */ - if (!commitTsShared->commitTsActive) - error_commit_ts_disabled(); - - /* - * If we're asked for the cached value, return that. Otherwise, fall - * through to read from SLRU. - */ - if (commitTsShared->xidLastCommit == xid) - { - *ts = commitTsShared->dataLastCommit.time; - if (nodeid) - *nodeid = commitTsShared->dataLastCommit.nodeid; - - LWLockRelease(CommitTsLock); - return *ts != 0; - } - - oldestCommitTsXid = TransamVariables->oldestCommitTsXid; - newestCommitTsXid = TransamVariables->newestCommitTsXid; - /* neither is invalid, or both are */ - Assert(TransactionIdIsValid(oldestCommitTsXid) == TransactionIdIsValid(newestCommitTsXid)); - LWLockRelease(CommitTsLock); - - /* - * Return empty if the requested value is outside our valid range. - */ - if (!TransactionIdIsValid(oldestCommitTsXid) || - TransactionIdPrecedes(xid, oldestCommitTsXid) || - TransactionIdPrecedes(newestCommitTsXid, xid)) - { - *ts = 0; - if (nodeid) - *nodeid = InvalidRepOriginId; - return false; - } - - /* lock is acquired by SimpleLruReadPage_ReadOnly */ - slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, xid); - memcpy(&entry, - CommitTsCtl->shared->page_buffer[slotno] + - SizeOfCommitTimestampEntry * entryno, - SizeOfCommitTimestampEntry); - - *ts = entry.time; - if (nodeid) - *nodeid = entry.nodeid; - - LWLockRelease(SimpleLruGetBankLock(CommitTsCtl, pageno)); - return *ts != 0; +bool TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts, + RepOriginId *nodeid) { + int64 pageno = TransactionIdToCTsPage(xid); + int entryno = TransactionIdToCTsEntry(xid); + int slotno; + CommitTimestampEntry entry; + TransactionId oldestCommitTsXid; + TransactionId newestCommitTsXid; + + if (!TransactionIdIsValid(xid)) + ereport( + ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot retrieve commit timestamp for transaction %u", xid))); + else if (!TransactionIdIsNormal(xid)) { + /* frozen and bootstrap xids are always committed far in the past */ + *ts = 0; + if (nodeid) + *nodeid = 0; + return false; + } + + LWLockAcquire(CommitTsLock, LW_SHARED); + + /* Error if module not enabled */ + if (!commitTsShared->commitTsActive) + error_commit_ts_disabled(); + + /* + * If we're asked for the cached value, return that. Otherwise, fall + * through to read from SLRU. + */ + if (commitTsShared->xidLastCommit == xid) { + *ts = commitTsShared->dataLastCommit.time; + if (nodeid) + *nodeid = commitTsShared->dataLastCommit.nodeid; + + LWLockRelease(CommitTsLock); + return *ts != 0; + } + + oldestCommitTsXid = TransamVariables->oldestCommitTsXid; + newestCommitTsXid = TransamVariables->newestCommitTsXid; + /* neither is invalid, or both are */ + Assert(TransactionIdIsValid(oldestCommitTsXid) == + TransactionIdIsValid(newestCommitTsXid)); + LWLockRelease(CommitTsLock); + + /* + * Return empty if the requested value is outside our valid range. + */ + if (!TransactionIdIsValid(oldestCommitTsXid) || + TransactionIdPrecedes(xid, oldestCommitTsXid) || + TransactionIdPrecedes(newestCommitTsXid, xid)) { + *ts = 0; + if (nodeid) + *nodeid = InvalidRepOriginId; + return false; + } + + /* lock is acquired by SimpleLruReadPage_ReadOnly */ + slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, xid); + memcpy(&entry, + PageGetContents(CommitTsCtl->shared->page_buffer[slotno]) + + SizeOfCommitTimestampEntry * entryno, + SizeOfCommitTimestampEntry); + + *ts = entry.time; + if (nodeid) + *nodeid = entry.nodeid; + + LWLockRelease(SimpleLruGetBankLock(CommitTsCtl, pageno)); + return *ts != 0; } /* @@ -356,59 +340,53 @@ TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts, * ts and nodeid are filled with the corresponding data; they can be passed * as NULL if not wanted. */ -TransactionId -GetLatestCommitTsData(TimestampTz *ts, RepOriginId *nodeid) -{ - TransactionId xid; +TransactionId GetLatestCommitTsData(TimestampTz *ts, RepOriginId *nodeid) { + TransactionId xid; - LWLockAcquire(CommitTsLock, LW_SHARED); + LWLockAcquire(CommitTsLock, LW_SHARED); - /* Error if module not enabled */ - if (!commitTsShared->commitTsActive) - error_commit_ts_disabled(); + /* Error if module not enabled */ + if (!commitTsShared->commitTsActive) + error_commit_ts_disabled(); - xid = commitTsShared->xidLastCommit; - if (ts) - *ts = commitTsShared->dataLastCommit.time; - if (nodeid) - *nodeid = commitTsShared->dataLastCommit.nodeid; - LWLockRelease(CommitTsLock); + xid = commitTsShared->xidLastCommit; + if (ts) + *ts = commitTsShared->dataLastCommit.time; + if (nodeid) + *nodeid = commitTsShared->dataLastCommit.nodeid; + LWLockRelease(CommitTsLock); - return xid; + return xid; } -static void -error_commit_ts_disabled(void) -{ - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("could not get commit timestamp data"), - RecoveryInProgress() ? - errhint("Make sure the configuration parameter \"%s\" is set on the primary server.", - "track_commit_timestamp") : - errhint("Make sure the configuration parameter \"%s\" is set.", - "track_commit_timestamp"))); +static void error_commit_ts_disabled(void) { + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not get commit timestamp data"), + RecoveryInProgress() + ? errhint("Make sure the configuration parameter \"%s\" is set " + "on the primary server.", + "track_commit_timestamp") + : errhint("Make sure the configuration parameter \"%s\" is set.", + "track_commit_timestamp"))); } /* * SQL-callable wrapper to obtain commit time of a transaction */ -Datum -pg_xact_commit_timestamp(PG_FUNCTION_ARGS) -{ - TransactionId xid = PG_GETARG_TRANSACTIONID(0); - TimestampTz ts; - bool found; +Datum pg_xact_commit_timestamp(PG_FUNCTION_ARGS) { + TransactionId xid = PG_GETARG_TRANSACTIONID(0); + TimestampTz ts; + bool found; - found = TransactionIdGetCommitTsData(xid, &ts, NULL); + found = TransactionIdGetCommitTsData(xid, &ts, NULL); - if (!found) - PG_RETURN_NULL(); + if (!found) + PG_RETURN_NULL(); - PG_RETURN_TIMESTAMPTZ(ts); + PG_RETURN_TIMESTAMPTZ(ts); } - /* * pg_last_committed_xact * @@ -416,42 +394,37 @@ pg_xact_commit_timestamp(PG_FUNCTION_ARGS) * committed transaction: transaction ID, timestamp and replication * origin. */ -Datum -pg_last_committed_xact(PG_FUNCTION_ARGS) -{ - TransactionId xid; - RepOriginId nodeid; - TimestampTz ts; - Datum values[3]; - bool nulls[3]; - TupleDesc tupdesc; - HeapTuple htup; +Datum pg_last_committed_xact(PG_FUNCTION_ARGS) { + TransactionId xid; + RepOriginId nodeid; + TimestampTz ts; + Datum values[3]; + bool nulls[3]; + TupleDesc tupdesc; + HeapTuple htup; - /* and construct a tuple with our data */ - xid = GetLatestCommitTsData(&ts, &nodeid); + /* and construct a tuple with our data */ + xid = GetLatestCommitTsData(&ts, &nodeid); - if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) - elog(ERROR, "return type must be a row type"); + if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) + elog(ERROR, "return type must be a row type"); - if (!TransactionIdIsNormal(xid)) - { - memset(nulls, true, sizeof(nulls)); - } - else - { - values[0] = TransactionIdGetDatum(xid); - nulls[0] = false; + if (!TransactionIdIsNormal(xid)) { + memset(nulls, true, sizeof(nulls)); + } else { + values[0] = TransactionIdGetDatum(xid); + nulls[0] = false; - values[1] = TimestampTzGetDatum(ts); - nulls[1] = false; + values[1] = TimestampTzGetDatum(ts); + nulls[1] = false; - values[2] = ObjectIdGetDatum((Oid) nodeid); - nulls[2] = false; - } + values[2] = ObjectIdGetDatum((Oid)nodeid); + nulls[2] = false; + } - htup = heap_form_tuple(tupdesc, values, nulls); + htup = heap_form_tuple(tupdesc, values, nulls); - PG_RETURN_DATUM(HeapTupleGetDatum(htup)); + PG_RETURN_DATUM(HeapTupleGetDatum(htup)); } /* @@ -460,39 +433,34 @@ pg_last_committed_xact(PG_FUNCTION_ARGS) * SQL-callable wrapper to obtain commit timestamp and replication origin * of a given transaction. */ -Datum -pg_xact_commit_timestamp_origin(PG_FUNCTION_ARGS) -{ - TransactionId xid = PG_GETARG_TRANSACTIONID(0); - RepOriginId nodeid; - TimestampTz ts; - Datum values[2]; - bool nulls[2]; - TupleDesc tupdesc; - HeapTuple htup; - bool found; +Datum pg_xact_commit_timestamp_origin(PG_FUNCTION_ARGS) { + TransactionId xid = PG_GETARG_TRANSACTIONID(0); + RepOriginId nodeid; + TimestampTz ts; + Datum values[2]; + bool nulls[2]; + TupleDesc tupdesc; + HeapTuple htup; + bool found; - found = TransactionIdGetCommitTsData(xid, &ts, &nodeid); + found = TransactionIdGetCommitTsData(xid, &ts, &nodeid); - if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) - elog(ERROR, "return type must be a row type"); + if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) + elog(ERROR, "return type must be a row type"); - if (!found) - { - memset(nulls, true, sizeof(nulls)); - } - else - { - values[0] = TimestampTzGetDatum(ts); - nulls[0] = false; + if (!found) { + memset(nulls, true, sizeof(nulls)); + } else { + values[0] = TimestampTzGetDatum(ts); + nulls[0] = false; - values[1] = ObjectIdGetDatum((Oid) nodeid); - nulls[1] = false; - } + values[1] = ObjectIdGetDatum((Oid)nodeid); + nulls[1] = false; + } - htup = heap_form_tuple(tupdesc, values, nulls); + htup = heap_form_tuple(tupdesc, values, nulls); - PG_RETURN_DATUM(HeapTupleGetDatum(htup)); + PG_RETURN_DATUM(HeapTupleGetDatum(htup)); } /* @@ -502,88 +470,74 @@ pg_xact_commit_timestamp_origin(PG_FUNCTION_ARGS) * Otherwise just cap the configured amount to be between 16 and the maximum * allowed. */ -static int -CommitTsShmemBuffers(void) -{ - /* auto-tune based on shared buffers */ - if (commit_timestamp_buffers == 0) - return SimpleLruAutotuneBuffers(512, 1024); +static int CommitTsShmemBuffers(void) { + /* auto-tune based on shared buffers */ + if (commit_timestamp_buffers == 0) + return SimpleLruAutotuneBuffers(512, 1024); - return Min(Max(16, commit_timestamp_buffers), SLRU_MAX_ALLOWED_BUFFERS); + return Min(Max(16, commit_timestamp_buffers), SLRU_MAX_ALLOWED_BUFFERS); } /* * Shared memory sizing for CommitTs */ -Size -CommitTsShmemSize(void) -{ - return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) + - sizeof(CommitTimestampShared); +Size CommitTsShmemSize(void) { + return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) + + sizeof(CommitTimestampShared); } /* * Initialize CommitTs at system startup (postmaster start or standalone * backend) */ -void -CommitTsShmemInit(void) -{ - bool found; - - /* If auto-tuning is requested, now is the time to do it */ - if (commit_timestamp_buffers == 0) - { - char buf[32]; - - snprintf(buf, sizeof(buf), "%d", CommitTsShmemBuffers()); - SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER, - PGC_S_DYNAMIC_DEFAULT); - - /* - * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT. - * However, if the DBA explicitly set commit_timestamp_buffers = 0 in - * the config file, then PGC_S_DYNAMIC_DEFAULT will fail to override - * that and we must force the matter with PGC_S_OVERRIDE. - */ - if (commit_timestamp_buffers == 0) /* failed to apply it? */ - SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER, - PGC_S_OVERRIDE); - } - Assert(commit_timestamp_buffers != 0); - - CommitTsCtl->PagePrecedes = CommitTsPagePrecedes; - SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0, - "pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER, - LWTRANCHE_COMMITTS_SLRU, - SYNC_HANDLER_COMMIT_TS, - false); - SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE); - - commitTsShared = ShmemInitStruct("CommitTs shared", - sizeof(CommitTimestampShared), - &found); - - if (!IsUnderPostmaster) - { - Assert(!found); - - commitTsShared->xidLastCommit = InvalidTransactionId; - TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time); - commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId; - commitTsShared->commitTsActive = false; - } - else - Assert(found); +void CommitTsShmemInit(void) { + bool found; + + /* If auto-tuning is requested, now is the time to do it */ + if (commit_timestamp_buffers == 0) { + char buf[32]; + + snprintf(buf, sizeof(buf), "%d", CommitTsShmemBuffers()); + SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER, + PGC_S_DYNAMIC_DEFAULT); + + /* + * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT. + * However, if the DBA explicitly set commit_timestamp_buffers = 0 in + * the config file, then PGC_S_DYNAMIC_DEFAULT will fail to override + * that and we must force the matter with PGC_S_OVERRIDE. + */ + if (commit_timestamp_buffers == 0) /* failed to apply it? */ + SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER, + PGC_S_OVERRIDE); + } + Assert(commit_timestamp_buffers != 0); + + CommitTsCtl->PagePrecedes = CommitTsPagePrecedes; + SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0, + "pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER, + LWTRANCHE_COMMITTS_SLRU, SYNC_HANDLER_COMMIT_TS, false); + SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE); + + commitTsShared = + ShmemInitStruct("CommitTs shared", sizeof(CommitTimestampShared), &found); + + if (!IsUnderPostmaster) { + Assert(!found); + + commitTsShared->xidLastCommit = InvalidTransactionId; + TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time); + commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId; + commitTsShared->commitTsActive = false; + } else + Assert(found); } /* * GUC check_hook for commit_timestamp_buffers */ -bool -check_commit_ts_buffers(int *newval, void **extra, GucSource source) -{ - return check_slru_buffers("commit_timestamp_buffers", newval); +bool check_commit_ts_buffers(int *newval, void **extra, GucSource source) { + return check_slru_buffers("commit_timestamp_buffers", newval); } /* @@ -592,14 +546,12 @@ check_commit_ts_buffers(int *newval, void **extra, GucSource source) * (The CommitTs directory is assumed to have been created by initdb, and * CommitTsShmemInit must have been called already.) */ -void -BootStrapCommitTs(void) -{ - /* - * Nothing to do here at present, unlike most other SLRU modules; segments - * are created when the server is started with this module enabled. See - * ActivateCommitTs. - */ +void BootStrapCommitTs(void) { + /* + * Nothing to do here at present, unlike most other SLRU modules; segments + * are created when the server is started with this module enabled. See + * ActivateCommitTs. + */ } /* @@ -611,84 +563,77 @@ BootStrapCommitTs(void) * * Control lock must be held at entry, and will be held at exit. */ -static int -ZeroCommitTsPage(int64 pageno, bool writeXlog) -{ - int slotno; +static int ZeroCommitTsPage(int64 pageno, bool writeXlog) { + int slotno; + Page page; + XLogRecPtr lsn = 0; - slotno = SimpleLruZeroPage(CommitTsCtl, pageno); + slotno = SimpleLruZeroPage(CommitTsCtl, pageno); + page = CommitTsCtl->shared->page_buffer[slotno]; - if (writeXlog) - WriteZeroPageXlogRec(pageno); + if (writeXlog) { + lsn = WriteZeroPageXlogRec(pageno); + PageSetLSN(page, lsn); + } - return slotno; + return slotno; } /* * This must be called ONCE during postmaster or standalone-backend startup, * after StartupXLOG has initialized TransamVariables->nextXid. */ -void -StartupCommitTs(void) -{ - ActivateCommitTs(); -} +void StartupCommitTs(void) { ActivateCommitTs(); } /* * This must be called ONCE during postmaster or standalone-backend startup, * after recovery has finished. */ -void -CompleteCommitTsInitialization(void) -{ - /* - * If the feature is not enabled, turn it off for good. This also removes - * any leftover data. - * - * Conversely, we activate the module if the feature is enabled. This is - * necessary for primary and standby as the activation depends on the - * control file contents at the beginning of recovery or when a - * XLOG_PARAMETER_CHANGE is replayed. - */ - if (!track_commit_timestamp) - DeactivateCommitTs(); - else - ActivateCommitTs(); +void CompleteCommitTsInitialization(void) { + /* + * If the feature is not enabled, turn it off for good. This also removes + * any leftover data. + * + * Conversely, we activate the module if the feature is enabled. This is + * necessary for primary and standby as the activation depends on the + * control file contents at the beginning of recovery or when a + * XLOG_PARAMETER_CHANGE is replayed. + */ + if (!track_commit_timestamp) + DeactivateCommitTs(); + else + ActivateCommitTs(); } /* * Activate or deactivate CommitTs' upon reception of a XLOG_PARAMETER_CHANGE * XLog record during recovery. */ -void -CommitTsParameterChange(bool newvalue, bool oldvalue) -{ - /* - * If the commit_ts module is disabled in this server and we get word from - * the primary server that it is enabled there, activate it so that we can - * replay future WAL records involving it; also mark it as active on - * pg_control. If the old value was already set, we already did this, so - * don't do anything. - * - * If the module is disabled in the primary, disable it here too, unless - * the module is enabled locally. - * - * Note this only runs in the recovery process, so an unlocked read is - * fine. - */ - if (newvalue) - { - if (!commitTsShared->commitTsActive) - ActivateCommitTs(); - } - else if (commitTsShared->commitTsActive) - DeactivateCommitTs(); +void CommitTsParameterChange(bool newvalue, bool oldvalue) { + /* + * If the commit_ts module is disabled in this server and we get word from + * the primary server that it is enabled there, activate it so that we can + * replay future WAL records involving it; also mark it as active on + * pg_control. If the old value was already set, we already did this, so + * don't do anything. + * + * If the module is disabled in the primary, disable it here too, unless + * the module is enabled locally. + * + * Note this only runs in the recovery process, so an unlocked read is + * fine. + */ + if (newvalue) { + if (!commitTsShared->commitTsActive) + ActivateCommitTs(); + } else if (commitTsShared->commitTsActive) + DeactivateCommitTs(); } /* * Activate this module whenever necessary. - * This must happen during postmaster or standalone-backend startup, - * or during WAL replay anytime the track_commit_timestamp setting is + * This must happen during postmaster or standalone-backend + *startup, or during WAL replay anytime the track_commit_timestamp setting is * changed in the primary. * * The reason why this SLRU needs separate activation/deactivation functions is @@ -701,67 +646,62 @@ CommitTsParameterChange(bool newvalue, bool oldvalue) * running with this module disabled for a while and thus might have skipped * the normal creation point. */ -static void -ActivateCommitTs(void) -{ - TransactionId xid; - int64 pageno; - - /* If we've done this already, there's nothing to do */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - if (commitTsShared->commitTsActive) - { - LWLockRelease(CommitTsLock); - return; - } - LWLockRelease(CommitTsLock); - - xid = XidFromFullTransactionId(TransamVariables->nextXid); - pageno = TransactionIdToCTsPage(xid); - - /* - * Re-Initialize our idea of the latest page number. - */ - pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number, pageno); - - /* - * If CommitTs is enabled, but it wasn't in the previous server run, we - * need to set the oldest and newest values to the next Xid; that way, we - * will not try to read data that might not have been set. - * - * XXX does this have a problem if a server is started with commitTs - * enabled, then started with commitTs disabled, then restarted with it - * enabled again? It doesn't look like it does, because there should be a - * checkpoint that sets the value to InvalidTransactionId at end of - * recovery; and so any chance of injecting new transactions without - * CommitTs values would occur after the oldestCommitTsXid has been set to - * Invalid temporarily. - */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - if (TransamVariables->oldestCommitTsXid == InvalidTransactionId) - { - TransamVariables->oldestCommitTsXid = - TransamVariables->newestCommitTsXid = ReadNextTransactionId(); - } - LWLockRelease(CommitTsLock); - - /* Create the current segment file, if necessary */ - if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno)) - { - LWLock *lock = SimpleLruGetBankLock(CommitTsCtl, pageno); - int slotno; - - LWLockAcquire(lock, LW_EXCLUSIVE); - slotno = ZeroCommitTsPage(pageno, false); - SimpleLruWritePage(CommitTsCtl, slotno); - Assert(!CommitTsCtl->shared->page_dirty[slotno]); - LWLockRelease(lock); - } - - /* Change the activation status in shared memory. */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - commitTsShared->commitTsActive = true; - LWLockRelease(CommitTsLock); +static void ActivateCommitTs(void) { + TransactionId xid; + int64 pageno; + + /* If we've done this already, there's nothing to do */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + if (commitTsShared->commitTsActive) { + LWLockRelease(CommitTsLock); + return; + } + LWLockRelease(CommitTsLock); + + xid = XidFromFullTransactionId(TransamVariables->nextXid); + pageno = TransactionIdToCTsPage(xid); + + /* + * Re-Initialize our idea of the latest page number. + */ + pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number, pageno); + + /* + * If CommitTs is enabled, but it wasn't in the previous server run, we + * need to set the oldest and newest values to the next Xid; that way, we + * will not try to read data that might not have been set. + * + * XXX does this have a problem if a server is started with commitTs + * enabled, then started with commitTs disabled, then restarted with it + * enabled again? It doesn't look like it does, because there should be a + * checkpoint that sets the value to InvalidTransactionId at end of + * recovery; and so any chance of injecting new transactions without + * CommitTs values would occur after the oldestCommitTsXid has been set to + * Invalid temporarily. + */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + if (TransamVariables->oldestCommitTsXid == InvalidTransactionId) { + TransamVariables->oldestCommitTsXid = TransamVariables->newestCommitTsXid = + ReadNextTransactionId(); + } + LWLockRelease(CommitTsLock); + + /* Create the current segment file, if necessary */ + if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno)) { + LWLock *lock = SimpleLruGetBankLock(CommitTsCtl, pageno); + int slotno; + + LWLockAcquire(lock, LW_EXCLUSIVE); + slotno = ZeroCommitTsPage(pageno, false); + SimpleLruWritePage(CommitTsCtl, slotno); + Assert(!CommitTsCtl->shared->page_dirty[slotno]); + LWLockRelease(lock); + } + + /* Change the activation status in shared memory. */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + commitTsShared->commitTsActive = true; + LWLockRelease(CommitTsLock); } /* @@ -774,57 +714,53 @@ ActivateCommitTs(void) * Resets CommitTs into invalid state to make sure we don't hand back * possibly-invalid data; also removes segments of old data. */ -static void -DeactivateCommitTs(void) -{ - /* - * Cleanup the status in the shared memory. - * - * We reset everything in the commitTsShared record to prevent user from - * getting confusing data about last committed transaction on the standby - * when the module was activated repeatedly on the primary. - */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - - commitTsShared->commitTsActive = false; - commitTsShared->xidLastCommit = InvalidTransactionId; - TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time); - commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId; - - TransamVariables->oldestCommitTsXid = InvalidTransactionId; - TransamVariables->newestCommitTsXid = InvalidTransactionId; - - /* - * Remove *all* files. This is necessary so that there are no leftover - * files; in the case where this feature is later enabled after running - * with it disabled for some time there may be a gap in the file sequence. - * (We can probably tolerate out-of-sequence files, as they are going to - * be overwritten anyway when we wrap around, but it seems better to be - * tidy.) - * - * Note that we do this with CommitTsLock acquired in exclusive mode. This - * is very heavy-handed, but since this routine can only be called in the - * replica and should happen very rarely, we don't worry too much about - * it. Note also that no process should be consulting this SLRU if we - * have just deactivated it. - */ - (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL); - - LWLockRelease(CommitTsLock); +static void DeactivateCommitTs(void) { + /* + * Cleanup the status in the shared memory. + * + * We reset everything in the commitTsShared record to prevent user from + * getting confusing data about last committed transaction on the standby + * when the module was activated repeatedly on the primary. + */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + + commitTsShared->commitTsActive = false; + commitTsShared->xidLastCommit = InvalidTransactionId; + TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time); + commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId; + + TransamVariables->oldestCommitTsXid = InvalidTransactionId; + TransamVariables->newestCommitTsXid = InvalidTransactionId; + + /* + * Remove *all* files. This is necessary so that there are no leftover + * files; in the case where this feature is later enabled after running + * with it disabled for some time there may be a gap in the file sequence. + * (We can probably tolerate out-of-sequence files, as they are going to + * be overwritten anyway when we wrap around, but it seems better to be + * tidy.) + * + * Note that we do this with CommitTsLock acquired in exclusive mode. This + * is very heavy-handed, but since this routine can only be called in the + * replica and should happen very rarely, we don't worry too much about + * it. Note also that no process should be consulting this SLRU if we + * have just deactivated it. + */ + (void)SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL); + + LWLockRelease(CommitTsLock); } /* * Perform a checkpoint --- either during shutdown, or on-the-fly */ -void -CheckPointCommitTs(void) -{ - /* - * Write dirty CommitTs pages to disk. This may result in sync requests - * queued for later handling by ProcessSyncRequests(), as part of the - * checkpoint. - */ - SimpleLruWriteAll(CommitTsCtl, true); +void CheckPointCommitTs(void) { + /* + * Write dirty CommitTs pages to disk. This may result in sync requests + * queued for later handling by ProcessSyncRequests(), as part of the + * checkpoint. + */ + SimpleLruWriteAll(CommitTsCtl, true); } /* @@ -838,39 +774,37 @@ CheckPointCommitTs(void) * NB: the current implementation relies on track_commit_timestamp being * PGC_POSTMASTER. */ -void -ExtendCommitTs(TransactionId newestXact) -{ - int64 pageno; - LWLock *lock; +void ExtendCommitTs(TransactionId newestXact) { + int64 pageno; + LWLock *lock; - /* - * Nothing to do if module not enabled. Note we do an unlocked read of - * the flag here, which is okay because this routine is only called from - * GetNewTransactionId, which is never called in a standby. - */ - Assert(!InRecovery); - if (!commitTsShared->commitTsActive) - return; + /* + * Nothing to do if module not enabled. Note we do an unlocked read of + * the flag here, which is okay because this routine is only called from + * GetNewTransactionId, which is never called in a standby. + */ + Assert(!InRecovery); + if (!commitTsShared->commitTsActive) + return; - /* - * No work except at first XID of a page. But beware: just after - * wraparound, the first XID of page zero is FirstNormalTransactionId. - */ - if (TransactionIdToCTsEntry(newestXact) != 0 && - !TransactionIdEquals(newestXact, FirstNormalTransactionId)) - return; + /* + * No work except at first XID of a page. But beware: just after + * wraparound, the first XID of page zero is FirstNormalTransactionId. + */ + if (TransactionIdToCTsEntry(newestXact) != 0 && + !TransactionIdEquals(newestXact, FirstNormalTransactionId)) + return; - pageno = TransactionIdToCTsPage(newestXact); + pageno = TransactionIdToCTsPage(newestXact); - lock = SimpleLruGetBankLock(CommitTsCtl, pageno); + lock = SimpleLruGetBankLock(CommitTsCtl, pageno); - LWLockAcquire(lock, LW_EXCLUSIVE); + LWLockAcquire(lock, LW_EXCLUSIVE); - /* Zero the page and make an XLOG entry about it */ - ZeroCommitTsPage(pageno, !InRecovery); + /* Zero the page and make an XLOG entry about it */ + ZeroCommitTsPage(pageno, !InRecovery); - LWLockRelease(lock); + LWLockRelease(lock); } /* @@ -879,70 +813,59 @@ ExtendCommitTs(TransactionId newestXact) * * Note that we don't need to flush XLOG here. */ -void -TruncateCommitTs(TransactionId oldestXact) -{ - int64 cutoffPage; +void TruncateCommitTs(TransactionId oldestXact) { + int64 cutoffPage; - /* - * The cutoff point is the start of the segment containing oldestXact. We - * pass the *page* containing oldestXact to SimpleLruTruncate. - */ - cutoffPage = TransactionIdToCTsPage(oldestXact); + /* + * The cutoff point is the start of the segment containing oldestXact. We + * pass the *page* containing oldestXact to SimpleLruTruncate. + */ + cutoffPage = TransactionIdToCTsPage(oldestXact); - /* Check to see if there's any files that could be removed */ - if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence, - &cutoffPage)) - return; /* nothing to remove */ + /* Check to see if there's any files that could be removed */ + if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence, &cutoffPage)) + return; /* nothing to remove */ - /* Write XLOG record */ - WriteTruncateXlogRec(cutoffPage, oldestXact); + /* Write XLOG record */ + WriteTruncateXlogRec(cutoffPage, oldestXact); - /* Now we can remove the old CommitTs segment(s) */ - SimpleLruTruncate(CommitTsCtl, cutoffPage); + /* Now we can remove the old CommitTs segment(s) */ + SimpleLruTruncate(CommitTsCtl, cutoffPage); } /* * Set the limit values between which commit TS can be consulted. */ -void -SetCommitTsLimit(TransactionId oldestXact, TransactionId newestXact) -{ - /* - * Be careful not to overwrite values that are either further into the - * "future" or signal a disabled committs. - */ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - if (TransamVariables->oldestCommitTsXid != InvalidTransactionId) - { - if (TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact)) - TransamVariables->oldestCommitTsXid = oldestXact; - if (TransactionIdPrecedes(newestXact, TransamVariables->newestCommitTsXid)) - TransamVariables->newestCommitTsXid = newestXact; - } - else - { - Assert(TransamVariables->newestCommitTsXid == InvalidTransactionId); - TransamVariables->oldestCommitTsXid = oldestXact; - TransamVariables->newestCommitTsXid = newestXact; - } - LWLockRelease(CommitTsLock); +void SetCommitTsLimit(TransactionId oldestXact, TransactionId newestXact) { + /* + * Be careful not to overwrite values that are either further into the + * "future" or signal a disabled committs. + */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + if (TransamVariables->oldestCommitTsXid != InvalidTransactionId) { + if (TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact)) + TransamVariables->oldestCommitTsXid = oldestXact; + if (TransactionIdPrecedes(newestXact, TransamVariables->newestCommitTsXid)) + TransamVariables->newestCommitTsXid = newestXact; + } else { + Assert(TransamVariables->newestCommitTsXid == InvalidTransactionId); + TransamVariables->oldestCommitTsXid = oldestXact; + TransamVariables->newestCommitTsXid = newestXact; + } + LWLockRelease(CommitTsLock); } /* * Move forwards the oldest commitTS value that can be consulted */ -void -AdvanceOldestCommitTsXid(TransactionId oldestXact) -{ - LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); - if (TransamVariables->oldestCommitTsXid != InvalidTransactionId && - TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact)) - TransamVariables->oldestCommitTsXid = oldestXact; - LWLockRelease(CommitTsLock); +void AdvanceOldestCommitTsXid(TransactionId oldestXact) { + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + if (TransamVariables->oldestCommitTsXid != InvalidTransactionId && + TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact)) + TransamVariables->oldestCommitTsXid = oldestXact; + LWLockRelease(CommitTsLock); } - /* * Decide whether a commitTS page number is "older" for truncation purposes. * Analogous to CLOGPagePrecedes(). @@ -985,12 +908,12 @@ CommitTsPagePrecedes(int64 page1, int64 page2) /* * Write a ZEROPAGE xlog record */ -static void +static XLogRecPtr WriteZeroPageXlogRec(int64 pageno) { XLogBeginInsert(); XLogRegisterData(&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 3c06ac45532..126445f3178 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -84,6 +84,7 @@ #include "pg_trace.h" #include "pgstat.h" #include "postmaster/autovacuum.h" +#include "storage/bufpage.h" #include "storage/pmsignal.h" #include "storage/proc.h" #include "storage/procarray.h" @@ -107,7 +108,7 @@ */ /* We need four bytes per offset */ -#define MULTIXACT_OFFSETS_PER_PAGE (BLCKSZ / sizeof(MultiXactOffset)) +#define MULTIXACT_OFFSETS_PER_PAGE (SizeOfPageContents / sizeof(MultiXactOffset)) static inline int64 MultiXactIdToOffsetPage(MultiXactId multi) @@ -132,8 +133,8 @@ MultiXactIdToOffsetSegment(MultiXactId multi) * 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 @@ -151,7 +152,7 @@ MultiXactIdToOffsetSegment(MultiXactId multi) /* 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) @@ -413,7 +414,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, @@ -939,7 +940,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; @@ -994,12 +995,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); @@ -1427,7 +1428,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; @@ -1476,7 +1477,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; @@ -1544,7 +1545,7 @@ retry: } xactptr = (TransactionId *) - (MultiXactMemberCtl->shared->page_buffer[slotno] + memberoff); + (PageGetContents(MultiXactMemberCtl->shared->page_buffer[slotno]) + memberoff); if (!TransactionIdIsValid(*xactptr)) { @@ -1555,7 +1556,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; @@ -2074,11 +2075,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; } @@ -2090,11 +2097,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; } @@ -2218,10 +2231,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); @@ -2252,9 +2265,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 @@ -2909,7 +2922,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)); @@ -3351,12 +3364,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(&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 fe56286d9a9..2500d5d6621 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.h" @@ -169,6 +170,7 @@ typedef enum SLRU_WRITE_FAILED, SLRU_FSYNC_FAILED, SLRU_CLOSE_FAILED, + SLRU_DATA_CORRUPTED, } SlruErrorCause; static SlruErrorCause slru_errcause; @@ -391,8 +393,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); + /* Initialize the page. */ + PageInitSLRU(shared->page_buffer[slotno], BLCKSZ, 0); /* Set the LSNs for this new page to zero */ SimpleLruZeroLSNs(ctl, slotno); @@ -808,6 +810,8 @@ SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno) off_t offset = rpageno * BLCKSZ; char path[MAXPGPATH]; int fd; + bool checksum_failure; + bool verified; SlruFileName(ctl, path, segno); @@ -831,7 +835,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; } @@ -854,6 +858,14 @@ SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno) return false; } + if (!PageIsVerified(shared->page_buffer[slotno], pageno, PIV_LOG_WARNING, + &checksum_failure)) + { + slru_errcause = SLRU_DATA_CORRUPTED; + slru_errno = 0; + return false; + } + return true; } @@ -880,6 +892,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); @@ -888,41 +902,21 @@ SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata) * Honor the write-WAL-before-data rule, if appropriate, so that we do not * write out data before associated WAL records. This is the same action * performed during FlushBuffer() in the main buffer manager. + * + * The largest async-commit LSN for the page is maintained through page LSN. */ - 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(); } /* @@ -987,6 +981,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) @@ -1107,6 +1103,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 15153618fad..13bf26a942b 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; @@ -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 4bd37d5beb5..ce4d95b7719 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 */ /* @@ -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 d82114ffca1..39fd1afbbba 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 diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c index dbb49ed9197..7819f9480a8 100644 --- a/src/backend/storage/page/bufpage.c +++ b/src/backend/storage/page/bufpage.c @@ -59,6 +59,32 @@ 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); +} + /* * PageIsVerified diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c index f20be82862a..7ba06e09d4a 100644 --- a/src/bin/pg_checksums/pg_checksums.c +++ b/src/bin/pg_checksums/pg_checksums.c @@ -16,6 +16,7 @@ #include #include +#include #include #include #include @@ -592,12 +593,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_DIR, 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_DIR, 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 d6bbbd0ceda..0d31cdb2055 100644 --- a/src/bin/pg_resetwal/t/001_basic.pl +++ b/src/bin/pg_resetwal/t/001_basic.pl @@ -213,7 +213,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; # --multixact-ids argument is "new,old" push @cmd, '--multixact-ids' => sprintf("%d,%d", @@ -221,11 +221,11 @@ push @cmd, 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, '--multixact-offset' => (hex($files[-1]) + 1) * $mult; @files = get_slru_files('pg_xact'); -$mult = 32 * $blcksz * 4; +$mult = 32 * ($blcksz - 24) * 4; push @cmd, '--oldest-transaction-id' => (hex($files[0]) == 0 ? 3 : hex($files[0]) * $mult), diff --git a/src/bin/pg_upgrade/file.c b/src/bin/pg_upgrade/file.c index 91ed16acb08..2831dd5ec2b 100644 --- a/src/bin/pg_upgrade/file.c +++ b/src/bin/pg_upgrade/file.c @@ -9,6 +9,7 @@ #include "postgres_fe.h" +#include #include #include #include @@ -457,3 +458,180 @@ check_hard_link(transferMode transfer_mode) 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); +} + +/* + * Upgrade a single clog segment to add a page header on each page. + */ +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); +} + +/* + * Upgrade the clog files to add a page header to each SLRU page. + */ +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 536e49d2616..bad7613e291 100644 --- a/src/bin/pg_upgrade/pg_upgrade.c +++ b/src/bin/pg_upgrade/pg_upgrade.c @@ -749,14 +749,28 @@ 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"; + + /* + * In post-17 clusters, a page header is added to each SLRU page. + * Perform a one-time conversion of the clog files if the old + * cluster and the new cluster use different SLRU formats. + */ + 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, @@ -791,7 +805,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"); @@ -811,7 +826,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 69c965bb7d0..34b03df2f08 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -114,6 +114,11 @@ extern char *output_files[]; */ #define MULTIXACT_FORMATCHANGE_CAT_VER 201301231 +/* + * A page header was added to each SLRU page in 18.0. + */ +#define SLRU_PAGE_HEADER_CAT_VER 202506261 + /* * large object chunk size added to pg_controldata, * commit 5f93c37805e7485488480916b4585e098d3cc883 @@ -425,6 +430,7 @@ void rewriteVisibilityMap(const char *fromfile, const char *tofile, void check_file_clone(void); void check_copy_file_range(void); void check_hard_link(transferMode transfer_mode); +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/catalog/catversion.h b/src/include/catalog/catversion.h index d63db42ed7b..c1e086be1ea 100644 --- a/src/include/catalog/catversion.h +++ b/src/include/catalog/catversion.h @@ -57,6 +57,6 @@ */ /* yyyymmddN */ -#define CATALOG_VERSION_NO 202506251 +#define CATALOG_VERSION_NO 202506261 #endif diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h index aeb67c498c5..4c9ab9302bb 100644 --- a/src/include/storage/bufpage.h +++ b/src/include/storage/bufpage.h @@ -205,6 +205,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 /* ---------------------------------------------------------------- @@ -261,6 +262,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 PageIsVerified(PageData *page, BlockNumber blkno, int flags, bool *checksum_failure_p); extern OffsetNumber PageAddItemExtended(Page page, Item item, Size size, diff --git a/src/test/modules/test_slru/test_slru.c b/src/test/modules/test_slru/test_slru.c index 32750930e43..6b24b19e266 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)); -- 2.47.1