diff options
Diffstat (limited to 'src/backend/access/transam/commit_ts.c')
| -rw-r--r-- | src/backend/access/transam/commit_ts.c | 1032 |
1 files changed, 1032 insertions, 0 deletions
diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c new file mode 100644 index 0000000..edbe3cf --- /dev/null +++ b/src/backend/access/transam/commit_ts.c @@ -0,0 +1,1032 @@ +/*------------------------------------------------------------------------- + * + * commit_ts.c + * PostgreSQL commit timestamp manager + * + * This module is a pg_xact-like system that stores the commit timestamp + * for each transaction. + * + * XLOG interactions: this module generates an XLOG record whenever a new + * CommitTs page is initialized to zeroes. Also, one XLOG record is + * generated for setting of values when the caller requests it; this allows + * us to support values coming from places other than transaction commit. + * Other writes of CommitTS come from recording of transaction commit in + * xact.c, which generates its own XLOG records for these events and will + * re-perform the status update on redo; so we need make no additional XLOG + * entry here. + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/backend/access/transam/commit_ts.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/commit_ts.h" +#include "access/htup_details.h" +#include "access/slru.h" +#include "access/transam.h" +#include "catalog/pg_type.h" +#include "funcapi.h" +#include "miscadmin.h" +#include "pg_trace.h" +#include "storage/shmem.h" +#include "utils/builtins.h" +#include "utils/snapmgr.h" +#include "utils/timestamp.h" + +/* + * Defines for CommitTs page sizes. A page is the same BLCKSZ as is used + * everywhere else in Postgres. + * + * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF, + * CommitTs page numbering also wraps around at + * 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE, and CommitTs segment numbering at + * 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT. We need take no + * explicit notice of that fact in this module, except when comparing segment + * and page numbers in TruncateCommitTs (see CommitTsPagePrecedes). + */ + +/* + * We need 8+2 bytes per xact. Note that enlarging this struct might mean + * the largest possible file name is more than 5 chars long; see + * SlruScanDirectory. + */ +typedef struct CommitTimestampEntry +{ + TimestampTz time; + RepOriginId nodeid; +} CommitTimestampEntry; + +#define SizeOfCommitTimestampEntry (offsetof(CommitTimestampEntry, nodeid) + \ + sizeof(RepOriginId)) + +#define COMMIT_TS_XACTS_PER_PAGE \ + (BLCKSZ / SizeOfCommitTimestampEntry) + +#define TransactionIdToCTsPage(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 + */ +static SlruCtlData CommitTsCtlData; + +#define CommitTsCtl (&CommitTsCtlData) + +/* + * We keep a cache of the last value set in shared memory. + * + * This is also good place to keep the activation status. We keep this + * separate from the GUC so that the standby can activate the module if the + * primary has it active independently of the value of the GUC. + * + * This is protected by CommitTsLock. In some places, we use commitTsActive + * without acquiring the lock; where this happens, a comment explains the + * rationale for it. + */ +typedef struct CommitTimestampShared +{ + TransactionId xidLastCommit; + CommitTimestampEntry dataLastCommit; + bool commitTsActive; +} CommitTimestampShared; + +CommitTimestampShared *commitTsShared; + + +/* GUC variable */ +bool track_commit_timestamp; + +static void SetXidCommitTsInPage(TransactionId xid, int nsubxids, + TransactionId *subxids, TimestampTz ts, + RepOriginId nodeid, int pageno); +static void TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, + RepOriginId nodeid, int slotno); +static void error_commit_ts_disabled(void); +static int ZeroCommitTsPage(int pageno, bool writeXlog); +static bool CommitTsPagePrecedes(int page1, int page2); +static void ActivateCommitTs(void); +static void DeactivateCommitTs(void); +static void WriteZeroPageXlogRec(int pageno); +static void WriteTruncateXlogRec(int pageno, TransactionId oldestXid); + +/* + * TransactionTreeSetCommitTsData + * + * Record the final commit timestamp of transaction entries in the commit log + * for a transaction and its subtransaction tree, as efficiently as possible. + * + * xid is the top level transaction id. + * + * subxids is an array of xids of length nsubxids, representing subtransactions + * in the tree of xid. In various cases nsubxids may be zero. + * The reason why tracking just the parent xid commit timestamp is not enough + * is that the subtrans SLRU does not stay valid across crashes (it's not + * permanent) so we need to keep the information about them here. If the + * 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 (;;) + { + int 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(ShmemVariableCache->newestCommitTsXid, newestXact)) + ShmemVariableCache->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, int pageno) +{ + int slotno; + int i; + + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + + slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, xid); + + TransactionIdSetCommitTs(xid, ts, nodeid, slotno); + for (i = 0; i < nsubxids; i++) + TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno); + + CommitTsCtl->shared->page_dirty[slotno] = true; + + LWLockRelease(CommitTsSLRULock); +} + +/* + * Sets the commit timestamp of a single transaction. + * + * Must be called with CommitTsSLRULock held + */ +static void +TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts, + RepOriginId nodeid, int slotno) +{ + int entryno = TransactionIdToCTsEntry(xid); + CommitTimestampEntry entry; + + Assert(TransactionIdIsNormal(xid)); + + entry.time = ts; + entry.nodeid = nodeid; + + memcpy(CommitTsCtl->shared->page_buffer[slotno] + + SizeOfCommitTimestampEntry * entryno, + &entry, SizeOfCommitTimestampEntry); +} + +/* + * Interrogate the commit timestamp of a transaction. + * + * The return value indicates whether a commit timestamp record was found for + * the given xid. The timestamp value is returned in *ts (which may not be + * 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) +{ + int 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 = ShmemVariableCache->oldestCommitTsXid; + newestCommitTsXid = ShmemVariableCache->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(CommitTsSLRULock); + return *ts != 0; +} + +/* + * Return the Xid of the latest committed transaction. (As far as this module + * is concerned, anyway; it's up to the caller to ensure the value is useful + * for its purposes.) + * + * 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; + + LWLockAcquire(CommitTsLock, LW_SHARED); + + /* 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); + + 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"))); +} + +/* + * 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; + + found = TransactionIdGetCommitTsData(xid, &ts, NULL); + + if (!found) + PG_RETURN_NULL(); + + PG_RETURN_TIMESTAMPTZ(ts); +} + + +/* + * pg_last_committed_xact + * + * SQL-callable wrapper to obtain some information about the latest + * 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; + + /* and construct a tuple with our data */ + xid = GetLatestCommitTsData(&ts, &nodeid); + + /* + * Construct a tuple descriptor for the result row. This must match this + * function's pg_proc entry! + */ + tupdesc = CreateTemplateTupleDesc(3); + TupleDescInitEntry(tupdesc, (AttrNumber) 1, "xid", + XIDOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 2, "timestamp", + TIMESTAMPTZOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 3, "roident", + OIDOID, -1, 0); + tupdesc = BlessTupleDesc(tupdesc); + + if (!TransactionIdIsNormal(xid)) + { + memset(nulls, true, sizeof(nulls)); + } + else + { + values[0] = TransactionIdGetDatum(xid); + nulls[0] = false; + + values[1] = TimestampTzGetDatum(ts); + nulls[1] = false; + + values[2] = ObjectIdGetDatum((Oid) nodeid); + nulls[2] = false; + } + + htup = heap_form_tuple(tupdesc, values, nulls); + + PG_RETURN_DATUM(HeapTupleGetDatum(htup)); +} + +/* + * pg_xact_commit_timestamp_origin + * + * 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; + + found = TransactionIdGetCommitTsData(xid, &ts, &nodeid); + + /* + * Construct a tuple descriptor for the result row. This must match this + * function's pg_proc entry! + */ + tupdesc = CreateTemplateTupleDesc(2); + TupleDescInitEntry(tupdesc, (AttrNumber) 1, "timestamp", + TIMESTAMPTZOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 2, "roident", + OIDOID, -1, 0); + tupdesc = BlessTupleDesc(tupdesc); + + if (!found) + { + memset(nulls, true, sizeof(nulls)); + } + else + { + values[0] = TimestampTzGetDatum(ts); + nulls[0] = false; + + values[1] = ObjectIdGetDatum((Oid) nodeid); + nulls[1] = false; + } + + htup = heap_form_tuple(tupdesc, values, nulls); + + PG_RETURN_DATUM(HeapTupleGetDatum(htup)); +} + +/* + * Number of shared CommitTS buffers. + * + * We use a very similar logic as for the number of CLOG buffers; see comments + * in CLOGShmemBuffers. + */ +Size +CommitTsShmemBuffers(void) +{ + return Min(16, Max(4, NBuffers / 1024)); +} + +/* + * Shared memory sizing for CommitTs + */ +Size +CommitTsShmemSize(void) +{ + return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) + + sizeof(CommitTimestampShared); +} + +/* + * Initialize CommitTs at system startup (postmaster start or standalone + * backend) + */ +void +CommitTsShmemInit(void) +{ + bool found; + + CommitTsCtl->PagePrecedes = CommitTsPagePrecedes; + SimpleLruInit(CommitTsCtl, "CommitTs", CommitTsShmemBuffers(), 0, + CommitTsSLRULock, "pg_commit_ts", + LWTRANCHE_COMMITTS_BUFFER, + SYNC_HANDLER_COMMIT_TS); + 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); +} + +/* + * This function must be called ONCE on system install. + * + * (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. + */ +} + +/* + * Initialize (or reinitialize) a page of CommitTs to zeroes. + * If writeXlog is true, also emit an XLOG record saying we did this. + * + * The page is not actually written, just set up in shared memory. + * The slot number of the new page is returned. + * + * Control lock must be held at entry, and will be held at exit. + */ +static int +ZeroCommitTsPage(int pageno, bool writeXlog) +{ + int slotno; + + slotno = SimpleLruZeroPage(CommitTsCtl, pageno); + + if (writeXlog) + WriteZeroPageXlogRec(pageno); + + return slotno; +} + +/* + * This must be called ONCE during postmaster or standalone-backend startup, + * after StartupXLOG has initialized ShmemVariableCache->nextXid. + */ +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(); +} + +/* + * 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(); +} + +/* + * 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 + * changed in the primary. + * + * The reason why this SLRU needs separate activation/deactivation functions is + * that it can be enabled/disabled during start and the activation/deactivation + * on the primary is propagated to the standby via replay. Other SLRUs don't + * have this property and they can be just initialized during normal startup. + * + * This is in charge of creating the currently active segment, if it's not + * already there. The reason for this is that the server might have been + * running with this module disabled for a while and thus might have skipped + * the normal creation point. + */ +static void +ActivateCommitTs(void) +{ + TransactionId xid; + int 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(ShmemVariableCache->nextXid); + pageno = TransactionIdToCTsPage(xid); + + /* + * Re-Initialize our idea of the latest page number. + */ + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + CommitTsCtl->shared->latest_page_number = pageno; + LWLockRelease(CommitTsSLRULock); + + /* + * 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 (ShmemVariableCache->oldestCommitTsXid == InvalidTransactionId) + { + ShmemVariableCache->oldestCommitTsXid = + ShmemVariableCache->newestCommitTsXid = ReadNextTransactionId(); + } + LWLockRelease(CommitTsLock); + + /* Create the current segment file, if necessary */ + if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno)) + { + int slotno; + + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + slotno = ZeroCommitTsPage(pageno, false); + SimpleLruWritePage(CommitTsCtl, slotno); + Assert(!CommitTsCtl->shared->page_dirty[slotno]); + LWLockRelease(CommitTsSLRULock); + } + + /* Change the activation status in shared memory. */ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + commitTsShared->commitTsActive = true; + LWLockRelease(CommitTsLock); +} + +/* + * Deactivate this module. + * + * This must be called when the track_commit_timestamp parameter is turned off. + * This happens during postmaster or standalone-backend startup, or during WAL + * replay. + * + * 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; + + ShmemVariableCache->oldestCommitTsXid = InvalidTransactionId; + ShmemVariableCache->newestCommitTsXid = InvalidTransactionId; + + LWLockRelease(CommitTsLock); + + /* + * 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.) + */ + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL); + LWLockRelease(CommitTsSLRULock); +} + +/* + * 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); +} + +/* + * Make sure that CommitTs has room for a newly-allocated XID. + * + * NB: this is called while holding XidGenLock. We want it to be very fast + * most of the time; even when it's not so fast, no actual I/O need happen + * unless we're forced to write out a dirty CommitTs or xlog page to make room + * in shared memory. + * + * NB: the current implementation relies on track_commit_timestamp being + * PGC_POSTMASTER. + */ +void +ExtendCommitTs(TransactionId newestXact) +{ + int pageno; + + /* + * 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; + + pageno = TransactionIdToCTsPage(newestXact); + + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + + /* Zero the page and make an XLOG entry about it */ + ZeroCommitTsPage(pageno, !InRecovery); + + LWLockRelease(CommitTsSLRULock); +} + +/* + * Remove all CommitTs segments before the one holding the passed + * transaction ID. + * + * Note that we don't need to flush XLOG here. + */ +void +TruncateCommitTs(TransactionId oldestXact) +{ + int cutoffPage; + + /* + * 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 */ + + /* Write XLOG record */ + WriteTruncateXlogRec(cutoffPage, oldestXact); + + /* 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 (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId) + { + if (TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, oldestXact)) + ShmemVariableCache->oldestCommitTsXid = oldestXact; + if (TransactionIdPrecedes(newestXact, ShmemVariableCache->newestCommitTsXid)) + ShmemVariableCache->newestCommitTsXid = newestXact; + } + else + { + Assert(ShmemVariableCache->newestCommitTsXid == InvalidTransactionId); + ShmemVariableCache->oldestCommitTsXid = oldestXact; + ShmemVariableCache->newestCommitTsXid = newestXact; + } + LWLockRelease(CommitTsLock); +} + +/* + * Move forwards the oldest commitTS value that can be consulted + */ +void +AdvanceOldestCommitTsXid(TransactionId oldestXact) +{ + LWLockAcquire(CommitTsLock, LW_EXCLUSIVE); + if (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId && + TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, oldestXact)) + ShmemVariableCache->oldestCommitTsXid = oldestXact; + LWLockRelease(CommitTsLock); +} + + +/* + * Decide whether a commitTS page number is "older" for truncation purposes. + * Analogous to CLOGPagePrecedes(). + * + * At default BLCKSZ, (1 << 31) % COMMIT_TS_XACTS_PER_PAGE == 128. This + * introduces differences compared to CLOG and the other SLRUs having (1 << + * 31) % per_page == 0. This function never tests exactly + * TransactionIdPrecedes(x-2^31, x). When the system reaches xidStopLimit, + * there are two possible counts of page boundaries between oldestXact and the + * latest XID assigned, depending on whether oldestXact is within the first + * 128 entries of its page. Since this function doesn't know the location of + * oldestXact within page2, it returns false for one page that actually is + * expendable. This is a wider (yet still negligible) version of the + * truncation opportunity that CLOGPagePrecedes() cannot recognize. + * + * For the sake of a worked example, number entries with decimal values such + * that page1==1 entries range from 1.0 to 1.999. Let N+0.15 be the number of + * pages that 2^31 entries will span (N is an integer). If oldestXact=N+2.1, + * then the final safe XID assignment leaves newestXact=1.95. We keep page 2, + * because entry=2.85 is the border that toggles whether entries precede the + * last entry of the oldestXact page. While page 2 is expendable at + * oldestXact=N+2.1, it would be precious at oldestXact=N+2.9. + */ +static bool +CommitTsPagePrecedes(int page1, int page2) +{ + TransactionId xid1; + TransactionId xid2; + + xid1 = ((TransactionId) page1) * COMMIT_TS_XACTS_PER_PAGE; + xid1 += FirstNormalTransactionId + 1; + xid2 = ((TransactionId) page2) * COMMIT_TS_XACTS_PER_PAGE; + xid2 += FirstNormalTransactionId + 1; + + return (TransactionIdPrecedes(xid1, xid2) && + TransactionIdPrecedes(xid1, xid2 + COMMIT_TS_XACTS_PER_PAGE - 1)); +} + + +/* + * Write a ZEROPAGE xlog record + */ +static void +WriteZeroPageXlogRec(int pageno) +{ + XLogBeginInsert(); + XLogRegisterData((char *) (&pageno), sizeof(int)); + (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE); +} + +/* + * Write a TRUNCATE xlog record + */ +static void +WriteTruncateXlogRec(int pageno, TransactionId oldestXid) +{ + xl_commit_ts_truncate xlrec; + + xlrec.pageno = pageno; + xlrec.oldestXid = oldestXid; + + XLogBeginInsert(); + XLogRegisterData((char *) (&xlrec), SizeOfCommitTsTruncate); + (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_TRUNCATE); +} + +/* + * CommitTS resource manager's routines + */ +void +commit_ts_redo(XLogReaderState *record) +{ + uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK; + + /* Backup blocks are not used in commit_ts records */ + Assert(!XLogRecHasAnyBlockRefs(record)); + + if (info == COMMIT_TS_ZEROPAGE) + { + int pageno; + int slotno; + + memcpy(&pageno, XLogRecGetData(record), sizeof(int)); + + LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE); + + slotno = ZeroCommitTsPage(pageno, false); + SimpleLruWritePage(CommitTsCtl, slotno); + Assert(!CommitTsCtl->shared->page_dirty[slotno]); + + LWLockRelease(CommitTsSLRULock); + } + else if (info == COMMIT_TS_TRUNCATE) + { + xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) XLogRecGetData(record); + + AdvanceOldestCommitTsXid(trunc->oldestXid); + + /* + * During XLOG replay, latest_page_number isn't set up yet; insert a + * suitable value to bypass the sanity test in SimpleLruTruncate. + */ + CommitTsCtl->shared->latest_page_number = trunc->pageno; + + SimpleLruTruncate(CommitTsCtl, trunc->pageno); + } + else + elog(PANIC, "commit_ts_redo: unknown op code %u", info); +} + +/* + * Entrypoint for sync.c to sync commit_ts files. + */ +int +committssyncfiletag(const FileTag *ftag, char *path) +{ + return SlruSyncFileTag(CommitTsCtl, ftag, path); +} |