From 22d80716f06e1b6000f788b5b96da70c0d7aa581 Mon Sep 17 00:00:00 2001 From: Andrey Lepikhov Date: Mon, 22 Mar 2021 16:36:38 +0500 Subject: [PATCH 2/4] CSN Snapshot --- doc/src/sgml/config.sgml | 50 +- src/backend/access/transam/Makefile | 1 + src/backend/access/transam/csn_log.c | 30 +- src/backend/access/transam/csn_snapshot.c | 724 ++++++++++++++++++++++ src/backend/access/transam/twophase.c | 158 +++++ src/backend/access/transam/xact.c | 29 + src/backend/access/transam/xlog.c | 26 +- src/backend/storage/ipc/ipci.c | 3 + src/backend/storage/ipc/procarray.c | 75 +++ src/backend/storage/lmgr/lwlocknames.txt | 1 + src/backend/storage/lmgr/proc.c | 5 + src/backend/utils/misc/guc.c | 21 + src/backend/utils/time/snapmgr.c | 146 ++++- src/bin/pg_waldump/csnlogdesc.c | 1 + src/include/access/csn_log.h | 3 +- src/include/access/csn_snapshot.h | 65 ++ src/include/catalog/pg_control.h | 6 + src/include/catalog/pg_proc.dat | 14 + src/include/datatype/timestamp.h | 2 + src/include/fmgr.h | 1 + src/include/storage/proc.h | 13 + src/include/storage/procarray.h | 3 + src/include/utils/snapmgr.h | 2 + src/include/utils/snapshot.h | 3 +- 24 files changed, 1367 insertions(+), 15 deletions(-) create mode 100644 src/backend/access/transam/csn_snapshot.c create mode 120000 src/bin/pg_waldump/csnlogdesc.c create mode 100644 src/include/access/csn_snapshot.h diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 5679b40dd5..fd421ea84d 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -9409,8 +9409,56 @@ dynamic_library_path = 'C:\tools\postgresql;H:\my_project\lib;$libdir' - + + CSN Based Snapshot + + + By default, The snapshots in PostgreSQL uses the + XID (TransactionID) to identify the status of the transaction, the in-progress + transactions, and the future transactions for all its visibility calculations. + + + + PostgreSQL also provides the CSN (commit-sequence-number) + based mechanism to identify the past-transactions and the ones that are yet to + be started/committed. + + + + + enable_csn_snapshot (boolean) + + enable_csn_snapshot configuration parameter + + + + + + Enable/disable the CSN based transaction visibility tracking for the snapshot. + + + + PostgreSQL uses the clock timestamp as a CSN, + so enabling the CSN based snapshots can be useful for implementing the global + snapshots and global transaction visibility. + + + + when enabled PostgreSQL creates + pg_csn directory under PGDATA to keep + the track of CSN and XID mappings. + + + + The default value is off. + + + + + + + Version and Platform Compatibility diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile index d1196cba87..fc0321ee6b 100644 --- a/src/backend/access/transam/Makefile +++ b/src/backend/access/transam/Makefile @@ -16,6 +16,7 @@ OBJS = \ clog.o \ commit_ts.o \ csn_log.o \ + csn_snapshot.o \ generic_xlog.o \ multixact.o \ parallel.o \ diff --git a/src/backend/access/transam/csn_log.c b/src/backend/access/transam/csn_log.c index 7a6d63e953..9511ecf48a 100644 --- a/src/backend/access/transam/csn_log.c +++ b/src/backend/access/transam/csn_log.c @@ -659,20 +659,38 @@ get_xmin_for_csn(void) } void -prepare_csn_env(bool enable) +prepare_csn_env(bool enable, bool same, TransactionId *xmin_for_csn_in_control) { if (enable) { - TransactionId nextxid = + if (same) + { + /* + * Database startup with no enable_csn_snapshot change and value is true, + * it can just transmit xmin_for_csn from pg_control to csnState->xmin_for_csn. + */ + csnShared->xmin_for_csn = *xmin_for_csn_in_control; + } + else + { + TransactionId nextxid = XidFromFullTransactionId(ShmemVariableCache->nextXid); - /* 'xmin_for_csn' for when turn xid-snapshot to csn-snapshot */ - csnShared->xmin_for_csn = nextxid; - /* produce the csnlog segment we want now and seek to current page */ - ActivateCSNlog(); + + /* 'xmin_for_csn' for when turn xid-snapshot to csn-snapshot */ + csnShared->xmin_for_csn = nextxid; + *xmin_for_csn_in_control = nextxid; + + /* produce the csnlog segment we want now and seek to current page */ + ActivateCSNlog(); + } } else + { /* Try to drop all csnlog seg */ DeactivateCSNlog(); + /* Clear xmin_for_csn in pg_control because we are xid-base snaposhot now. */ + *xmin_for_csn_in_control = InvalidTransactionId; + } } /* diff --git a/src/backend/access/transam/csn_snapshot.c b/src/backend/access/transam/csn_snapshot.c new file mode 100644 index 0000000000..33d769857f --- /dev/null +++ b/src/backend/access/transam/csn_snapshot.c @@ -0,0 +1,724 @@ +/*------------------------------------------------------------------------- + * + * csn_snapshot.c + * Support for cross-node snapshot isolation. + * + * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/backend/access/transam/csn_snapshot.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include + +#include "access/csn_log.h" +#include "access/csn_snapshot.h" +#include "access/transam.h" +#include "access/twophase.h" +#include "access/xact.h" +#include "storage/lmgr.h" +#include "storage/proc.h" +#include "storage/procarray.h" +#include "storage/shmem.h" +#include "utils/builtins.h" +#include "utils/guc.h" +#include "utils/snapmgr.h" +#include "miscadmin.h" + +/* Raise a warning if imported global_csn exceeds ours by this value. */ +#define SNAP_DESYNC_COMPLAIN (1*NSECS_PER_SEC) /* 1 second */ + +TransactionId xmin_for_csn = InvalidTransactionId; + +/* + * GUC to delay advance of oldestXid for this amount of time. Also determines + * the size CSNSnapshotXidMap circular buffer. + */ +int csn_snapshot_defer_time; + +/* DEBUG variable to simplify time skew modelling in tests. */ +int csn_time_shift; + +/* + * CSNSnapshotXidMap + * + * To be able to install csn snapshot that points to past we need to keep + * old versions of tuples and therefore delay advance of oldestXid. Here we + * keep track of correspondence between snapshot's snapshot_csn and oldestXid + * that was set at the time when the snapshot was taken. Much like the + * snapshot too old's OldSnapshotControlData does, but with finer granularity + * to seconds. + * + * Different strategies can be employed to hold oldestXid (e.g. we can track + * oldest csn-based snapshot among cluster nodes and map it oldestXid + * on each node). + * + * On each snapshot acquisition CSNSnapshotMapXmin() is called and stores + * correspondence between current snapshot_csn and oldestXmin in a sparse way: + * snapshot_csn is rounded to seconds (and here we use the fact that snapshot_csn + * is just a timestamp) and oldestXmin is stored in the circular buffer where + * rounded snapshot_csn acts as an offset from current circular buffer head. + * Size of the circular buffer is controlled by csn_snapshot_defer_time GUC. + * + * When csn snapshot arrives we check that its + * snapshot_csn is still in our map, otherwise we'll error out with "snapshot too + * old" message. If snapshot_csn is successfully mapped to oldestXid we move + * backend's pgxact->xmin to proc->originalXmin and fill pgxact->xmin to + * mapped oldestXid. That way GetOldestXmin() can take into account backends + * with imported csn snapshot and old tuple versions will be preserved. + * + * Also while calculating oldestXmin for our map in presence of imported + * csn snapshots we should use proc->originalXmin instead of pgxact->xmin + * that was set during import. Otherwise, we can create a feedback loop: + * xmin's of imported csn snapshots were calculated using our map and new + * entries in map going to be calculated based on that xmin's, and there is + * a risk to stuck forever with one non-increasing oldestXmin. All other + * callers of GetOldestXmin() are using pgxact->xmin so the old tuple versions + * are preserved. + */ +typedef struct CSNSnapshotXidMap +{ + int head; /* offset of current freshest value */ + int size; /* total size of circular buffer */ + CSN_atomic last_csn_seconds; /* last rounded csn that changed + * xmin_by_second[] */ + TransactionId *xmin_by_second; /* circular buffer of oldestXmin's */ +} +CSNSnapshotXidMap; + +static CSNSnapshotXidMap *csnXidMap; + +/* Estimate shared memory space needed */ +Size +CSNSnapshotShmemSize(void) +{ + Size size = 0; + + if (csn_snapshot_defer_time > 0) + { + size += sizeof(CSNSnapshotXidMap); + size += csn_snapshot_defer_time * sizeof(TransactionId); + size = MAXALIGN(size); + } + + return size; +} + +/* Init shared memory structures */ +void +CSNSnapshotShmemInit(void) +{ + bool found; + + if (csn_snapshot_defer_time > 0) + { + csnXidMap = ShmemInitStruct("gsXidMap", + sizeof(CSNSnapshotXidMap), + &found); + if (!found) + { + int i; + + pg_atomic_init_u64(&csnXidMap->last_csn_seconds, + (uint64) get_last_max_csn()); + csnXidMap->head = 0; + csnXidMap->size = csn_snapshot_defer_time; + csnXidMap->xmin_by_second = + ShmemAlloc(sizeof(TransactionId) * csnXidMap->size); + + for (i = 0; i < csnXidMap->size; i++) + csnXidMap->xmin_by_second[i] = InvalidTransactionId; + } + } +} + +/* + * GenerateCSN + * + * Generate CSN which is actually a local time. Also we are forcing + * this time to be always increasing. For portability reasons gettimeofday() + * and milliseconds resolution are used, but always increasing time helps to + * handle multiple transactions in one ms. + */ +CSN +GenerateCSN(CSN assign) +{ + struct timeval current_time; + CSN csn; + + Assert(get_csnlog_status() || csn_snapshot_defer_time > 0); + + if(assign != InvalidCSN && assign < get_last_max_csn()) + { + /* + * While assign not 0, we just want to make sure we log the max csn + * in last_max_csn. So we not care about the return value. + * + * We can return an Invalid value. + */ + return InvalidCSN; + } + + /* + * TODO: create some macro that add small random shift to current time. + */ + gettimeofday(¤t_time, NULL); + csn = (CSN) (((uint64) (current_time.tv_sec + csn_time_shift) * + (uint64) 1000000000) + (uint64) (current_time).tv_usec * (uint64) 1000); + + if (assign != InvalidCSN && csn < assign) + csn = assign; + + csn = set_last_max_csn(csn, false); + WriteAssignCSNXlogRec(csn); + return csn; +} + +/* + * CSNSnapshotPrepareCurrent + * + * Set InDoubt state for currently active transaction and return commit's + * global snapshot. + */ +CSN +CSNSnapshotPrepareCurrent(void) +{ + TransactionId xid = GetCurrentTransactionIdIfAny(); + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (TransactionIdIsValid(xid)) + { + TransactionId *subxids; + int nsubxids = xactGetCommittedChildren(&subxids); + CSNLogSetCSN(xid, nsubxids, subxids, InDoubtCSN, true); + } + + /* Nothing to write if we don't have xid */ + + return GenerateCSN(InvalidCSN); +} + + +/* + * CSNSnapshotAssignCurrent + * + * Assign CSN to the currently active transaction. CSN is supposedly + * maximal among of values returned by CSNSnapshotPrepareCurrent and + * pg_csn_snapshot_prepare. + */ +void +CSNSnapshotAssignCurrent(CSN csn) +{ + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (!CSNIsNormal(csn)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("pg_csn_snapshot_assign expects normal csn"))); + + Assert(csn != InvalidCSN); + /* We do not care the Generate result, we just want to make sure max + * csnState->last_max_csn value. + */ + GenerateCSN(csn); + + /* Set csn and defuse ProcArrayEndTransaction from assigning one */ + pg_atomic_write_u64(&MyProc->assignedCSN, csn); +} + +/* + * CSNSnapshotSync + * + * Due to time desynchronization on different nodes we can receive snapshot_csn + * which is greater than snapshot_csn on this node. To preserve proper isolation + * this node needs to wait when such snapshot_csn comes on local clock. + * + * This should happend relatively rare if nodes have running NTP/PTP/etc. + * Complain if wait time is more than SNAP_SYNC_COMPLAIN. + */ +void +CSNSnapshotSync(CSN remote_csn) +{ + CSN local_csn; + CSN delta; + + Assert(enable_csn_snapshot); + + for(;;) + { + if (get_last_max_csn() > remote_csn) + /* Everything is fine */ + return; + else if ((local_csn = GenerateCSN(InvalidCSN)) >= remote_csn) + /* + * Everything is fine too, but last_max_csn wasn't updated for + * some time. + */ + return; + + /* Okay we need to sleep now */ + delta = remote_csn - local_csn; + if (delta > SNAP_DESYNC_COMPLAIN) + ereport(WARNING, + (errmsg("remote global snapshot exceeds ours by more than a second"), + errhint("Consider running NTPd on servers participating in global transaction"))); + + /* TODO: report this sleeptime somewhere? */ + pg_usleep((long) (delta/NSECS_PER_USEC)); + + /* + * Loop that checks to ensure that we actually slept for specified + * amount of time. + */ + } + + Assert(false); /* Should not happend */ + return; +} + +/* + * TransactionIdGetCSN + * + * Get CSN for specified TransactionId taking care about special xids, + * xids beyond TransactionXmin and InDoubt states. + */ +CSN +TransactionIdGetCSN(TransactionId xid) +{ + CSN csn; + + Assert(get_csnlog_status()); + + /* Handle permanent TransactionId's for which we don't have mapping */ + if (!TransactionIdIsNormal(xid)) + { + if (xid == InvalidTransactionId) + return AbortedCSN; + if (xid == FrozenTransactionId || xid == BootstrapTransactionId) + return FrozenCSN; + Assert(false); /* Should not happend */ + } + + /* + * If we just switch a xid-snapsot to a csn_snapshot, we should handle a start + * xid for csn base check. Just in case we have prepared transaction which + * hold the TransactionXmin but without CSN. + */ + if (xmin_for_csn == InvalidTransactionId) + { + csn = get_xmin_for_csn(); + + /* + * If the instance started with enabled csn snapshots, use csn of its + * start point. If csn snapshots were enabled in this backend only, use + * xmin of this transaction. + */ + xmin_for_csn = (csn != InvalidTransactionId) ? csn : TransactionXmin; + } + + /* + * For the xid with 'xid >= TransactionXmin and xid < xmin_for_csn', + * it defined as unclear csn which follow xid-snapshot result. + */ + if(!TransactionIdPrecedes(xid, TransactionXmin) && + TransactionIdPrecedes(xid, xmin_for_csn)) + return UnclearCSN; + + /* + * For xids which less then TransactionXmin CSNLog can be already + * trimmed but we know that such transaction is definetly not concurrently + * running according to any snapshot including timetravel ones. Callers + * should check TransactionDidCommit after. + */ + if (TransactionIdPrecedes(xid, xmin_for_csn)) + return FrozenCSN; + + /* Read CSN from SLRU */ + csn = CSNLogGetCSNByXid(xid); + + /* + * If we faced InDoubt state then transaction is being committed and we + * should wait until CSN will be assigned so that visibility check + * could decide whether tuple is in snapshot. See also comments in + * CSNSnapshotPrecommit(). + */ + if (CSNIsInDoubt(csn)) + { + XactLockTableWait(xid, NULL, NULL, XLTW_None); + csn = CSNLogGetCSNByXid(xid); + Assert(CSNIsNormal(csn) || CSNIsAborted(csn)); + } + + Assert(CSNIsNormal(csn) || CSNIsInProgress(csn) || CSNIsAborted(csn)); + + return csn; +} + +/* + * CSNSnapshotStartup + * + * Set csnXidMap entries to oldestActiveXID during startup. + */ +void +CSNSnapshotStartup(TransactionId oldestActiveXID) +{ + /* + * Run only if we have initialized shared memory and csnXidMap + * is enabled. + */ + if (IsNormalProcessingMode() && + enable_csn_snapshot && csn_snapshot_defer_time > 0) + { + int i; + + Assert(TransactionIdIsValid(oldestActiveXID)); + for (i = 0; i < csnXidMap->size; i++) + csnXidMap->xmin_by_second[i] = oldestActiveXID; + + ProcArraySetCSNSnapshotXmin(oldestActiveXID); + } +} + +/* + * CSNSnapshotMapXmin + * + * Maintain circular buffer of oldestXmins for several seconds in past. This + * buffer allows to shift oldestXmin in the past when backend is importing + * CSN snapshot. Otherwise old versions of tuples that were needed for + * this transaction can be recycled by other processes (vacuum, HOT, etc). + * + * Locking here is not trivial. Called upon each snapshot creation after + * ProcArrayLock is released. Such usage creates several race conditions. It + * is possible that backend who got csn called CSNSnapshotMapXmin() + * only after other backends managed to get snapshot and complete + * CSNSnapshotMapXmin() call, or even committed. This is safe because + * + * * We already hold our xmin in MyProc, so our snapshot will not be + * harmed even though ProcArrayLock is released. + * + * * snapshot_csn is always pessimistically rounded up to the next + * second. + * + * * For performance reasons, xmin value for particular second is filled + * only once. Because of that instead of writing to buffer just our + * xmin (which is enough for our snapshot), we bump oldestXmin there -- + * it mitigates the possibility of damaging someone else's snapshot by + * writing to the buffer too advanced value in case of slowness of + * another backend who generated csn earlier, but didn't manage to + * insert it before us. + * + * * if CSNSnapshotMapXmin() founds a gap in several seconds between + * current call and latest completed call then it should fill that gap + * with latest known values instead of new one. Otherwise it is + * possible (however highly unlikely) that this gap also happend + * between taking snapshot and call to CSNSnapshotMapXmin() for some + * backend. And we are at risk to fill circullar buffer with + * oldestXmin's that are bigger then they actually were. + */ +void +CSNSnapshotMapXmin(CSN snapshot_csn) +{ + int offset, gap, i; + CSN csn_seconds; + CSN last_csn_seconds; + volatile TransactionId oldest_deferred_xmin; + TransactionId current_oldest_xmin; + TransactionId previous_oldest_xmin; + + /* Callers should check config values */ + Assert(csn_snapshot_defer_time > 0); + Assert(csnXidMap != NULL); + /* + * Round up snapshot_csn to the next second -- pessimistically and safely. + */ + csn_seconds = (snapshot_csn / NSECS_PER_SEC + 1); + + /* + * Fast-path check. Avoid taking exclusive CSNSnapshotXidMapLock lock + * if oldestXid was already written to xmin_by_second[] for this rounded + * snapshot_csn. + */ + if (pg_atomic_read_u64(&csnXidMap->last_csn_seconds) >= csn_seconds) + return; + + /* Ok, we have new entry (or entries) */ + LWLockAcquire(CSNSnapshotXidMapLock, LW_EXCLUSIVE); + + /* Re-check last_csn_seconds under lock */ + last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds); + if (last_csn_seconds >= csn_seconds) + { + LWLockRelease(CSNSnapshotXidMapLock); + return; + } + pg_atomic_write_u64(&csnXidMap->last_csn_seconds, csn_seconds); + + /* + * Count oldest_xmin. + * + * It was possible to calculate oldest_xmin during corresponding snapshot + * creation, but GetSnapshotData() intentionally reads only PgXact, but not + * PgProc. And we need info about originalXmin (see comment to csnXidMap) + * which is stored in PgProc because of threats in comments around PgXact + * about extending it with new fields. So just calculate oldest_xmin again, + * that anyway happens quite rarely. + */ + current_oldest_xmin = GetOldestNonRemovableTransactionId(NULL); + Assert(TransactionIdIsNormal(current_oldest_xmin)); + + previous_oldest_xmin = csnXidMap->xmin_by_second[csnXidMap->head]; + Assert(TransactionIdIsNormal(previous_oldest_xmin) || !enable_csn_snapshot); + + gap = csn_seconds - last_csn_seconds; + offset = csn_seconds % csnXidMap->size; + + /* Sanity check before we update head and gap */ + Assert( gap >= 1 ); + Assert( (csnXidMap->head + gap) % csnXidMap->size == offset ); + + gap = gap > csnXidMap->size ? csnXidMap->size : gap; + csnXidMap->head = offset; + + /* Fill new entry with current_oldest_xmin */ + csnXidMap->xmin_by_second[offset] = current_oldest_xmin; + + /* + * If we have gap then fill it with previous_oldest_xmin for reasons + * outlined in comment above this function. + */ + for (i = 1; i < gap; i++) + { + offset = (offset + csnXidMap->size - 1) % csnXidMap->size; + csnXidMap->xmin_by_second[offset] = previous_oldest_xmin; + } + + oldest_deferred_xmin = + csnXidMap->xmin_by_second[ (csnXidMap->head + 1) % csnXidMap->size ]; + + LWLockRelease(CSNSnapshotXidMapLock); + + /* + * Advance procArray->csn_snapshot_xmin after we released + * CSNSnapshotXidMapLock. Since we gather not xmin but oldestXmin, it + * never goes backwards regardless of how slow we can do that. + */ + Assert(TransactionIdFollowsOrEquals(oldest_deferred_xmin, + ProcArrayGetCSNSnapshotXmin())); + ProcArraySetCSNSnapshotXmin(oldest_deferred_xmin); +} + + +/* + * CSNSnapshotToXmin + * + * Get oldestXmin that took place when snapshot_csn was taken. + */ +TransactionId +CSNSnapshotToXmin(CSN snapshot_csn) +{ + TransactionId xmin; + CSN csn_seconds; + volatile CSN last_csn_seconds; + + /* Callers should check config values */ + Assert(csn_snapshot_defer_time > 0); + Assert(csnXidMap != NULL); + + /* Round down to get conservative estimates */ + csn_seconds = (snapshot_csn / NSECS_PER_SEC); + + LWLockAcquire(CSNSnapshotXidMapLock, LW_SHARED); + last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds); + if (csn_seconds > last_csn_seconds) + { + /* we don't have entry for this snapshot_csn yet, return latest known */ + xmin = csnXidMap->xmin_by_second[csnXidMap->head]; + } + else if (last_csn_seconds - csn_seconds < csnXidMap->size) + { + /* we are good, retrieve value from our map */ + Assert(last_csn_seconds % csnXidMap->size == csnXidMap->head); + xmin = csnXidMap->xmin_by_second[csn_seconds % csnXidMap->size]; + } + else + { + /* requested snapshot_csn is too old, let caller know */ + xmin = InvalidTransactionId; + } + LWLockRelease(CSNSnapshotXidMapLock); + + return xmin; +} + +/* + * XidInvisibleInCSNSnapshot + * + * Version of XidInMVCCSnapshot for transactions. For non-imported + * csn snapshots this should give same results as XidInLocalMVCCSnapshot + * (except that aborts will be shown as invisible without going to clog) and to + * ensure such behaviour XidInMVCCSnapshot is coated with asserts that checks + * identicalness of XidInvisibleInCSNSnapshot/XidInLocalMVCCSnapshot in + * case of ordinary snapshot. + */ +bool +XidInvisibleInCSNSnapshot(TransactionId xid, Snapshot snapshot) +{ + CSN csn; + + Assert(get_csnlog_status()); + + csn = TransactionIdGetCSN(xid); + + if (CSNIsNormal(csn)) + return (csn < snapshot->csn) ? false : true; + else if (CSNIsFrozen(csn)) + /* It is bootstrap or frozen transaction */ + return false; + else if (CSNIsUnclear(csn)) + /* + * Some xid can not figure out csn because of snapshot switch, + * and we can follow xid-base result. + */ + return true; + else + { + /* It is aborted or in-progress */ + Assert(CSNIsAborted(csn) || CSNIsInProgress(csn)); + if (CSNIsAborted(csn)) + Assert(TransactionIdDidAbort(xid)); + return true; + } +} + + +/***************************************************************************** + * Functions to handle transactions commit. + * + * For local transactions CSNSnapshotPrecommit sets InDoubt state before + * ProcArrayEndTransaction is called and transaction data potetntially becomes + * visible to other backends. ProcArrayEndTransaction (or ProcArrayRemove in + * twophase case) then acquires csn under ProcArray lock and stores it + * in proc->assignedCSN. It's important that csn for commit is + * generated under ProcArray lock, otherwise snapshots won't + * be equivalent. Consequent call to CSNSnapshotCommit will write + * proc->assignedCSN to CSNLog. + * + * + * CSNSnapshotAbort is slightly different comparing to commit because abort + * can skip InDoubt phase and can be called for transaction subtree. + *****************************************************************************/ + + +/* + * CSNSnapshotAbort + * + * Abort transaction in CsnLog. We can skip InDoubt state for aborts + * since no concurrent transactions allowed to see aborted data anyway. + */ +void +CSNSnapshotAbort(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + if (!get_csnlog_status()) + return; + + CSNLogSetCSN(xid, nsubxids, subxids, AbortedCSN, true); + + /* + * Clean assignedCSN anyway, as it was possibly set in + * XidSnapshotAssignCsnCurrent. + */ + pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN); +} + +/* + * CSNSnapshotPrecommit + * + * Set InDoubt status for local transaction that we are going to commit. + * This step is needed to achieve consistency between local snapshots and + * csn-based snapshots. We don't hold ProcArray lock while writing + * csn for transaction in SLRU but instead we set InDoubt status before + * transaction is deleted from ProcArray so the readers who will read csn + * in the gap between ProcArray removal and CSN assignment can wait + * until CSN is finally assigned. See also TransactionIdGetCSN(). + * + * This should be called only from parallel group leader before backend is + * deleted from ProcArray. + */ +void +CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + CSN oldassignedCSN = InProgressCSN; + bool in_progress; + + if (!get_csnlog_status()) + return; + + /* Set InDoubt status if it is local transaction */ + in_progress = pg_atomic_compare_exchange_u64(&proc->assignedCSN, + &oldassignedCSN, + InDoubtCSN); + if (in_progress) + { + Assert(CSNIsInProgress(oldassignedCSN)); + CSNLogSetCSN(xid, nsubxids, + subxids, InDoubtCSN, true); + } + else + { + /* Otherwise we should have valid CSN by this time */ + Assert(CSNIsNormal(oldassignedCSN)); + Assert(CSNIsInDoubt(CSNLogGetCSNByXid(xid))); + } +} + +/* + * CSNSnapshotCommit + * + * Write CSN that were acquired earlier to CsnLog. Should be + * preceded by CSNSnapshotPrecommit() so readers can wait until we finally + * finished writing to SLRU. + * + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks, so that TransactionIdGetCSN can wait on this + * lock for CSN. + */ +void +CSNSnapshotCommit(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + volatile CSN assignedCSN; + + if (!get_csnlog_status()) + return; + + if (!TransactionIdIsValid(xid)) + { + assignedCSN = pg_atomic_read_u64(&proc->assignedCSN); + Assert(CSNIsInProgress(assignedCSN)); + return; + } + + /* Finally write resulting CSN in SLRU */ + assignedCSN = pg_atomic_read_u64(&proc->assignedCSN); + Assert(CSNIsNormal(assignedCSN)); + CSNLogSetCSN(xid, nsubxids, subxids, assignedCSN, true); + + /* Reset for next transaction */ + pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN); +} diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 89335b64a2..6b91953193 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -77,6 +77,8 @@ #include #include "access/commit_ts.h" +#include "access/csn_log.h" +#include "access/csn_snapshot.h" #include "access/htup_details.h" #include "access/subtrans.h" #include "access/transam.h" @@ -1483,8 +1485,34 @@ FinishPreparedTransaction(const char *gid, bool isCommit) hdr->nabortrels, abortrels, gid); + /* + * CSNSnapshot callbacks that should be called right before we are + * going to become visible. Details in comments to this functions. + */ + if (isCommit) + CSNSnapshotPrecommit(proc, xid, hdr->nsubxacts, children); + else + CSNSnapshotAbort(proc, xid, hdr->nsubxacts, children); + + ProcArrayRemove(proc, latestXid); + /* + * Stamp our transaction with CSN in CSNLog. + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks, since TransactionIdGetCSN relies on + * XactLockTableWait to await csn. + */ + if (isCommit) + { + CSNSnapshotCommit(proc, xid, hdr->nsubxacts, children); + } + else + { + Assert(CSNIsInProgress( + pg_atomic_read_u64(&proc->assignedCSN))); + } + /* * In case we fail while running the callbacks, mark the gxact invalid so * no one else will try to commit/rollback, and so it will be recycled if @@ -2458,3 +2486,133 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning) RemoveTwoPhaseFile(xid, giveWarning); RemoveGXact(gxact); } + +/* + * CSNSnapshotPrepareTwophase + * + * Set InDoubt state for currently active transaction and return commit's + * global snapshot. + */ +static CSN +CSNSnapshotPrepareTwophase(const char *gid) +{ + GlobalTransaction gxact; + PGPROC *proc; + char *buf; + TransactionId xid; + xl_xact_parsed_prepare parsed; + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + /* + * Validate the GID, and lock the GXACT to ensure that two backends do not + * try to access the same GID at once. + */ + gxact = LockGXact(gid, GetUserId()); + proc = &ProcGlobal->allProcs[gxact->pgprocno]; + xid = proc->xid; + + if (gxact->ondisk) + buf = ReadTwoPhaseFile(xid, true); + else + XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); + + ParsePrepareRecord(0, (xl_xact_prepare *)buf, &parsed); + + CSNLogSetCSN(xid, parsed.nsubxacts, + parsed.subxacts, InDoubtCSN, true); + + /* Unlock our GXACT */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + gxact->locking_backend = InvalidBackendId; + LWLockRelease(TwoPhaseStateLock); + + pfree(buf); + + return GenerateCSN(InvalidCSN); +} + +/* + * CSNSnapshotAssignTwoPhase + * + * Asign CSN for currently active transaction. CSN is supposedly + * maximal among of values returned by CSNSnapshotPrepareCurrent and + * pg_csn_snapshot_prepare. + * + * This function is a counterpart of CSNSnapshotAssignCurrent() for + * twophase transactions. + */ +static void +CSNSnapshotAssignTwoPhase(const char *gid, CSN csn) +{ + GlobalTransaction gxact; + PGPROC *proc; + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (!CSNIsNormal(csn)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("pg_csn_snapshot_assign expects normal csn"))); + + /* + * Validate the GID, and lock the GXACT to ensure that two backends do not + * try to access the same GID at once. + */ + gxact = LockGXact(gid, GetUserId()); + proc = &ProcGlobal->allProcs[gxact->pgprocno]; + + Assert(csn != InvalidCSN); + /* We do not care the Generate result, we just want to make sure max + * csnState->last_max_csn value. + */ + GenerateCSN(csn); + /* Set csn and defuse ProcArrayRemove from assigning one. */ + pg_atomic_write_u64(&proc->assignedCSN, csn); + + /* Unlock our GXACT */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + gxact->locking_backend = InvalidBackendId; + LWLockRelease(TwoPhaseStateLock); +} + +/* + * SQL interface to CSNSnapshotPrepareTwophase() + * + * TODO: Rewrite this as PREPARE TRANSACTION 'gid' RETURNING SNAPSHOT + */ +Datum +pg_csn_snapshot_prepare(PG_FUNCTION_ARGS) +{ + const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0)); + CSN csn; + + csn = CSNSnapshotPrepareTwophase(gid); + + PG_RETURN_INT64(csn); +} + +/* + * SQL interface to CSNSnapshotAssignTwoPhase() + * + * TODO: Rewrite this as COMMIT PREPARED 'gid' SNAPSHOT 'csn' + */ +Datum +pg_csn_snapshot_assign(PG_FUNCTION_ARGS) +{ + const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0)); + CSN csn = PG_GETARG_INT64(1); + + CSNSnapshotAssignTwoPhase(gid, csn); + PG_RETURN_VOID(); +} \ No newline at end of file diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 6395a9b240..deabedbe37 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -21,6 +21,7 @@ #include #include "access/commit_ts.h" +#include "access/csn_snapshot.h" #include "access/multixact.h" #include "access/parallel.h" #include "access/subtrans.h" @@ -1486,6 +1487,14 @@ RecordTransactionCommit(void) /* Reset XactLastRecEnd until the next transaction writes something */ XactLastRecEnd = 0; + + /* + * Mark our transaction as InDoubt in CsnLog and get ready for + * commit. + */ + if (markXidCommitted) + CSNSnapshotPrecommit(MyProc, xid, nchildren, children); + cleanup: /* Clean up local data */ if (rels) @@ -1747,6 +1756,11 @@ RecordTransactionAbort(bool isSubXact) */ TransactionIdAbortTree(xid, nchildren, children); + /* + * Mark our transaction as Aborted in CsnLog. + */ + CSNSnapshotAbort(MyProc, xid, nchildren, children); + END_CRIT_SECTION(); /* Compute latestXid while we have the child XIDs handy */ @@ -2237,6 +2251,21 @@ CommitTransaction(void) */ ProcArrayEndTransaction(MyProc, latestXid); + /* + * Stamp our transaction with CSN in CsnLog. + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks. + */ + if (!is_parallel_worker) + { + TransactionId xid = GetTopTransactionIdIfAny(); + TransactionId *subxids; + int nsubxids; + + nsubxids = xactGetCommittedChildren(&subxids); + CSNSnapshotCommit(MyProc, xid, nsubxids, subxids); + } + /* * This is all post-commit cleanup. Note that if an error is raised here, * it's too late to abort the transaction. This should be just diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 46f016dce7..200c387084 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -4652,6 +4652,7 @@ InitControlFile(uint64 sysidentifier) ControlFile->track_commit_timestamp = track_commit_timestamp; ControlFile->enable_csn_snapshot = enable_csn_snapshot; ControlFile->data_checksum_version = bootstrap_data_checksum_version; + ControlFile->xmin_for_csn = InvalidTransactionId; } static void @@ -7211,6 +7212,7 @@ StartupXLOG(void) * maintained during recovery and need not be started yet. */ StartupSUBTRANS(oldestActiveXID); + CSNSnapshotStartup(oldestActiveXID); /* * If we're beginning at a shutdown checkpoint, we know that @@ -8030,7 +8032,10 @@ StartupXLOG(void) * timestamps are started below, if necessary.) */ if (standbyState == STANDBY_DISABLED) + { StartupSUBTRANS(oldestActiveXID); + CSNSnapshotStartup(oldestActiveXID); + } /* * Perform end of recovery actions for any SLRUs that need it. @@ -9925,6 +9930,8 @@ XLogRestorePoint(const char *rpName) static void XLogReportParameters(void) { + TransactionId xmin_for_csn = ControlFile->xmin_for_csn; + if (wal_level != ControlFile->wal_level || wal_log_hints != ControlFile->wal_log_hints || MaxConnections != ControlFile->MaxConnections || @@ -9964,11 +9971,12 @@ XLogReportParameters(void) XLogFlush(recptr); } - LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); - - if (enable_csn_snapshot != ControlFile->enable_csn_snapshot) - prepare_csn_env(enable_csn_snapshot); + prepare_csn_env(enable_csn_snapshot, + enable_csn_snapshot == ControlFile->enable_csn_snapshot, + &xmin_for_csn); + LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); + ControlFile->xmin_for_csn = xmin_for_csn; ControlFile->MaxConnections = MaxConnections; ControlFile->max_worker_processes = max_worker_processes; ControlFile->max_wal_senders = max_wal_senders; @@ -9982,6 +9990,16 @@ XLogReportParameters(void) LWLockRelease(ControlFileLock); } + else + { + /* + * When no GUC change, but for xmin_for_csn it should transmit the xmin_for_csn + * from pg_control to csnState->xmin_for_csn. Or it will cause issue when prepare + * transaction exists and with 'xid-snapshot start -> csn-snapshot start -> + * csn-snapshot start' sequence. + */ + prepare_csn_env(enable_csn_snapshot, true, &xmin_for_csn); + } } /* diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index e06bc13841..bbc5f1adf7 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -17,6 +17,7 @@ #include "access/clog.h" #include "access/commit_ts.h" #include "access/csn_log.h" +#include "access/csn_snapshot.h" #include "access/heapam.h" #include "access/multixact.h" #include "access/nbtree.h" @@ -152,6 +153,7 @@ CreateSharedMemoryAndSemaphores(void) size = add_size(size, BTreeShmemSize()); size = add_size(size, SyncScanShmemSize()); size = add_size(size, AsyncShmemSize()); + size = add_size(size, CSNSnapshotShmemSize()); #ifdef EXEC_BACKEND size = add_size(size, ShmemBackendArraySize()); #endif @@ -272,6 +274,7 @@ CreateSharedMemoryAndSemaphores(void) BTreeShmemInit(); SyncScanShmemInit(); AsyncShmemInit(); + CSNSnapshotShmemInit(); #ifdef EXEC_BACKEND diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 4a200b0815..62138fddcc 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -49,6 +49,7 @@ #include "access/clog.h" #include "access/csn_log.h" +#include "access/csn_snapshot.h" #include "access/subtrans.h" #include "access/transam.h" #include "access/twophase.h" @@ -98,6 +99,9 @@ typedef struct ProcArrayStruct /* oldest catalog xmin of any replication slot */ TransactionId replication_slot_catalog_xmin; + /* xmin of oldest active csn snapshot */ + TransactionId csn_snapshot_xmin; + /* indexes into allProcs[], has PROCARRAY_MAXPROCS entries */ int pgprocnos[FLEXIBLE_ARRAY_MEMBER]; } ProcArrayStruct; @@ -419,6 +423,7 @@ CreateSharedProcArray(void) procArray->lastOverflowedXid = InvalidTransactionId; procArray->replication_slot_xmin = InvalidTransactionId; procArray->replication_slot_catalog_xmin = InvalidTransactionId; + procArray->csn_snapshot_xmin = InvalidTransactionId; ShmemVariableCache->xactCompletionCount = 1; } @@ -547,6 +552,14 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid) /* Advance global latestCompletedXid while holding the lock */ MaintainLatestCompletedXid(latestXid); + /* + * Assign xid csn while holding ProcArrayLock for non-distributed + * COMMIT PREPARED. After lock is released consequent + * CSNSnapshotCommit() will write this value to CsnLog. + */ + if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN))) + pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(InvalidCSN)); + /* Same with xactCompletionCount */ ShmemVariableCache->xactCompletionCount++; @@ -656,6 +669,7 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid) proc->lxid = InvalidLocalTransactionId; proc->xmin = InvalidTransactionId; + proc->originalXmin = InvalidTransactionId; proc->delayChkpt = false; /* be sure this is cleared in abort */ proc->recoveryConflictPending = false; @@ -695,6 +709,7 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid) proc->xid = InvalidTransactionId; proc->lxid = InvalidLocalTransactionId; proc->xmin = InvalidTransactionId; + proc->originalXmin = InvalidTransactionId; proc->delayChkpt = false; /* be sure this is cleared in abort */ proc->recoveryConflictPending = false; @@ -720,6 +735,16 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid) /* Also advance global latestCompletedXid while holding the lock */ MaintainLatestCompletedXid(latestXid); + /* + * Assign xid csn while holding ProcArrayLock for + * COMMIT. + * + * TODO: in case of group commit we can generate one CSNSnapshot for + * whole group to save time on timestamp aquisition. + */ + if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN))) + pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(InvalidCSN)); + /* Same with xactCompletionCount */ ShmemVariableCache->xactCompletionCount++; } @@ -878,6 +903,7 @@ ProcArrayClearTransaction(PGPROC *proc) proc->lxid = InvalidLocalTransactionId; proc->xmin = InvalidTransactionId; + proc->originalXmin = InvalidTransactionId; proc->recoveryConflictPending = false; Assert(!(proc->statusFlags & PROC_VACUUM_STATE_MASK)); @@ -1667,6 +1693,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) TransactionId kaxmin; bool in_recovery = RecoveryInProgress(); TransactionId *other_xids = ProcGlobal->xids; + TransactionId csn_snapshot_xmin = InvalidTransactionId; LWLockAcquire(ProcArrayLock, LW_SHARED); @@ -1715,6 +1742,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) */ h->slot_xmin = procArray->replication_slot_xmin; h->slot_catalog_xmin = procArray->replication_slot_catalog_xmin; + csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin(); for (int index = 0; index < arrayP->numProcs; index++) { @@ -1723,6 +1751,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) int8 statusFlags = ProcGlobal->statusFlags[index]; TransactionId xid; TransactionId xmin; + TransactionId original_xmin = UINT32_ACCESS_ONCE(proc->originalXmin); /* Fetch xid just once - see GetNewTransactionId */ xid = UINT32_ACCESS_ONCE(other_xids[index]); @@ -1735,6 +1764,9 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) * (yet) an Xid; conversely, if it has an Xid, that could determine * some not-yet-set Xmin. */ + if (TransactionIdIsValid(original_xmin)) + xmin = original_xmin; + xmin = TransactionIdOlder(xmin, xid); /* if neither is set, this proc doesn't influence the horizon */ @@ -1854,6 +1886,10 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) /* defer doesn't apply to temp relations */ } + if (TransactionIdIsValid(csn_snapshot_xmin) && + TransactionIdOlder(csn_snapshot_xmin, h->shared_oldest_nonremovable)) + h->shared_oldest_nonremovable = csn_snapshot_xmin; + /* * Check whether there are replication slots requiring an older xmin. */ @@ -2064,6 +2100,8 @@ GetSnapshotDataReuse(Snapshot snapshot) if (curXactCompletionCount != snapshot->snapXactCompletionCount) return false; + if (get_csnlog_status()) + return false; /* * If the current xactCompletionCount is still the same as it was at the * time the snapshot was built, we can be sure that rebuilding the @@ -2143,6 +2181,7 @@ GetSnapshotData(Snapshot snapshot) size_t count = 0; int subcount = 0; bool suboverflowed = false; + CSN csn = FrozenCSN; FullTransactionId latest_completed; TransactionId oldestxid; int mypgxactoff; @@ -2151,6 +2190,7 @@ GetSnapshotData(Snapshot snapshot) TransactionId replication_slot_xmin = InvalidTransactionId; TransactionId replication_slot_catalog_xmin = InvalidTransactionId; + TransactionId csn_snapshot_xmin = InvalidTransactionId; Assert(snapshot != NULL); @@ -2371,10 +2411,18 @@ GetSnapshotData(Snapshot snapshot) */ replication_slot_xmin = procArray->replication_slot_xmin; replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin; + csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin(); if (!TransactionIdIsValid(MyProc->xmin)) MyProc->xmin = TransactionXmin = xmin; + /* + * Take CSN under ProcArrayLock so the snapshot stays + * synchronized. + */ + if (!snapshot->takenDuringRecovery && get_csnlog_status()) + csn = GenerateCSN(InvalidCSN); + LWLockRelease(ProcArrayLock); /* maintain state for GlobalVis* */ @@ -2396,6 +2444,9 @@ GetSnapshotData(Snapshot snapshot) def_vis_xid_data = TransactionIdRetreatedBy(xmin, vacuum_defer_cleanup_age); + if (TransactionIdIsValid(csn_snapshot_xmin)) + def_vis_xid_data = TransactionIdOlder(csn_snapshot_xmin, def_vis_xid_data); + /* Check whether there's a replication slot requiring an older xmin. */ def_vis_xid_data = TransactionIdOlder(def_vis_xid_data, replication_slot_xmin); @@ -2480,6 +2531,11 @@ GetSnapshotData(Snapshot snapshot) snapshot->active_count = 0; snapshot->regd_count = 0; snapshot->copied = false; + snapshot->imported_csn = false; + snapshot->csn = csn; + + if (csn_snapshot_defer_time > 0 && IsUnderPostmaster) + CSNSnapshotMapXmin(snapshot->csn); GetSnapshotDataInitOldSnapshot(snapshot); @@ -3832,6 +3888,25 @@ ProcArrayGetReplicationSlotXmin(TransactionId *xmin, LWLockRelease(ProcArrayLock); } +/* + * ProcArraySetCSNSnapshotXmin + */ +void +ProcArraySetCSNSnapshotXmin(TransactionId xmin) +{ + /* We rely on atomic fetch/store of xid */ + procArray->csn_snapshot_xmin = xmin; +} + +/* + * ProcArrayGetCSNSnapshotXmin + */ +TransactionId +ProcArrayGetCSNSnapshotXmin(void) +{ + return procArray->csn_snapshot_xmin; +} + /* * XidCacheRemoveRunningXids * diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt index 56196b2d0d..ac0e6a33fe 100644 --- a/src/backend/storage/lmgr/lwlocknames.txt +++ b/src/backend/storage/lmgr/lwlocknames.txt @@ -54,3 +54,4 @@ XactTruncationLock 44 WrapLimitsVacuumLock 46 NotifyQueueTailLock 47 CSNLogControlLock 48 +CSNSnapshotXidMapLock 49 diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 897045ee27..6e66801001 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -37,6 +37,7 @@ #include "access/transam.h" #include "access/twophase.h" +#include "access/csn_snapshot.h" #include "access/xact.h" #include "miscadmin.h" #include "pgstat.h" @@ -440,6 +441,9 @@ InitProcess(void) MyProc->clogGroupMemberLsn = InvalidXLogRecPtr; Assert(pg_atomic_read_u32(&MyProc->clogGroupNext) == INVALID_PGPROCNO); + MyProc->originalXmin = InvalidTransactionId; + pg_atomic_init_u64(&MyProc->assignedCSN, InProgressCSN); + /* * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch * on it. That allows us to repoint the process latch, which so far @@ -582,6 +586,7 @@ InitAuxiliaryProcess(void) MyProc->lwWaitMode = 0; MyProc->waitLock = NULL; MyProc->waitProcLock = NULL; + MyProc->originalXmin = InvalidTransactionId; pg_atomic_write_u64(&MyProc->waitStart, 0); #ifdef USE_ASSERT_CHECKING { diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 14063bcf9e..cc01d72d64 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -30,6 +30,7 @@ #include #include "access/commit_ts.h" +#include "access/csn_snapshot.h" #include "access/gin.h" #include "access/rmgr.h" #include "access/tableam.h" @@ -3140,6 +3141,26 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"csn_snapshot_defer_time", PGC_POSTMASTER, PRESET_OPTIONS, + gettext_noop("Minimal age of records which allowed to be vacuumed, in seconds."), + NULL + }, + &csn_snapshot_defer_time, + 5, 0, INT_MAX, + NULL, NULL, NULL + }, + + { + {"csn_time_shift", PGC_USERSET, RESOURCES_MEM, + gettext_noop("Do the time shift in the CSN generator."), + gettext_noop("Used for debug purposes.") + }, + &csn_time_shift, + 0, INT_MIN, INT_MAX, + NULL, NULL, NULL + }, + { {"block_size", PGC_INTERNAL, PRESET_OPTIONS, gettext_noop("Shows the size of a disk block."), diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 95704265b6..2d4c5f53af 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -48,6 +48,7 @@ #include #include +#include "access/csn_log.h" #include "access/subtrans.h" #include "access/transam.h" #include "access/xact.h" @@ -173,6 +174,7 @@ static TimestampTz AlignTimestampToMinuteBoundary(TimestampTz ts); static Snapshot CopySnapshot(Snapshot snapshot); static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); +static bool XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot); /* * Snapshot fields to be serialized. @@ -191,6 +193,8 @@ typedef struct SerializedSnapshotData CommandId curcid; TimestampTz whenTaken; XLogRecPtr lsn; + CSN csn; + bool imported_csn; } SerializedSnapshotData; Size @@ -291,6 +295,7 @@ GetTransactionSnapshot(void) CurrentSnapshot = GetSerializableTransactionSnapshot(&CurrentSnapshotData); else CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData); + /* Make a saved copy */ CurrentSnapshot = CopySnapshot(CurrentSnapshot); FirstXactSnapshot = CurrentSnapshot; @@ -2115,6 +2120,8 @@ SerializeSnapshot(Snapshot snapshot, char *start_address) serialized_snapshot.curcid = snapshot->curcid; serialized_snapshot.whenTaken = snapshot->whenTaken; serialized_snapshot.lsn = snapshot->lsn; + serialized_snapshot.csn = snapshot->csn; + serialized_snapshot.imported_csn = snapshot->imported_csn; /* * Ignore the SubXID array if it has overflowed, unless the snapshot was @@ -2190,6 +2197,8 @@ RestoreSnapshot(char *start_address) snapshot->whenTaken = serialized_snapshot.whenTaken; snapshot->lsn = serialized_snapshot.lsn; snapshot->snapXactCompletionCount = 0; + snapshot->csn = serialized_snapshot.csn; + snapshot->imported_csn = serialized_snapshot.imported_csn; /* Copy XIDs, if present. */ if (serialized_snapshot.xcnt > 0) @@ -2230,6 +2239,44 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc) /* * XidInMVCCSnapshot + * + * Check whether this xid is in snapshot. When enable_csn_snapshot is + * switched off just call XidInLocalMVCCSnapshot(). + */ +bool +XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +{ + bool in_snapshot; + + if (snapshot->imported_csn) + { + Assert(enable_csn_snapshot); + /* No point to using snapshot info except CSN */ + return XidInvisibleInCSNSnapshot(xid, snapshot); + } + + in_snapshot = XidInLocalMVCCSnapshot(xid, snapshot); + + if (!get_csnlog_status()) + { + Assert(CSNIsFrozen(snapshot->csn)); + return in_snapshot; + } + + if (in_snapshot) + { + /* + * This xid may be already in unknown state and in that case + * we must wait and recheck. + */ + return XidInvisibleInCSNSnapshot(xid, snapshot); + } + else + return false; +} + +/* + * XidInLocalMVCCSnapshot * Is the given XID still-in-progress according to the snapshot? * * Note: GetSnapshotData never stores either top xid or subxids of our own @@ -2238,8 +2285,8 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc) * TransactionIdIsCurrentTransactionId first, except when it's known the * XID could not be ours anyway. */ -bool -XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +static bool +XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot) { uint32 i; @@ -2349,3 +2396,98 @@ XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) return false; } + +/* + * ExportCSNSnapshot + * + * Export csn so that caller can expand this transaction to other + * nodes. + * + * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and + * add some additional checks that transaction did not yet acquired xid, but + * for current iteration of this patch I don't want to hack on parser. + */ +CSN +ExportCSNSnapshot() +{ + if (!get_csnlog_status()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not export csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + return CurrentSnapshot->csn; +} + +/* SQL accessor to ExportCSNSnapshot() */ +Datum +pg_csn_snapshot_export(PG_FUNCTION_ARGS) +{ + CSN export_csn = ExportCSNSnapshot(); + PG_RETURN_UINT64(export_csn); +} + +/* + * ImportCSNSnapshot + * + * Import csn and retract this backends xmin to the value that was + * actual when we had such csn. + * + * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and + * add some additional checks that transaction did not yet acquired xid, but + * for current iteration of this patch I don't want to hack on parser. + */ +void +ImportCSNSnapshot(CSN csn) +{ + volatile TransactionId xmin; + + if (!get_csnlog_status()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not import csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (csn_snapshot_defer_time <= 0) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not import csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is positive.", + "csn_snapshot_defer_time"))); + + /* + * Call CSNSnapshotToXmin under ProcArrayLock to avoid situation that + * resulting xmin will be evicted from map before we will set it into our + * backend's xmin. + */ + LWLockAcquire(ProcArrayLock, LW_SHARED); + xmin = CSNSnapshotToXmin(csn); + + if (!TransactionIdIsValid(xmin)) + { + LWLockRelease(ProcArrayLock); + elog(ERROR, "CSNSnapshotToXmin: csn snapshot too old: %lu", csn); + } + + MyProc->originalXmin = MyProc->xmin; + MyProc->xmin = TransactionXmin = xmin; + LWLockRelease(ProcArrayLock); + + CurrentSnapshot->xmin = xmin; /* defuse SnapshotResetXmin() */ + CurrentSnapshot->csn = csn; + CurrentSnapshot->imported_csn = true; + + CSNSnapshotSync(csn); +} + +/* SQL accessor to ImportCSNSnapshot() */ +Datum +pg_csn_snapshot_import(PG_FUNCTION_ARGS) +{ + CSN csn = PG_GETARG_UINT64(0); + + ImportCSNSnapshot(csn); + PG_RETURN_VOID(); +} diff --git a/src/bin/pg_waldump/csnlogdesc.c b/src/bin/pg_waldump/csnlogdesc.c new file mode 120000 index 0000000000..dcde44b3ee --- /dev/null +++ b/src/bin/pg_waldump/csnlogdesc.c @@ -0,0 +1 @@ +../../../src/backend/access/rmgrdesc/csnlogdesc.c \ No newline at end of file diff --git a/src/include/access/csn_log.h b/src/include/access/csn_log.h index 01fb0daad4..9857724dca 100644 --- a/src/include/access/csn_log.h +++ b/src/include/access/csn_log.h @@ -71,7 +71,8 @@ extern void WriteAssignCSNXlogRec(CSN csn); extern CSN set_last_max_csn(CSN csn, bool force); extern CSN get_last_max_csn(void); extern CSN get_xmin_for_csn(void); -extern void prepare_csn_env(bool enable_csn_snapshot); +extern void prepare_csn_env(bool enable_csn_snapshot, bool same, + TransactionId *xmin_for_csn_in_control); extern void CatchCSNLog(void); extern void ActivateCSNlog(void); extern void DeactivateCSNlog(void); diff --git a/src/include/access/csn_snapshot.h b/src/include/access/csn_snapshot.h new file mode 100644 index 0000000000..3a739e6744 --- /dev/null +++ b/src/include/access/csn_snapshot.h @@ -0,0 +1,65 @@ +/*------------------------------------------------------------------------- + * + * csn_snapshot.h + * Support for cross-node snapshot isolation. + * + * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/access/csn_snapshot.h + * + *------------------------------------------------------------------------- + */ +#ifndef CSN_SNAPSHOT_H +#define CSN_SNAPSHOT_H + +#include "port/atomics.h" +#include "storage/lock.h" +#include "utils/snapshot.h" +#include "utils/guc.h" + +/* + * snapshot.h is used in frontend code so atomic variant of CSN type + * is defined here. + */ +typedef pg_atomic_uint64 CSN_atomic; + +#define InProgressCSN UINT64CONST(0x0) +#define AbortedCSN UINT64CONST(0x1) +#define FrozenCSN UINT64CONST(0x2) +#define InDoubtCSN UINT64CONST(0x3) +#define UnclearCSN UINT64CONST(0x4) +#define FirstNormalCSN UINT64CONST(0x5) + +#define CSNIsInProgress(csn) ((csn) == InProgressCSN) +#define CSNIsAborted(csn) ((csn) == AbortedCSN) +#define CSNIsFrozen(csn) ((csn) == FrozenCSN) +#define CSNIsInDoubt(csn) ((csn) == InDoubtCSN) +#define CSNIsUnclear(csn) ((csn) == UnclearCSN) +#define CSNIsNormal(csn) ((csn) >= FirstNormalCSN) + + +extern int csn_snapshot_defer_time; +extern int csn_time_shift; + + +extern Size CSNSnapshotShmemSize(void); +extern void CSNSnapshotShmemInit(void); + +extern void CSNSnapshotStartup(TransactionId oldestActiveXID); +extern void CSNSnapshotMapXmin(CSN snapshot_csn); +extern TransactionId CSNSnapshotToXmin(CSN snapshot_csn); +extern CSN GenerateCSN(CSN assign); +extern bool XidInvisibleInCSNSnapshot(TransactionId xid, Snapshot snapshot); +extern CSN TransactionIdGetCSN(TransactionId xid); +extern void CSNSnapshotAbort(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotCommit(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotAssignCurrent(CSN snapshot_csn); +extern CSN CSNSnapshotPrepareCurrent(void); +extern void CSNSnapshotSync(CSN remote_csn); + +#endif /* CSN_SNAPSHOT_H */ diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h index 194d100ed4..fbd63f50d9 100644 --- a/src/include/catalog/pg_control.h +++ b/src/include/catalog/pg_control.h @@ -185,6 +185,12 @@ typedef struct ControlFileData bool track_commit_timestamp; bool enable_csn_snapshot; + /* + * Used to record a xmin when database startup with a snapshot-switch to csn snapshot, + * and will hold the value until it switch to xid-snapshot. + */ + TransactionId xmin_for_csn; + /* * This data is used to check for hardware-architecture compatibility of * the database and the backend executable. We need not check endianness diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index b9f4afba05..8585464c5b 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11411,4 +11411,18 @@ proname => 'is_normalized', prorettype => 'bool', proargtypes => 'text text', prosrc => 'unicode_is_normalized' }, +# csn shnapshot handling +{ oid => '10001', descr => 'export csn snapshot', + proname => 'pg_csn_snapshot_export', provolatile => 'v', proparallel => 'u', + prorettype => 'int8', proargtypes => '', prosrc => 'pg_csn_snapshot_export' }, +{ oid => '10002', descr => 'import csn snapshot', + proname => 'pg_csn_snapshot_import', provolatile => 'v', proparallel => 'u', + prorettype => 'void', proargtypes => 'int8', prosrc => 'pg_csn_snapshot_import' }, +{ oid => '10003', descr => 'prepare distributed transaction for commit, get csn', + proname => 'pg_csn_snapshot_prepare', provolatile => 'v', proparallel => 'u', + prorettype => 'int8', proargtypes => 'text', prosrc => 'pg_csn_snapshot_prepare' }, +{ oid => '10004', descr => 'assign csn to distributed transaction', + proname => 'pg_csn_snapshot_assign', provolatile => 'v', proparallel => 'u', + prorettype => 'void', proargtypes => 'text int8', prosrc => 'pg_csn_snapshot_assign' }, + ] diff --git a/src/include/datatype/timestamp.h b/src/include/datatype/timestamp.h index 99873497a6..d8a681aaa4 100644 --- a/src/include/datatype/timestamp.h +++ b/src/include/datatype/timestamp.h @@ -92,6 +92,8 @@ typedef struct #define USECS_PER_HOUR INT64CONST(3600000000) #define USECS_PER_MINUTE INT64CONST(60000000) #define USECS_PER_SEC INT64CONST(1000000) +#define NSECS_PER_SEC INT64CONST(1000000000) +#define NSECS_PER_USEC INT64CONST(1000) /* * We allow numeric timezone offsets up to 15:59:59 either way from Greenwich. diff --git a/src/include/fmgr.h b/src/include/fmgr.h index ab7b85c86e..f08999740b 100644 --- a/src/include/fmgr.h +++ b/src/include/fmgr.h @@ -281,6 +281,7 @@ extern struct varlena *pg_detoast_datum_packed(struct varlena *datum); #define PG_GETARG_FLOAT4(n) DatumGetFloat4(PG_GETARG_DATUM(n)) #define PG_GETARG_FLOAT8(n) DatumGetFloat8(PG_GETARG_DATUM(n)) #define PG_GETARG_INT64(n) DatumGetInt64(PG_GETARG_DATUM(n)) +#define PG_GETARG_UINT64(n) DatumGetUInt64(PG_GETARG_DATUM(n)) /* use this if you want the raw, possibly-toasted input datum: */ #define PG_GETARG_RAW_VARLENA_P(n) ((struct varlena *) PG_GETARG_POINTER(n)) /* use this if you want the input datum de-toasted: */ diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index f0e920f462..6edb492638 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -15,6 +15,7 @@ #define _PROC_H_ #include "access/clog.h" +#include "access/csn_snapshot.h" #include "access/xlogdefs.h" #include "lib/ilist.h" #include "utils/snapshot.h" @@ -252,6 +253,18 @@ struct PGPROC PGPROC *lockGroupLeader; /* lock group leader, if I'm a member */ dlist_head lockGroupMembers; /* list of members, if I'm a leader */ dlist_node lockGroupLink; /* my member link, if I'm a member */ + + /* + * assignedCSN holds CSN for this transaction. It is generated + * under a ProcArray lock and later is writter to a CSNLog. This + * variable defined as atomic only for case of group commit, in all other + * scenarios only backend responsible for this proc entry is working with + * this variable. + */ + CSN_atomic assignedCSN; + + /* Original xmin of this backend before csn snapshot was imported */ + TransactionId originalXmin; }; /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */ diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index b01fa52139..755a0743b6 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -94,4 +94,7 @@ extern void ProcArraySetReplicationSlotXmin(TransactionId xmin, extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin, TransactionId *catalog_xmin); +extern void ProcArraySetCSNSnapshotXmin(TransactionId xmin); +extern TransactionId ProcArrayGetCSNSnapshotXmin(void); + #endif /* PROCARRAY_H */ diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index f66ac58188..c339eb7384 100644 --- a/src/include/utils/snapmgr.h +++ b/src/include/utils/snapmgr.h @@ -129,6 +129,8 @@ extern void AtSubCommit_Snapshot(int level); extern void AtSubAbort_Snapshot(int level); extern void AtEOXact_Snapshot(bool isCommit, bool resetXmin); +extern CSN ExportCSNSnapshot(void); +extern void ImportCSNSnapshot(CSN csn); extern void ImportSnapshot(const char *idstr); extern bool XactHasExportedSnapshots(void); extern void DeleteAllExportedSnapshotFiles(void); diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h index 92575a207f..98cff1d872 100644 --- a/src/include/utils/snapshot.h +++ b/src/include/utils/snapshot.h @@ -217,7 +217,8 @@ typedef struct SnapshotData * CSN for snapshot isolation support. * Will be used only if enable_csn_snapshot is enabled. */ - CSN csn; + CSN csn; + bool imported_csn; /* * The transaction completion count at the time GetSnapshotData() built -- 2.25.1