From: Andrey V. Lepikhov <a.lepikhov@postgrespro.ru>
To: tsunakawa.takay@fujitsu.com <tsunakawa.takay@fujitsu.com>
Cc: Fujii Masao <masao.fujii@oss.nttdata.com>
Cc: Masahiko Sawada <masahiko.sawada@2ndquadrant.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>
Cc: movead.li@highgo.ca <movead.li@highgo.ca>
Cc: PostgreSQL-Dev <pgsql-hackers@postgresql.org>
Subject: Re: Global snapshots
Date: Tue, 23 Mar 2021 12:54:57 +0500
Message-ID: <0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru> (raw)
In-Reply-To: <TYAPR01MB299079A57E722E71869DBE0CFE9D9@TYAPR01MB2990.jpnprd01.prod.outlook.com>
References: <07b2c899-4ed0-4c87-1327-23c750311248@postgrespro.ru>
<806c3708-26ea-b94d-87b1-d48e41e93726@postgrespro.ru>
<TYAPR01MB2990F6F2A75657899A7E3A30FE270@TYAPR01MB2990.jpnprd01.prod.outlook.com>
<d17da3c0-54bb-6221-7ec7-151f6be39e5f@oss.nttdata.com>
<TYAPR01MB299093D3F5E8BC2D80D5DFD6FE270@TYAPR01MB2990.jpnprd01.prod.outlook.com>
<88a2642f-1ce6-a9d2-5d26-1221a2c4e072@oss.nttdata.com>
<CAA4eK1LMQYtY1kJX5RpxkUCnrFnGqj--zcAGVH7Sp3mcPRRuvg@mail.gmail.com>
<dc6d5fe2-7bdf-1b3f-1027-e152b979c12f@oss.nttdata.com>
<CA+fd4k5d9Jjt2zwCFwm9FmFF1m6J1_ixaU_X7noM9qkDWCbAzA@mail.gmail.com>
<f59e3cf8-84d7-4a2b-218d-48ffac488566@oss.nttdata.com>
<TYAPR01MB299083299DA94BA849358BEEFE170@TYAPR01MB2990.jpnprd01.prod.outlook.com>
<TYAPR01MB2990CAC9CE3D0AA63B45053CFED50@TYAPR01MB2990.jpnprd01.prod.outlook.com>
<a0503174-9ac9-fe2c-1788-1a1337d72453@postgrespro.ru>
<TYAPR01MB299079A57E722E71869DBE0CFE9D9@TYAPR01MB2990.jpnprd01.prod.outlook.com>
Current state of the patch set rebased on master, 5aed6a1fc2.
It is development version. Here some problems with visibility still
detected in two tests:
1. CSN Snapshot module - TAP test on time skew.
2. Clock SI implementation - TAP test on emulation of bank transaction.
--
regards,
Andrey Lepikhov
Postgres Professional
Attachments:
[text/x-patch] 0001-CSN-Log.patch (44.1K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/2-0001-CSN-Log.patch)
download | inline diff:
From 875606471aaae9ee422593da08be40f5db4305dc Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Mon, 22 Mar 2021 15:56:06 +0500
Subject: [PATCH 1/4] CSN Log
---
src/backend/access/rmgrdesc/Makefile | 1 +
src/backend/access/rmgrdesc/csnlogdesc.c | 95 ++++
src/backend/access/rmgrdesc/xlogdesc.c | 6 +-
src/backend/access/transam/Makefile | 1 +
src/backend/access/transam/csn_log.c | 685 +++++++++++++++++++++++
src/backend/access/transam/rmgr.c | 1 +
src/backend/access/transam/varsup.c | 2 +
src/backend/access/transam/xlog.c | 21 +-
src/backend/commands/vacuum.c | 3 +-
src/backend/storage/ipc/ipci.c | 3 +
src/backend/storage/ipc/procarray.c | 3 +
src/backend/storage/lmgr/lwlock.c | 2 +
src/backend/storage/lmgr/lwlocknames.txt | 1 +
src/backend/storage/sync/sync.c | 5 +
src/backend/utils/misc/guc.c | 18 +
src/backend/utils/probes.d | 2 +
src/bin/initdb/initdb.c | 3 +-
src/bin/pg_controldata/pg_controldata.c | 2 +
src/bin/pg_upgrade/pg_upgrade.c | 5 +
src/bin/pg_upgrade/pg_upgrade.h | 2 +
src/bin/pg_waldump/rmgrdesc.c | 1 +
src/include/access/csn_log.h | 83 +++
src/include/access/rmgrlist.h | 1 +
src/include/access/xlog_internal.h | 1 +
src/include/catalog/pg_control.h | 3 +
src/include/storage/lwlock.h | 1 +
src/include/storage/proc.h | 1 +
src/include/storage/sync.h | 1 +
src/include/utils/snapshot.h | 11 +
src/test/regress/expected/sysviews.out | 4 +-
30 files changed, 962 insertions(+), 6 deletions(-)
create mode 100644 src/backend/access/rmgrdesc/csnlogdesc.c
create mode 100644 src/backend/access/transam/csn_log.c
create mode 100644 src/include/access/csn_log.h
diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefileindex f88d72fd86..15fc36f7b4 100644--- a/src/backend/access/rmgrdesc/Makefile+++ b/src/backend/access/rmgrdesc/Makefile@@ -11,6 +11,7 @@ include $(top_builddir)/src/Makefile.global
OBJS = \
brindesc.o \
clogdesc.o \
+ csnlogdesc.o \
committsdesc.o \
dbasedesc.o \
genericdesc.o \
diff --git a/src/backend/access/rmgrdesc/csnlogdesc.c b/src/backend/access/rmgrdesc/csnlogdesc.cnew file mode 100644index 0000000000..0f52b2986d--- /dev/null+++ b/src/backend/access/rmgrdesc/csnlogdesc.c@@ -0,0 +1,95 @@+/*-------------------------------------------------------------------------+ *+ * clogdesc.c+ * rmgr descriptor routines for access/transam/csn_log.c+ *+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group+ * Portions Copyright (c) 1994, Regents of the University of California+ *+ *+ * IDENTIFICATION+ * src/backend/access/rmgrdesc/csnlogdesc.c+ *+ *-------------------------------------------------------------------------+ */+#include "postgres.h"++#include "access/csn_log.h"+++void+csnlog_desc(StringInfo buf, XLogReaderState *record)+{+ char *rec = XLogRecGetData(record);+ uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;++ if (info == XLOG_CSN_ZEROPAGE)+ {+ int pageno;++ memcpy(&pageno, XLogRecGetData(record), sizeof(int));+ appendStringInfo(buf, "pageno %d", pageno);+ }+ else if (info == XLOG_CSN_TRUNCATE)+ {+ int pageno;++ memcpy(&pageno, XLogRecGetData(record), sizeof(int));+ appendStringInfo(buf, "pageno %d", pageno);+ }+ else if (info == XLOG_CSN_ASSIGNMENT)+ {+ CSN csn;++ memcpy(&csn, XLogRecGetData(record), sizeof(CSN));+ appendStringInfo(buf, "assign "INT64_FORMAT"", csn);+ }+ else if (info == XLOG_CSN_SETCSN)+ {+ xl_csn_set *xlrec = (xl_csn_set *) rec;+ int nsubxids;++ appendStringInfo(buf, "set "INT64_FORMAT" for: %u",+ xlrec->csn,+ xlrec->xtop);+ nsubxids = ((XLogRecGetDataLen(record) - MinSizeOfCSNSet) /+ sizeof(TransactionId));+ if (nsubxids > 0)+ {+ int i;+ TransactionId *subxids;++ subxids = palloc(sizeof(TransactionId) * nsubxids);+ memcpy(subxids,+ XLogRecGetData(record) + MinSizeOfCSNSet,+ sizeof(TransactionId) * nsubxids);+ for (i = 0; i < nsubxids; i++)+ appendStringInfo(buf, ", %u", subxids[i]);+ pfree(subxids);+ }+ }+}++const char *+csnlog_identify(uint8 info)+{+ const char *id = NULL;++ switch (info & ~XLR_INFO_MASK)+ {+ case XLOG_CSN_ASSIGNMENT:+ id = "ASSIGNMENT";+ break;+ case XLOG_CSN_SETCSN:+ id = "SETCSN";+ break;+ case XLOG_CSN_ZEROPAGE:+ id = "ZEROPAGE";+ break;+ case XLOG_CSN_TRUNCATE:+ id = "TRUNCATE";+ break;+ }++ return id;+}diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.cindex e6090a9dad..aeba6db37d 100644--- a/src/backend/access/rmgrdesc/xlogdesc.c+++ b/src/backend/access/rmgrdesc/xlogdesc.c@@ -113,7 +113,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
appendStringInfo(buf, "max_connections=%d max_worker_processes=%d "
"max_wal_senders=%d max_prepared_xacts=%d "
"max_locks_per_xact=%d wal_level=%s "
- "wal_log_hints=%s track_commit_timestamp=%s",+ "wal_log_hints=%s track_commit_timestamp=%s "+ "enable_csn_snapshot=%s",
xlrec.MaxConnections,
xlrec.max_worker_processes,
xlrec.max_wal_senders,
@@ -121,7 +122,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
xlrec.max_locks_per_xact,
wal_level_str,
xlrec.wal_log_hints ? "on" : "off",
- xlrec.track_commit_timestamp ? "on" : "off");+ xlrec.track_commit_timestamp ? "on" : "off",+ xlrec.enable_csn_snapshot ? "on" : "off");
}
else if (info == XLOG_FPW_CHANGE)
{
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefileindex 595e02de72..d1196cba87 100644--- a/src/backend/access/transam/Makefile+++ b/src/backend/access/transam/Makefile@@ -15,6 +15,7 @@ include $(top_builddir)/src/Makefile.global
OBJS = \
clog.o \
commit_ts.o \
+ csn_log.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.cnew file mode 100644index 0000000000..7a6d63e953--- /dev/null+++ b/src/backend/access/transam/csn_log.c@@ -0,0 +1,685 @@+/*-----------------------------------------------------------------------------+ *+ * csn_log.c+ * Track commit sequence numbers of finished transactions+ *+ * This module provides SLRU to store CSN for each transaction. This+ * mapping need to be kept only for xid's greater then oldestXid, but+ * that can require arbitrary large amounts of memory in case of long-lived+ * transactions. Because of same lifetime and persistancy requirements+ * this module is quite similar to subtrans.c+ *+ * If we switch database from CSN-base snapshot to xid-base snapshot then,+ * nothing wrong. But if we switch xid-base snapshot to CSN-base snapshot+ * it should decide a new xid whwich begin csn-base check. It can not be+ * oldestActiveXID because of prepared transaction.+ *+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group+ * Portions Copyright (c) 1994, Regents of the University of California+ *+ * src/backend/access/transam/csn_log.c+ *+ *-----------------------------------------------------------------------------+ */+#include "postgres.h"++#include "access/csn_log.h"+#include "access/slru.h"+#include "access/subtrans.h"+#include "access/transam.h"+#include "access/xlog_internal.h"+#include "miscadmin.h"+#include "pg_trace.h"+#include "storage/spin.h"+#include "storage/shmem.h"+#include "utils/snapmgr.h"+++/*+ * We use csnActive to judge if csn snapshot enabled instead of by+ * enable_csn_snapshot, this design is similar to 'track_commit_timestamp'.+ *+ * Because in process of replication if master change 'enable_csn_snapshot'+ * in a database restart, standby should apply wal record for GUC changed,+ * then it's difficult to notice all backends about that. So they can get+ * the message by 'csnActive' which in share buffer. It will not+ * acquire a lock, so without performance issue.+ *+ */+typedef struct CSNShared+{+ bool csnActive;+ CSN last_csn_log_wal; /* for interval we log the assign csn to wal */+ CSN last_max_csn; /* Record the max csn till now */+ TransactionId xmin_for_csn; /*'xmin_for_csn' for when turn xid-snapshot to csn-snapshot*/+ volatile slock_t lock;+} CSNShared;+++static CSNShared *csnShared;++bool enable_csn_snapshot;+bool enable_csn_wal;++/*+ * Defines for CSNLog 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,+ * CSNLog page numbering also wraps around at+ * 0xFFFFFFFF/CSN_LOG_XACTS_PER_PAGE, and CSNLog segment numbering at+ * 0xFFFFFFFF/CLOG_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 TruncateCSNLog (see CSNLogPagePrecedes).+ */++/* We store the commit CSN for each xid */+#define CSN_LOG_XACTS_PER_PAGE (BLCKSZ / sizeof(CSN))++#define TransactionIdToPage(xid) ((xid) / (TransactionId) CSN_LOG_XACTS_PER_PAGE)+#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CSN_LOG_XACTS_PER_PAGE)++/* Link to shared-memory data structures for CLOG control */+static SlruCtlData CSNLogCtlData;+#define CsnlogCtl (&CSNLogCtlData)+++static int ZeroCSNLogPage(int pageno, bool write_xlog);+static void ZeroTruncateCSNLogPage(int pageno, bool write_xlog);+static bool CSNLogPagePrecedes(int page1, int page2);+static void CSNLogSetPageStatus(TransactionId xid, int nsubxids,+ TransactionId *subxids,+ CSN csn, int pageno);+static void CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno);++static void WriteCSNXlogRec(TransactionId xid, int nsubxids,+ TransactionId *subxids, CSN csn);+static void WriteZeroCSNPageXlogRec(int pageno);+static void WriteTruncateCSNXlogRec(int pageno);+static void set_last_log_wal_csn(CSN csn);+static CSN get_last_log_wal_csn(void);+++/*+ * CSNLogSetCSN+ *+ * Record CSN of transaction and its subtransaction tree.+ *+ * xid is a single xid to set status for. This will typically be the top level+ * transactionid for a top level commit or abort. It can also be a+ * subtransaction when we record transaction aborts.+ *+ * subxids is an array of xids of length nsubxids, representing subtransactions+ * in the tree of xid. In various cases nsubxids may be zero.+ *+ * csn is the commit sequence number of the transaction. It should be+ * AbortedCSN for abort cases.+ */+void+CSNLogSetCSN(TransactionId xid, int nsubxids, TransactionId *subxids, CSN csn,+ bool write_xlog)+{+ int pageno;+ int i = 0;+ int offset = 0;++ Assert(TransactionIdIsValid(xid));++ pageno = TransactionIdToPage(xid); /* get page of parent */++ if(write_xlog)+ WriteCSNXlogRec(xid, nsubxids, subxids, csn);++ for (;;)+ {+ int num_on_page = 0;++ while (i < nsubxids && TransactionIdToPage(subxids[i]) == pageno)+ {+ num_on_page++;+ i++;+ }++ CSNLogSetPageStatus(xid,+ num_on_page, subxids + offset,+ csn, pageno);+ if (i >= nsubxids)+ break;++ offset = i;+ pageno = TransactionIdToPage(subxids[offset]);+ xid = InvalidTransactionId;+ }+}++/*+ * Record the final state of transaction entries in the csn log for+ * all entries on a single page. Atomic only on this page.+ *+ * Otherwise API is same as TransactionIdSetTreeStatus()+ */+static void+CSNLogSetPageStatus(TransactionId xid, int nsubxids,+ TransactionId *subxids,+ CSN csn, int pageno)+{+ int slotno;+ int i;++ LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);++ slotno = SimpleLruReadPage(CsnlogCtl, pageno, true, xid);++ /* Subtransactions first, if needed ... */+ for (i = 0; i < nsubxids; i++)+ {+ Assert(CsnlogCtl->shared->page_number[slotno] ==+ TransactionIdToPage(subxids[i]));+ CSNLogSetCSNInSlot(subxids[i], csn, slotno);+ }++ /* ... then the main transaction */+ if (TransactionIdIsValid(xid))+ CSNLogSetCSNInSlot(xid, csn, slotno);++ CsnlogCtl->shared->page_dirty[slotno] = true;++ LWLockRelease(CSNLogControlLock);+}++/*+ * Sets the commit status of a single transaction.+ */+static void+CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno)+{+ int entryno = TransactionIdToPgIndex(xid);+ CSN *ptr;++ Assert(LWLockHeldByMe(CSNLogControlLock));++ ptr = (CSN *) (CsnlogCtl->shared->page_buffer[slotno] ++ entryno * sizeof(CSN));++ *ptr = csn;+}++/*+ * Interrogate the state of a transaction in the log.+ *+ * NB: this is a low-level routine and is NOT the preferred entry point+ * for most uses; TransactionIdGetCSN() in csn_snapshot.c is the+ * intended caller.+ */+CSN+CSNLogGetCSNByXid(TransactionId xid)+{+ int pageno = TransactionIdToPage(xid);+ int entryno = TransactionIdToPgIndex(xid);+ int slotno;+ CSN csn;++ /* lock is acquired by SimpleLruReadPage_ReadOnly */+ slotno = SimpleLruReadPage_ReadOnly(CsnlogCtl, pageno, xid);+ csn = *(CSN *) (CsnlogCtl->shared->page_buffer[slotno] ++ entryno * sizeof(CSN));++ LWLockRelease(CSNLogControlLock);++ return csn;+}++/*+ * Number of shared CSNLog buffers.+ */+static Size+CSNLogShmemBuffers(void)+{+ return Min(32, Max(4, NBuffers / 512));+}++/*+ * Reserve shared memory for CsnlogCtl.+ */+Size+CSNLogShmemSize(void)+{+ return SimpleLruShmemSize(CSNLogShmemBuffers(), 0);+}++/*+ * Initialization of shared memory for CSNLog.+ */+void+CSNLogShmemInit(void)+{+ bool found;++ CsnlogCtl->PagePrecedes = CSNLogPagePrecedes;+ SimpleLruInit(CsnlogCtl, "CSNLog Ctl", CSNLogShmemBuffers(), 0,+ CSNLogControlLock, "pg_csn", LWTRANCHE_CSN_LOG_BUFFERS,+ SYNC_HANDLER_CSN_LOG);++ csnShared = ShmemInitStruct("CSNlog shared",+ sizeof(CSNShared),+ &found);+ if (!found)+ {+ set_last_max_csn(InvalidCSN, true);+ csnShared->last_csn_log_wal = InvalidCSN;+ csnShared->xmin_for_csn = InvalidTransactionId;+ csnShared->csnActive = false;+ SpinLockInit(&csnShared->lock);+ }+}++/*+ * Initialize (or reinitialize) a page of CSNLog to zeroes.+ *+ * 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+ZeroCSNLogPage(int pageno, bool write_xlog)+{+ int slotno;++ Assert(LWLockHeldByMe(CSNLogControlLock));++ slotno = SimpleLruZeroPage(CsnlogCtl, pageno);++ if(write_xlog)+ WriteZeroCSNPageXlogRec(pageno);++ return slotno;+}++static void+ZeroTruncateCSNLogPage(int pageno, bool write_xlog)+{+ if(write_xlog)+ WriteTruncateCSNXlogRec(pageno);++ SimpleLruTruncate(CsnlogCtl, pageno);+}++void+ActivateCSNlog(void)+{+ int startPage;+ TransactionId nextXid = InvalidTransactionId;++ if (csnShared->csnActive)+ return;++ nextXid = XidFromFullTransactionId(ShmemVariableCache->nextXid);+ startPage = TransactionIdToPage(nextXid);++ /* Create the current segment file, if necessary */+ if (!SimpleLruDoesPhysicalPageExist(CsnlogCtl, startPage))+ {+ int slotno;+ LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);+ slotno = ZeroCSNLogPage(startPage, false);+ SimpleLruWritePage(CsnlogCtl, slotno);+ LWLockRelease(CSNLogControlLock);+ }+ csnShared->csnActive = true;+}++bool+get_csnlog_status(void)+{+ if(!csnShared)+ {+ /* Should not arrived */+ elog(ERROR, "We do not have csnShared point");+ }+ return csnShared->csnActive;+}++void+DeactivateCSNlog(void)+{+ csnShared->csnActive = false;+ LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);+ (void) SlruScanDirectory(CsnlogCtl, SlruScanDirCbDeleteAll, NULL);+ LWLockRelease(CSNLogControlLock);+}++void+StartupCSN(void)+{+ ActivateCSNlog();+}++void+CompleteCSNInitialization(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 (!get_csnlog_status())+ DeactivateCSNlog();+ else+ ActivateCSNlog();+}++void+CSNlogParameterChange(bool newvalue, bool oldvalue)+{+ if (newvalue)+ {+ if (!csnShared->csnActive)+ ActivateCSNlog();+ }+ else if (csnShared->csnActive)+ DeactivateCSNlog();+}++/*+ * Perform a checkpoint --- either during shutdown, or on-the-fly+ */+void+CheckPointCSNLog(void)+{+ if (!get_csnlog_status())+ return;++ /*+ * Flush dirty CSNLog pages to disk.+ *+ * This is not actually necessary from a correctness point of view. We do+ * it merely to improve the odds that writing of dirty pages is done by+ * the checkpoint process and not by backends.+ */+ TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_START(true);+ SimpleLruWriteAll(CsnlogCtl, true);+ TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_DONE(true);+}++/*+ * Make sure that CSNLog 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 clog or xlog page to make room+ * in shared memory.+ */+void+ExtendCSNLog(TransactionId newestXact)+{+ int pageno;++ if (!get_csnlog_status())+ return;++ /*+ * No work except at first XID of a page. But beware: just after+ * wraparound, the first XID of page zero is FirstNormalTransactionId.+ */+ if (TransactionIdToPgIndex(newestXact) != 0 &&+ !TransactionIdEquals(newestXact, FirstNormalTransactionId))+ return;++ pageno = TransactionIdToPage(newestXact);++ LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);++ /* Zero the page and make an XLOG entry about it */+ ZeroCSNLogPage(pageno, !InRecovery);++ LWLockRelease(CSNLogControlLock);+}++/*+ * Remove all CSNLog segments before the one holding the passed+ * transaction ID.+ *+ * This is normally called during checkpoint, with oldestXact being the+ * oldest TransactionXmin of any running transaction.+ */+void+TruncateCSNLog(TransactionId oldestXact)+{+ int cutoffPage;++ if (!get_csnlog_status())+ return;++ /*+ * The cutoff point is the start of the segment containing oldestXact. We+ * pass the *page* containing oldestXact to SimpleLruTruncate. We step+ * back one transaction to avoid passing a cutoff page that hasn't been+ * created yet in the rare case that oldestXact would be the first item on+ * a page and oldestXact == next XID. In that case, if we didn't subtract+ * one, we'd trigger SimpleLruTruncate's wraparound detection.+ */+ TransactionIdRetreat(oldestXact);+ cutoffPage = TransactionIdToPage(oldestXact);+ ZeroTruncateCSNLogPage(cutoffPage, true);+}++/*+ * Decide which of two CSNLog page numbers is "older" for truncation+ * purposes.+ *+ * We need to use comparison of TransactionIds here in order to do the right+ * thing with wraparound XID arithmetic. However, if we are asked about+ * page number zero, we don't want to hand InvalidTransactionId to+ * TransactionIdPrecedes: it'll get weird about permanent xact IDs. So,+ * offset both xids by FirstNormalTransactionId to avoid that.+ */+static bool+CSNLogPagePrecedes(int page1, int page2)+{+ TransactionId xid1;+ TransactionId xid2;++ xid1 = ((TransactionId) page1) * CSN_LOG_XACTS_PER_PAGE;+ xid1 += FirstNormalTransactionId;+ xid2 = ((TransactionId) page2) * CSN_LOG_XACTS_PER_PAGE;+ xid2 += FirstNormalTransactionId;++ return TransactionIdPrecedes(xid1, xid2);+}++void+WriteAssignCSNXlogRec(CSN csn)+{+ CSN log_csn = 0;++ if (!enable_csn_wal || csn <= get_last_log_wal_csn())+ return;++ /*+ * We log the CSN 5s greater than generated, you can see comments on+ * CSN_ASSIGN_TIME_INTERVAL define.+ */+ log_csn = CSNAddByNanosec(csn, CSN_ASSIGN_TIME_INTERVAL);+ set_last_log_wal_csn(log_csn);++ XLogBeginInsert();+ XLogRegisterData((char *) (&log_csn), sizeof(CSN));+ XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ASSIGNMENT);+}++static void+WriteCSNXlogRec(TransactionId xid, int nsubxids,+ TransactionId *subxids, CSN csn)+{+ xl_csn_set xlrec;++ if (!enable_csn_wal)+ return;++ xlrec.xtop = xid;+ xlrec.nsubxacts = nsubxids;+ xlrec.csn = csn;++ XLogBeginInsert();+ XLogRegisterData((char *) &xlrec, MinSizeOfCSNSet);+ XLogRegisterData((char *) subxids, nsubxids * sizeof(TransactionId));+ (void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_SETCSN);+}++/*+ * Write a ZEROPAGE xlog record+ */+static void+WriteZeroCSNPageXlogRec(int pageno)+{+ if (!enable_csn_wal)+ return;++ XLogBeginInsert();+ XLogRegisterData((char *) (&pageno), sizeof(int));+ (void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ZEROPAGE);+}++/*+ * Write a TRUNCATE xlog record+ */+static void+WriteTruncateCSNXlogRec(int pageno)+{+ if (!enable_csn_wal)+ return;++ XLogBeginInsert();+ XLogRegisterData((char *) (&pageno), sizeof(int));+ (void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_TRUNCATE);+}++void+csnlog_redo(XLogReaderState *record)+{+ uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;++ /* Backup blocks are not used in csnlog records */+ Assert(!XLogRecHasAnyBlockRefs(record));++ if (info == XLOG_CSN_ASSIGNMENT)+ {+ CSN csn;++ memcpy(&csn, XLogRecGetData(record), sizeof(CSN));+ set_last_max_csn(csn, true);+ }+ else if (info == XLOG_CSN_SETCSN)+ {+ xl_csn_set *xlrec = (xl_csn_set *) XLogRecGetData(record);+ CSNLogSetCSN(xlrec->xtop, xlrec->nsubxacts, xlrec->xsub, xlrec->csn, false);+ }+ else if (info == XLOG_CSN_ZEROPAGE)+ {+ int pageno;+ int slotno;++ memcpy(&pageno, XLogRecGetData(record), sizeof(int));+ LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);+ slotno = ZeroCSNLogPage(pageno, false);+ SimpleLruWritePage(CsnlogCtl, slotno);+ LWLockRelease(CSNLogControlLock);+ Assert(!CsnlogCtl->shared->page_dirty[slotno]);++ }+ else if (info == XLOG_CSN_TRUNCATE)+ {+ int pageno;++ memcpy(&pageno, XLogRecGetData(record), sizeof(int));+ CsnlogCtl->shared->latest_page_number = pageno;+ ZeroTruncateCSNLogPage(pageno, false);+ }+ else+ elog(PANIC, "csnlog_redo: unknown op code %u", info);+}++static void+set_last_log_wal_csn(CSN csn)+{+ SpinLockAcquire(&csnShared->lock);+ csnShared->last_csn_log_wal = csn;+ SpinLockRelease(&csnShared->lock);+}++static CSN+get_last_log_wal_csn(void)+{+ CSN csn;++ SpinLockAcquire(&csnShared->lock);+ csn = csnShared->last_csn_log_wal;+ SpinLockRelease(&csnShared->lock);+ return csn;+}++CSN+set_last_max_csn(CSN csn, bool force)+{+ SpinLockAcquire(&csnShared->lock);+ if (csn <= csnShared->last_max_csn && !force)+ csn = csnShared->last_max_csn + 1;++ csnShared->last_max_csn = csn;++ SpinLockRelease(&csnShared->lock);+ return csn;+}++CSN+get_last_max_csn(void)+{+ CSN csn;++ SpinLockAcquire(&csnShared->lock);+ csn = csnShared->last_max_csn;+ SpinLockRelease(&csnShared->lock);+ return csn;+}++CSN+get_xmin_for_csn(void)+{+ CSN csn;++ SpinLockAcquire(&csnShared->lock);+ csn = csnShared->xmin_for_csn;+ SpinLockRelease(&csnShared->lock);+ return csn;+}++void+prepare_csn_env(bool enable)+{+ if (enable)+ {+ 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();+ }+ else+ /* Try to drop all csnlog seg */+ DeactivateCSNlog();+}++/*+ * Entrypoint for sync.c to sync csnlog files.+ */+int+csnlogsyncfiletag(const FileTag *ftag, char *path)+{+ return SlruSyncFileTag(CsnlogCtl, ftag, path);+}diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.cindex 58091f6b52..b1e5ec350e 100644--- a/src/backend/access/transam/rmgr.c+++ b/src/backend/access/transam/rmgr.c@@ -28,6 +28,7 @@
#include "replication/origin.h"
#include "storage/standby.h"
#include "utils/relmapper.h"
+#include "access/csn_log.h"
/* must be kept in sync with RmgrData definition in xlog_internal.h */
#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask) \
diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.cindex 142da4aaff..5791a2e66e 100644--- a/src/backend/access/transam/varsup.c+++ b/src/backend/access/transam/varsup.c@@ -15,6 +15,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/csn_log.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/xact.h"
@@ -175,6 +176,7 @@ GetNewTransactionId(bool isSubXact)
* Extend pg_subtrans and pg_commit_ts too.
*/
ExtendCLOG(xid);
+ ExtendCSNLog(xid);
ExtendCommitTs(xid);
ExtendSUBTRANS(xid);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.cindex 6f8810e149..46f016dce7 100644--- a/src/backend/access/transam/xlog.c+++ b/src/backend/access/transam/xlog.c@@ -24,6 +24,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/csn_log.h"
#include "access/heaptoast.h"
#include "access/multixact.h"
#include "access/rewriteheap.h"
@@ -4649,6 +4650,7 @@ InitControlFile(uint64 sysidentifier)
ControlFile->wal_level = wal_level;
ControlFile->wal_log_hints = wal_log_hints;
ControlFile->track_commit_timestamp = track_commit_timestamp;
+ ControlFile->enable_csn_snapshot = enable_csn_snapshot;
ControlFile->data_checksum_version = bootstrap_data_checksum_version;
}
@@ -6910,6 +6912,7 @@ StartupXLOG(void)
SetCommitTsLimit(checkPoint.oldestCommitTsXid,
checkPoint.newestCommitTsXid);
XLogCtl->ckptFullXid = checkPoint.nextXid;
+ set_last_max_csn(checkPoint.oldestUsedCSN, true);
/*
* Initialize replication slots, before there's a chance to remove
@@ -6945,6 +6948,9 @@ StartupXLOG(void)
if (ControlFile->track_commit_timestamp)
StartupCommitTs();
+ if(ControlFile->enable_csn_snapshot)+ StartupCSN();+
/*
* Recover knowledge about replay progress of known replication partners.
*/
@@ -8062,6 +8068,7 @@ StartupXLOG(void)
* commit timestamp.
*/
CompleteCommitTsInitialization();
+ CompleteCSNInitialization();
/*
* All done with end-of-recovery actions.
@@ -8981,6 +8988,8 @@ CreateCheckPoint(int flags)
*/
last_important_lsn = GetLastImportantRecPtr();
+ checkPoint.oldestUsedCSN = get_last_max_csn();+
/*
* We must block concurrent insertions while examining insert state to
* determine the checkpoint REDO pointer.
@@ -9370,6 +9379,7 @@ CheckPointGuts(XLogRecPtr checkPointRedo, int flags)
TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags);
CheckpointStats.ckpt_write_t = GetCurrentTimestamp();
CheckPointCLOG();
+ CheckPointCSNLog();
CheckPointCommitTs();
CheckPointSUBTRANS();
CheckPointMultiXact();
@@ -9922,7 +9932,8 @@ XLogReportParameters(void)
max_wal_senders != ControlFile->max_wal_senders ||
max_prepared_xacts != ControlFile->max_prepared_xacts ||
max_locks_per_xact != ControlFile->max_locks_per_xact ||
- track_commit_timestamp != ControlFile->track_commit_timestamp)+ track_commit_timestamp != ControlFile->track_commit_timestamp ||+ enable_csn_snapshot != ControlFile->enable_csn_snapshot)
{
/*
* The change in number of backend slots doesn't need to be WAL-logged
@@ -9944,6 +9955,7 @@ XLogReportParameters(void)
xlrec.wal_level = wal_level;
xlrec.wal_log_hints = wal_log_hints;
xlrec.track_commit_timestamp = track_commit_timestamp;
+ xlrec.enable_csn_snapshot = enable_csn_snapshot;
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, sizeof(xlrec));
@@ -9954,6 +9966,9 @@ XLogReportParameters(void)
LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+ if (enable_csn_snapshot != ControlFile->enable_csn_snapshot)+ prepare_csn_env(enable_csn_snapshot);+
ControlFile->MaxConnections = MaxConnections;
ControlFile->max_worker_processes = max_worker_processes;
ControlFile->max_wal_senders = max_wal_senders;
@@ -9962,6 +9977,7 @@ XLogReportParameters(void)
ControlFile->wal_level = wal_level;
ControlFile->wal_log_hints = wal_log_hints;
ControlFile->track_commit_timestamp = track_commit_timestamp;
+ ControlFile->enable_csn_snapshot = enable_csn_snapshot;
UpdateControlFile();
LWLockRelease(ControlFileLock);
@@ -10393,6 +10409,9 @@ xlog_redo(XLogReaderState *record)
CommitTsParameterChange(xlrec.track_commit_timestamp,
ControlFile->track_commit_timestamp);
ControlFile->track_commit_timestamp = xlrec.track_commit_timestamp;
+ CSNlogParameterChange(xlrec.enable_csn_snapshot,+ ControlFile->enable_csn_snapshot);+ ControlFile->enable_csn_snapshot = xlrec.enable_csn_snapshot;
UpdateControlFile();
LWLockRelease(ControlFileLock);
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.cindex c064352e23..515f2b3eb9 100644--- a/src/backend/commands/vacuum.c+++ b/src/backend/commands/vacuum.c@@ -53,7 +53,7 @@
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
-+#include "access/csn_log.h"
/*
* GUC parameters
@@ -1688,6 +1688,7 @@ vac_truncate_clog(TransactionId frozenXID,
*/
TruncateCLOG(frozenXID, oldestxid_datoid);
TruncateCommitTs(frozenXID);
+ TruncateCSNLog(frozenXID);
TruncateMultiXact(minMulti, minmulti_datoid);
/*
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.cindex 3e4ec53a97..e06bc13841 100644--- a/src/backend/storage/ipc/ipci.c+++ b/src/backend/storage/ipc/ipci.c@@ -16,6 +16,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/csn_log.h"
#include "access/heapam.h"
#include "access/multixact.h"
#include "access/nbtree.h"
@@ -127,6 +128,7 @@ CreateSharedMemoryAndSemaphores(void)
size = add_size(size, ProcGlobalShmemSize());
size = add_size(size, XLOGShmemSize());
size = add_size(size, CLOGShmemSize());
+ size = add_size(size, CSNLogShmemSize());
size = add_size(size, CommitTsShmemSize());
size = add_size(size, SUBTRANSShmemSize());
size = add_size(size, TwoPhaseShmemSize());
@@ -218,6 +220,7 @@ CreateSharedMemoryAndSemaphores(void)
*/
XLOGShmemInit();
CLOGShmemInit();
+ CSNLogShmemInit();
CommitTsShmemInit();
SUBTRANSShmemInit();
MultiXactShmemInit();
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.cindex 4fc6ffb917..4a200b0815 100644--- a/src/backend/storage/ipc/procarray.c+++ b/src/backend/storage/ipc/procarray.c@@ -48,6 +48,7 @@
#include <signal.h>
#include "access/clog.h"
+#include "access/csn_log.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
@@ -1170,6 +1171,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
while (TransactionIdPrecedes(latestObservedXid, running->nextXid))
{
ExtendSUBTRANS(latestObservedXid);
+ ExtendCSNLog(latestObservedXid);
TransactionIdAdvance(latestObservedXid);
}
TransactionIdRetreat(latestObservedXid); /* = running->nextXid - 1 */
@@ -4320,6 +4322,7 @@ RecordKnownAssignedTransactionIds(TransactionId xid)
while (TransactionIdPrecedes(next_expected_xid, xid))
{
TransactionIdAdvance(next_expected_xid);
+ ExtendCSNLog(next_expected_xid);
ExtendSUBTRANS(next_expected_xid);
}
Assert(next_expected_xid == xid);
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.cindex 975d547f34..3489e51ef3 100644--- a/src/backend/storage/lmgr/lwlock.c+++ b/src/backend/storage/lmgr/lwlock.c@@ -134,6 +134,8 @@ static const char *const BuiltinTrancheNames[] = {
"CommitTSBuffer",
/* LWTRANCHE_SUBTRANS_BUFFER: */
"SubtransBuffer",
+ /* LWTRANCHE_CSN_LOG_BUFFERS */+ "CSNLogBuffer",
/* LWTRANCHE_MULTIXACTOFFSET_BUFFER: */
"MultiXactOffsetBuffer",
/* LWTRANCHE_MULTIXACTMEMBER_BUFFER: */
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txtindex 6c7cf6c295..56196b2d0d 100644--- a/src/backend/storage/lmgr/lwlocknames.txt+++ b/src/backend/storage/lmgr/lwlocknames.txt@@ -53,3 +53,4 @@ XactTruncationLock 44
# 45 was XactTruncationLock until removal of BackendRandomLock
WrapLimitsVacuumLock 46
NotifyQueueTailLock 47
+CSNLogControlLock 48diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.cindex 708215614d..cbb2fd510b 100644--- a/src/backend/storage/sync/sync.c+++ b/src/backend/storage/sync/sync.c@@ -20,6 +20,7 @@
#include "access/commit_ts.h"
#include "access/clog.h"
+#include "access/csn_log.h"
#include "access/multixact.h"
#include "access/xlog.h"
#include "access/xlogutils.h"
@@ -118,6 +119,10 @@ static const SyncOps syncsw[] = {
/* pg_multixact/members */
[SYNC_HANDLER_MULTIXACT_MEMBER] = {
.sync_syncfiletag = multixactmemberssyncfiletag
+ },+ /* CSN_Log */+ [SYNC_HANDLER_CSN_LOG] = {+ .sync_syncfiletag = csnlogsyncfiletag
}
};
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.cindex 3b36a31a47..14063bcf9e 100644--- a/src/backend/utils/misc/guc.c+++ b/src/backend/utils/misc/guc.c@@ -1170,6 +1170,24 @@ static struct config_bool ConfigureNamesBool[] =
false,
NULL, NULL, NULL
},
+ {+ {"enable_csn_snapshot", PGC_POSTMASTER, RESOURCES_MEM,+ gettext_noop("Enable csn-base snapshot."),+ gettext_noop("Used to achieve REPEATEBLE READ isolation level for postgres_fdw transactions.")+ },+ &enable_csn_snapshot,+ false,+ NULL, NULL, NULL+ },+ {+ {"enable_csn_wal", PGC_POSTMASTER, RESOURCES_MEM,+ gettext_noop("Enable csn-wal record."),+ gettext_noop("Used to enable csn-wal record")+ },+ &enable_csn_wal,+ true,+ NULL, NULL, NULL+ },
{
{"ssl", PGC_SIGHUP, CONN_AUTH_SSL,
gettext_noop("Enables SSL connections."),
diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.dindex b0c50a3c7f..3fcd0f4ccf 100644--- a/src/backend/utils/probes.d+++ b/src/backend/utils/probes.d@@ -77,6 +77,8 @@ provider postgresql {
probe clog__checkpoint__done(bool);
probe subtrans__checkpoint__start(bool);
probe subtrans__checkpoint__done(bool);
+ probe csnlog__checkpoint__start(bool);+ probe csnlog__checkpoint__done(bool);
probe multixact__checkpoint__start(bool);
probe multixact__checkpoint__done(bool);
probe twophase__checkpoint__start();
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.cindex 3c1cf78b4f..6b3b9ff504 100644--- a/src/bin/initdb/initdb.c+++ b/src/bin/initdb/initdb.c@@ -223,7 +223,8 @@ static const char *const subdirs[] = {
"pg_xact",
"pg_logical",
"pg_logical/snapshots",
- "pg_logical/mappings"+ "pg_logical/mappings",+ "pg_csn"
};
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.cindex f911f98d94..325e6a0e2b 100644--- a/src/bin/pg_controldata/pg_controldata.c+++ b/src/bin/pg_controldata/pg_controldata.c@@ -300,6 +300,8 @@ main(int argc, char *argv[])
ControlFile->max_locks_per_xact);
printf(_("track_commit_timestamp setting: %s\n"),
ControlFile->track_commit_timestamp ? _("on") : _("off"));
+ printf(_("enable_csn_snapshot setting: %s\n"),+ ControlFile->enable_csn_snapshot ? _("on") : _("off"));
printf(_("Maximum data alignment: %u\n"),
ControlFile->maxAlign);
/* we don't print floatFormat since can't say much useful about it */
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.cindex e23b8ca88d..ed9112ed40 100644--- a/src/bin/pg_upgrade/pg_upgrade.c+++ b/src/bin/pg_upgrade/pg_upgrade.c@@ -541,6 +541,11 @@ copy_xact_xlog_xid(void)
check_ok();
}
+ if(old_cluster.controldata.cat_ver > CSN_BASE_SNAPSHOT_ADD_VER)+ {+ copy_subdir_files("pg_csn", "pg_csn");+ }+
/* now reset the wal archives in the new cluster */
prep_status("Resetting WAL archives");
exec_prog(UTILITY_LOG_FILE, NULL, true, true,
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.hindex 919a7849fd..77f5a030ae 100644--- a/src/bin/pg_upgrade/pg_upgrade.h+++ b/src/bin/pg_upgrade/pg_upgrade.h@@ -123,6 +123,8 @@ extern char *output_files[];
*/
#define JSONB_FORMAT_CHANGE_CAT_VER 201409291
+#define CSN_BASE_SNAPSHOT_ADD_VER 202002010+
/*
* Each relation is represented by a relinfo structure.
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.cindex 852d8ca4b1..2d280ce940 100644--- a/src/bin/pg_waldump/rmgrdesc.c+++ b/src/bin/pg_waldump/rmgrdesc.c@@ -11,6 +11,7 @@
#include "access/brin_xlog.h"
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/csn_log.h"
#include "access/generic_xlog.h"
#include "access/ginxlog.h"
#include "access/gistxlog.h"
diff --git a/src/include/access/csn_log.h b/src/include/access/csn_log.hnew file mode 100644index 0000000000..01fb0daad4--- /dev/null+++ b/src/include/access/csn_log.h@@ -0,0 +1,83 @@+/*+ * csn_log.h+ *+ * Commit-Sequence-Number log.+ *+ * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group+ * Portions Copyright (c) 1994, Regents of the University of California+ *+ * src/include/access/csn_log.h+ */+#ifndef CSNLOG_H+#define CSNLOG_H++#include "access/xlog.h"+#include "storage/sync.h"+#include "utils/snapshot.h"+++typedef struct xl_csn_set+{+ CSN csn;+ TransactionId xtop; /* XID's top-level XID */+ int nsubxacts; /* number of subtransaction XIDs */+ TransactionId xsub[FLEXIBLE_ARRAY_MEMBER]; /* assigned subxids */+} xl_csn_set;+++/* XLOG stuff */+#define XLOG_CSN_ASSIGNMENT 0x00+#define XLOG_CSN_SETCSN 0x10+#define XLOG_CSN_ZEROPAGE 0x20+#define XLOG_CSN_TRUNCATE 0x30++/*+ * We should log MAX generated CSN to wal, so that database will not generate+ * a historical CSN after database restart. This may appear when system time+ * turned back.+ *+ * However we can not log the MAX CSN every time it generated, if so it will+ * cause too many wal expend, so we log it 5s more in the future.+ *+ * As a trade off, when this database restart, there will be 5s bad performance+ * for time synchronization among postgres instances.+ *+ * It looks like we can redefine this as a configure parameter, and the user+ * can decide which way they prefer.+ *+ */+#define CSN_ASSIGN_TIME_INTERVAL 5++#define MinSizeOfCSNSet offsetof(xl_csn_set, xsub)+#define CSNAddByNanosec(csn,second) (csn + second * 1000000000L)+++extern void CSNLogSetCSN(TransactionId xid, int nsubxids,+ TransactionId *subxids, CSN csn, bool write_xlog);+extern CSN CSNLogGetCSNByXid(TransactionId xid);++extern Size CSNLogShmemSize(void);+extern void CSNLogShmemInit(void);+extern void ShutdownCSNLog(void);+extern void CheckPointCSNLog(void);+extern void ExtendCSNLog(TransactionId newestXact);+extern void TruncateCSNLog(TransactionId oldestXact);+extern int csnlogsyncfiletag(const FileTag *ftag, char *path);++extern void csnlog_redo(XLogReaderState *record);+extern void csnlog_desc(StringInfo buf, XLogReaderState *record);+extern const char *csnlog_identify(uint8 info);+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 CatchCSNLog(void);+extern void ActivateCSNlog(void);+extern void DeactivateCSNlog(void);+extern void StartupCSN(void);+extern void CompleteCSNInitialization(void);+extern void CSNlogParameterChange(bool newvalue, bool oldvalue);+extern bool get_csnlog_status(void);++#endif /* CSNLOG_H */diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.hindex f582cf535f..3cf0775176 100644--- a/src/include/access/rmgrlist.h+++ b/src/include/access/rmgrlist.h@@ -47,3 +47,4 @@ PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_i
PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL)
PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask)
PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL)
+PG_RMGR(RM_CSNLOG_ID, "CSN", csnlog_redo, csnlog_desc, csnlog_identify, NULL, NULL, NULL)diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.hindex b23e286406..d9690a23bb 100644--- a/src/include/access/xlog_internal.h+++ b/src/include/access/xlog_internal.h@@ -240,6 +240,7 @@ typedef struct xl_parameter_change
int wal_level;
bool wal_log_hints;
bool track_commit_timestamp;
+ bool enable_csn_snapshot;
} xl_parameter_change;
/* logs restore point */
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.hindex e3f48158ce..194d100ed4 100644--- a/src/include/catalog/pg_control.h+++ b/src/include/catalog/pg_control.h@@ -61,6 +61,8 @@ typedef struct CheckPoint
* set to InvalidTransactionId.
*/
TransactionId oldestActiveXid;
++ uint64 oldestUsedCSN;
} CheckPoint;
/* XLOG info values for XLOG rmgr */
@@ -181,6 +183,7 @@ typedef struct ControlFileData
int max_prepared_xacts;
int max_locks_per_xact;
bool track_commit_timestamp;
+ bool enable_csn_snapshot;
/*
* This data is used to check for hardware-architecture compatibility of
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.hindex a8f052e484..65d1e49fb2 100644--- a/src/include/storage/lwlock.h+++ b/src/include/storage/lwlock.h@@ -168,6 +168,7 @@ typedef enum BuiltinTrancheIds
LWTRANCHE_XACT_BUFFER = NUM_INDIVIDUAL_LWLOCKS,
LWTRANCHE_COMMITTS_BUFFER,
LWTRANCHE_SUBTRANS_BUFFER,
+ LWTRANCHE_CSN_LOG_BUFFERS,
LWTRANCHE_MULTIXACTOFFSET_BUFFER,
LWTRANCHE_MULTIXACTMEMBER_BUFFER,
LWTRANCHE_NOTIFY_BUFFER,
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.hindex 2fd1ff09a7..f0e920f462 100644--- a/src/include/storage/proc.h+++ b/src/include/storage/proc.h@@ -17,6 +17,7 @@
#include "access/clog.h"
#include "access/xlogdefs.h"
#include "lib/ilist.h"
+#include "utils/snapshot.h"
#include "storage/latch.h"
#include "storage/lock.h"
#include "storage/pg_sema.h"
diff --git a/src/include/storage/sync.h b/src/include/storage/sync.hindex fbdf34f762..e49dd48fe8 100644--- a/src/include/storage/sync.h+++ b/src/include/storage/sync.h@@ -39,6 +39,7 @@ typedef enum SyncRequestHandler
SYNC_HANDLER_COMMIT_TS,
SYNC_HANDLER_MULTIXACT_OFFSET,
SYNC_HANDLER_MULTIXACT_MEMBER,
+ SYNC_HANDLER_CSN_LOG,
SYNC_HANDLER_NONE
} SyncRequestHandler;
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.hindex 6b60755c53..92575a207f 100644--- a/src/include/utils/snapshot.h+++ b/src/include/utils/snapshot.h@@ -121,6 +121,11 @@ typedef enum SnapshotType
typedef struct SnapshotData *Snapshot;
#define InvalidSnapshot ((Snapshot) NULL)
+#define InvalidCSN ((CSN) 0)+typedef uint64 CSN;++extern bool enable_csn_snapshot;+extern bool enable_csn_wal;
/*
* Struct representing all kind of possible snapshots.
@@ -208,6 +213,12 @@ typedef struct SnapshotData
TimestampTz whenTaken; /* timestamp when snapshot was taken */
XLogRecPtr lsn; /* position in the WAL stream when taken */
+ /*+ * CSN for snapshot isolation support.+ * Will be used only if enable_csn_snapshot is enabled.+ */+ CSN csn;+
/*
* The transaction completion count at the time GetSnapshotData() built
* this snapshot. Allows to avoid re-computing static snapshots when no
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.outindex a62bf5dc92..e21403028d 100644--- a/src/test/regress/expected/sysviews.out+++ b/src/test/regress/expected/sysviews.out@@ -96,6 +96,8 @@ select name, setting from pg_settings where name like 'enable%';
name | setting
--------------------------------+---------
enable_bitmapscan | on
+ enable_csn_snapshot | off+ enable_csn_wal | on
enable_gathermerge | on
enable_hashagg | on
enable_hashjoin | on
@@ -114,7 +116,7 @@ select name, setting from pg_settings where name like 'enable%';
enable_seqscan | on
enable_sort | on
enable_tidscan | on
-(19 rows)+(21 rows)
-- Test that the pg_timezone_names and pg_timezone_abbrevs views are
-- more-or-less working. We can't test their contents in any great detail
--
2.25.1
[text/x-patch] 0002-CSN-Snapshot.patch (63.2K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/3-0002-CSN-Snapshot.patch)
download | inline diff:
From 22d80716f06e1b6000f788b5b96da70c0d7aa581 Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
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.sgmlindex 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'
</varlistentry>
</variablelist>
- </sect1>+ <sect2 id="runtime-config-CSN-base-snapshot">+ <title>CSN Based Snapshot</title>++ <para>+ By default, The snapshots in <productname>PostgreSQL</productname> uses the+ XID (TransactionID) to identify the status of the transaction, the in-progress+ transactions, and the future transactions for all its visibility calculations.+ </para>++ <para>+ <productname>PostgreSQL</productname> also provides the CSN (commit-sequence-number)+ based mechanism to identify the past-transactions and the ones that are yet to+ be started/committed.+ </para>++ <variablelist>+ <varlistentry id="guc-enable-csn-snapshot" xreflabel="enable_csn_snapshot">+ <term><varname>enable_csn_snapshot</varname> (<type>boolean</type>)+ <indexterm>+ <primary><varname>enable_csn_snapshot</varname> configuration parameter</primary>+ </indexterm>+ </term>+ <listitem>++ <para>+ Enable/disable the CSN based transaction visibility tracking for the snapshot.+ </para>++ <para>+ <productname>PostgreSQL</productname> 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.+ </para>++ <para>+ when enabled <productname>PostgreSQL</productname> creates+ <filename>pg_csn</filename> directory under <envar>PGDATA</envar> to keep+ the track of CSN and XID mappings.+ </para>++ <para>+ The default value is off.+ </para>+ </listitem>+ </varlistentry>++ </variablelist>+ </sect2>+ </sect1>
<sect1 id="runtime-config-compatible">
<title>Version and Platform Compatibility</title>
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefileindex 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.cindex 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.cnew file mode 100644index 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 <sys/time.h>++#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.cindex 89335b64a2..6b91953193 100644--- a/src/backend/access/transam/twophase.c+++ b/src/backend/access/transam/twophase.c@@ -77,6 +77,8 @@
#include <unistd.h>
#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.cindex 6395a9b240..deabedbe37 100644--- a/src/backend/access/transam/xact.c+++ b/src/backend/access/transam/xact.c@@ -21,6 +21,7 @@
#include <unistd.h>
#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.cindex 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.cindex 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.cindex 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.txtindex 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 49diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.cindex 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.cindex 14063bcf9e..cc01d72d64 100644--- a/src/backend/utils/misc/guc.c+++ b/src/backend/utils/misc/guc.c@@ -30,6 +30,7 @@
#include <unistd.h>
#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.cindex 95704265b6..2d4c5f53af 100644--- a/src/backend/utils/time/snapmgr.c+++ b/src/backend/utils/time/snapmgr.c@@ -48,6 +48,7 @@
#include <sys/stat.h>
#include <unistd.h>
+#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.cnew file mode 120000index 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.hindex 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.hnew file mode 100644index 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.hindex 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.datindex 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.hindex 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.hindex 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.hindex 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.hindex 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.hindex 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.hindex 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
[text/x-patch] 0003-CSN-Snapshot-Tests.patch (21.9K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/4-0003-CSN-Snapshot-Tests.patch)
download | inline diff:
From fa93fd78d70742d254d1972cc23d05006c19d4af Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Fri, 29 Jan 2021 10:37:43 +0500
Subject: [PATCH 3/4] CSN Snapshot Tests
---
src/test/modules/Makefile | 1 +
src/test/modules/csnsnapshot/Makefile | 22 ++
.../modules/csnsnapshot/csn_snapshot.conf | 1 +
src/test/modules/csnsnapshot/t/001_base.pl | 119 ++++++++++
src/test/modules/csnsnapshot/t/002_standby.pl | 66 ++++++
.../modules/csnsnapshot/t/003_time_skew.pl | 211 ++++++++++++++++++
.../csnsnapshot/t/004_read_committed.pl | 97 ++++++++
7 files changed, 517 insertions(+)
create mode 100644 src/test/modules/csnsnapshot/Makefile
create mode 100644 src/test/modules/csnsnapshot/csn_snapshot.conf
create mode 100644 src/test/modules/csnsnapshot/t/001_base.pl
create mode 100644 src/test/modules/csnsnapshot/t/002_standby.pl
create mode 100644 src/test/modules/csnsnapshot/t/003_time_skew.pl
create mode 100644 src/test/modules/csnsnapshot/t/004_read_committed.pl
diff --git a/src/test/modules/Makefile b/src/test/modules/Makefileindex 93e7829c67..353c57eb51 100644--- a/src/test/modules/Makefile+++ b/src/test/modules/Makefile@@ -7,6 +7,7 @@ include $(top_builddir)/src/Makefile.global
SUBDIRS = \
brin \
commit_ts \
+ csnsnapshot \
delay_execution \
dummy_index_am \
dummy_seclabel \
diff --git a/src/test/modules/csnsnapshot/Makefile b/src/test/modules/csnsnapshot/Makefilenew file mode 100644index 0000000000..e7c1195d0f--- /dev/null+++ b/src/test/modules/csnsnapshot/Makefile@@ -0,0 +1,22 @@+# src/test/modules/csnsnapshot/Makefile++REGRESS_OPTS = --temp-config=$(top_srcdir)/src/test/modules/csnsnapshot/csn_snapshot.conf+NO_INSTALLCHECK = 1++TAP_TESTS = 1+EXTRA_INSTALL=contrib/postgres_fdw++ifndef CSN_ALL+PROVE_TESTS=t/001_base.pl t/002_standby.pl t/003_time_skew.pl+endif++ifdef USE_PGXS+PG_CONFIG = pg_config+PGXS := $(shell $(PG_CONFIG) --pgxs)+include $(PGXS)+else+subdir = src/test/modules/csnsnapshot+top_builddir = ../../../..+include $(top_builddir)/src/Makefile.global+include $(top_srcdir)/contrib/contrib-global.mk+endifdiff --git a/src/test/modules/csnsnapshot/csn_snapshot.conf b/src/test/modules/csnsnapshot/csn_snapshot.confnew file mode 100644index 0000000000..e9d3c35756--- /dev/null+++ b/src/test/modules/csnsnapshot/csn_snapshot.conf@@ -0,0 +1 @@+track_commit_timestamp = ondiff --git a/src/test/modules/csnsnapshot/t/001_base.pl b/src/test/modules/csnsnapshot/t/001_base.plnew file mode 100644index 0000000000..a89b1beeea--- /dev/null+++ b/src/test/modules/csnsnapshot/t/001_base.pl@@ -0,0 +1,119 @@+# Single-node test: value can be set, and is still present after recovery++use strict;+use warnings;++use TestLib;+use Test::More tests => 6;+use PostgresNode;++my $node = get_new_node('csntest');+$node->init;+$node->append_conf('postgresql.conf', qq{+ enable_csn_snapshot = on+ csn_snapshot_defer_time = 10+ max_prepared_transactions = 10+ });++my $result1;+my $result2;+$node->start;++# Check CSN increased monotonically after restart+$result1 = $node->safe_psql('postgres', "SHOW csn_time_shift");+note("csn_time_shift: $result1");+$result1 = $node->safe_psql('postgres', 'SELECT pg_csn_snapshot_export()');+note("Snapshot CSN: $result1");+$node->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = -100");+$node->restart;+$result2 = $node->safe_psql('postgres', 'SHOW csn_time_shift');+note("csn_time_shift: $result2");+$result2 = $node->safe_psql('postgres', 'SELECT pg_csn_snapshot_export()');+note("Snapshot CSN after restart: $result2");+is($result1 < $result2, 1, 'CSN monotonically increases');++$node->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 0");+$node->restart;++# Create a table+$node->safe_psql('postgres', 'create table t1(i int, j int)');++# insert test record+$node->safe_psql('postgres', 'insert into t1 values(1,1)');+# export csn snapshot+my $test_snapshot = $node->safe_psql('postgres', 'select pg_csn_snapshot_export()');+# insert test record+$node->safe_psql('postgres', 'insert into t1 values(2,1)');++my $count1 = $node->safe_psql('postgres', "select count(*) from t1");+is($count1, '2', 'Get right number in normal query');+my $count2 = $node->safe_psql('postgres', "+ begin transaction isolation level repeatable read;+ select pg_csn_snapshot_import($test_snapshot);+ select count(*) from t1;+ commit;"+ );++is($count2, '+1', 'Get right number in csn import query');++#prepare transaction test+$node->safe_psql('postgres', "+ begin;+ insert into t1 values(3,1);+ insert into t1 values(3,2);+ prepare transaction 'pt3';+ ");+$node->safe_psql('postgres', "+ begin;+ insert into t1 values(4,1);+ insert into t1 values(4,2);+ prepare transaction 'pt4';+ ");+$node->safe_psql('postgres', "+ begin;+ insert into t1 values(5,1);+ insert into t1 values(5,2);+ prepare transaction 'pt5';+ ");+$node->safe_psql('postgres', "+ begin;+ insert into t1 values(6,1);+ insert into t1 values(6,2);+ prepare transaction 'pt6';+ ");+$node->safe_psql('postgres', "commit prepared 'pt4';");++# restart with enable_csn_snapshot off+$node->append_conf('postgresql.conf', "enable_csn_snapshot = off");+$node->restart;+$node->safe_psql('postgres', "+ insert into t1 values(7,1);+ insert into t1 values(7,2);+ ");+$node->safe_psql('postgres', "commit prepared 'pt3';");+$count1 = $node->safe_psql('postgres', "select count(*) from t1");+is($count1, '8', 'Get right number in normal query');+++# restart with enable_csn_snapshot on+$node->append_conf('postgresql.conf', "enable_csn_snapshot = on");+$node->restart;+$node->safe_psql('postgres', "+ insert into t1 values(8,1);+ insert into t1 values(8,2);+ ");+$node->safe_psql('postgres', "commit prepared 'pt5';");+$count1 = $node->safe_psql('postgres', "select count(*) from t1");+is($count1, '12', 'Get right number in normal query');++# restart with enable_csn_snapshot off+$node->append_conf('postgresql.conf', "enable_csn_snapshot = on");+$node->restart;+$node->safe_psql('postgres', "+ insert into t1 values(9,1);+ insert into t1 values(9,2);+ ");+$node->safe_psql('postgres', "commit prepared 'pt6';");+$count1 = $node->safe_psql('postgres', "select count(*) from t1");+is($count1, '16', 'Get right number in normal query');diff --git a/src/test/modules/csnsnapshot/t/002_standby.pl b/src/test/modules/csnsnapshot/t/002_standby.plnew file mode 100644index 0000000000..b7c4ea93b2--- /dev/null+++ b/src/test/modules/csnsnapshot/t/002_standby.pl@@ -0,0 +1,66 @@+# Test simple scenario involving a standby++use strict;+use warnings;++use TestLib;+use Test::More tests => 6;+use PostgresNode;++my $bkplabel = 'backup';+my $master = get_new_node('master');+$master->init(allows_streaming => 1);++$master->append_conf(+ 'postgresql.conf', qq{+ enable_csn_snapshot = on+ max_wal_senders = 5+ });+$master->start;+$master->backup($bkplabel);++my $standby = get_new_node('standby');+$standby->init_from_backup($master, $bkplabel, has_streaming => 1);+$standby->start;++$master->safe_psql('postgres', "create table t1(i int, j int)");++my $guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot');+is($guc_on_master, 'on', "GUC on master");++my $guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot');+is($guc_on_standby, 'on', "GUC on standby");++$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off');+$master->restart;++$guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot');+is($guc_on_master, 'off', "GUC off master");++$guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot');+is($guc_on_standby, 'on', "GUC on standby");++# We consume a large number of transaction,for skip page+for my $i (1 .. 4096) #4096+{+ $master->safe_psql('postgres', "insert into t1 values(1,$i)");+}+$master->safe_psql('postgres', "select pg_sleep(2)");+$master->append_conf('postgresql.conf', 'enable_csn_snapshot = on');+$master->restart;++my $count_standby = $standby->safe_psql('postgres', 'select count(*) from t1');+is($count_standby, '4096', "Ok for siwtch xid-base > csn-base"); #4096++# We consume a large number of transaction,for skip page+for my $i (1 .. 4096) #4096+{+ $master->safe_psql('postgres', "insert into t1 values(1,$i)");+}+$master->safe_psql('postgres', "select pg_sleep(2)");++$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off');+$master->restart;++$count_standby = $standby->safe_psql('postgres', 'select count(*) from t1');+is($count_standby, '8192', "Ok for siwtch csn-base > xid-base"); #8192
\ No newline at end of file
diff --git a/src/test/modules/csnsnapshot/t/003_time_skew.pl b/src/test/modules/csnsnapshot/t/003_time_skew.plnew file mode 100644index 0000000000..b01ad1a7c0--- /dev/null+++ b/src/test/modules/csnsnapshot/t/003_time_skew.pl@@ -0,0 +1,211 @@+use strict;+use warnings;++use TestLib;+use Test::More tests => 13;+use PostgresNode;++my $node1 = get_new_node('csn1');+$node1->init;+$node1->append_conf('postgresql.conf', qq{+ enable_csn_snapshot = on+ csn_snapshot_defer_time = 10+ max_prepared_transactions = 10+ csn_time_shift = 0+ shared_preload_libraries = 'postgres_fdw'+ postgres_fdw.use_csn_snapshots = true+ });+$node1->start;+my $node2 = get_new_node('csn2');+$node2->init;+$node2->append_conf('postgresql.conf', qq{+ enable_csn_snapshot = on+ csn_snapshot_defer_time = 10+ max_prepared_transactions = 10+ csn_time_shift = 0+ shared_preload_libraries = 'postgres_fdw'+ postgres_fdw.use_csn_snapshots = true+ });+$node2->start;++$node1->safe_psql('postgres', "+ CREATE EXTENSION postgres_fdw;+ CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node2->port."');+ CREATE USER MAPPING FOR PUBLIC SERVER remote;+ CREATE TABLE summary(value int, ntrans int);+ INSERT INTO summary (value, ntrans) VALUES (0, 0);+");+$node2->safe_psql('postgres', "+ CREATE EXTENSION postgres_fdw;+ CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node1->port."');+ CREATE USER MAPPING FOR PUBLIC SERVER remote;+ CREATE FOREIGN TABLE summary(value int, ntrans int) SERVER remote;+");++$node1->safe_psql('postgres', "+ CREATE TABLE t (id int, payload int) PARTITION BY HASH(id);+ CREATE TABLE t_0 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0);+ CREATE FOREIGN TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1) SERVER remote;+");+$node2->safe_psql('postgres', "+ CREATE TABLE t (id serial, payload int) PARTITION BY HASH(id);+ CREATE TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1);+ CREATE FOREIGN TABLE t_0 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0) SERVER remote;+");++$node1->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 1 FROM generate_series(1,100) AS gs)");+$node2->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 2 FROM generate_series(101,200) AS gs)");+my $count1 = $node1->safe_psql('postgres', "SELECT SUM(payload) FROM t");+my $count2 = $node2->safe_psql('postgres', "SELECT SUM(payload) FROM t");+is( (($count1 == 300) and ($count1 == $count2)), 1, 'Correct insert');++# ##############################################################################+#+# Basic test. Check REPEATABLE READ anomaly.+# ntrans is needed to control that some transactions were committed.+#+# ##############################################################################++my $q1 = File::Temp->new();+append_to_file($q1, q{+ START TRANSACTION ISOLATION LEVEL REPEATABLE READ;+ UPDATE summary SET value = value + (SELECT SUM(payload) FROM t);+ UPDATE summary SET value = value - (SELECT SUM(payload) FROM t);+ UPDATE summary SET ntrans = ntrans + 1;+ COMMIT;+});+my $q2 = File::Temp->new();+append_to_file($q2, q{+ BEGIN;+ \set pl random(-100, 100)+ \set id random(1, 200)+ UPDATE t SET payload = :pl WHERE id = :id;+ COMMIT;+});++my $seconds = 5;+my $pgb_handle1 = $node1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $q1, 'postgres' );+my $pgb_handle2 = $node2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $q2, 'postgres' );+$node1->pgbench_await($pgb_handle1);+$node2->pgbench_await($pgb_handle2);++$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary");+$count2 = $node2->safe_psql('postgres', "SELECT SUM(value) FROM summary");+my $ntrans = $node2->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary");+note("$count1, $count2, $ntrans");+is( ( ($ntrans > 0) and ($count1 == 0) and ($count1 == $count2)), 1, 'Correct update');++# ##############################################################################+#+# Test on 'snapshot too old'+#+# ##############################################################################+$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0;");+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 20");+$node1->restart();++# READ COMMITTED transactions ignores the time skew.+$node2->psql('postgres', "UPDATE summary SET ntrans = 1");+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");+note("$ntrans");+is( $ntrans, 1, 'Read committed behavior if snapshot turn sour');++# But REPEATABLE READ transactions isn't+my $err = '';+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 40");+$node1->restart();+$node2->psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 2; COMMIT;", stderr => \$err);+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");+note("$ntrans");+is( (($ntrans == 1) and (index($err, 'csn snapshot too old') != -1)), 1, 'Read committed can\'t update if snapshot turn sour');++# ##############################################################################+#+# Test on issue #1:+# 'xact confirmed as committed, so any following xact must see its effects'.+#+# ##############################################################################+$node1->safe_psql('postgres', "delete from t");+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 45");+$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 40");+$node1->restart();+$node2->restart();++my $st_sec; my $end_sec;+my $time_diff;++$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1), (3,1); COMMIT;");+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");+is( $ntrans, 2, 'Slow node can see mix node data change');+$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");+is( $ntrans, 2, 'Fast node can see mix node data change');++$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1); COMMIT;");+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");+is( $ntrans, 3, 'CURRENTLY FAILED:Data change to fast node on slow node, and slow node can see data change');++# READ COMMITED mode ignores the time skew.+$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 1");+$ntrans = $node2->safe_psql('postgres', "SELECT ntrans FROM summary");+note("ntrans: $ntrans\n");+is( $ntrans, 1, 'See committed values in the READ COMMITTED mode');++# Access from the future+$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = ntrans + 1; COMMIT;");+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");+note("ntrans: $ntrans\n");+is( $ntrans, 1, 'Do not see values, committed in the future at the REPEATABLE READ mode');++# But...+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 50");+$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 55");+$node1->restart();+$node2->restart();++# Check READ COMMITED mode+$node2->safe_psql('postgres', "UPDATE summary SET ntrans = 2");+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");+note("ntrans: $ntrans\n");+is( $ntrans, 2, 'See committed values in the READ COMMITTED mode, step 2');++# Node from the future will wait for a time before UPDATE table.+($st_sec) = localtime();+$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 3; COMMIT;");+($end_sec) = localtime(); $time_diff = $end_sec - $st_sec;+$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");+note("ntrans: $ntrans, Test time: $time_diff seconds");+is( ($ntrans == 3), 1, 'The test execution time correlates with the time offset.');++# Node from the future will wait for a time before SELECT from a table.+$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 4; COMMIT;");+($st_sec) = localtime();+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");+($end_sec) = localtime(); $time_diff = $end_sec - $st_sec;+note("ntrans: $ntrans, Test time: $time_diff seconds ($end_sec, $st_sec)");+is( ($ntrans == 4), 1, 'See values, committed in the past. The test execution time correlates with the time offset.');++$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0, value = 0");+$q1 = File::Temp->new();+append_to_file($q1, q{+ UPDATE summary SET value = value + 1, ntrans = ntrans + 1;+});+$q2 = File::Temp->new();+append_to_file($q2, q{+ START TRANSACTION ISOLATION LEVEL REPEATABLE READ;+ UPDATE summary SET value = value + (SELECT SUM(ntrans) FROM summary);+ UPDATE summary SET value = value - (SELECT SUM(ntrans) FROM summary);+ COMMIT;+});+$seconds = 3;+$pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q1, 'postgres' );+$pgb_handle2 = $node2->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q2, 'postgres' );+$node1->pgbench_await($pgb_handle1);+$node2->pgbench_await($pgb_handle2);++$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary");+$count2 = $node1->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary");+note("$count1, $count2");+is( ( ($count1 > 0) and ($count1 == $count2)), 1, 'Skew test');++$node1->stop();+$node2->stop();diff --git a/src/test/modules/csnsnapshot/t/004_read_committed.pl b/src/test/modules/csnsnapshot/t/004_read_committed.plnew file mode 100644index 0000000000..0918212bd8--- /dev/null+++ b/src/test/modules/csnsnapshot/t/004_read_committed.pl@@ -0,0 +1,97 @@+use strict;+use warnings;++use TestLib;+use Test::More tests => 2;+use PostgresNode;++my $node1 = get_new_node('csn1');+$node1->init;+$node1->append_conf('postgresql.conf', qq{+ max_prepared_transactions = 20+ shared_preload_libraries = 'postgres_fdw'+ enable_csn_snapshot = on+ csn_snapshot_defer_time = 10+ postgres_fdw.use_csn_snapshots = true+ csn_time_shift = 0+ });+$node1->start;+my $node2 = get_new_node('csn2');+$node2->init;+$node2->append_conf('postgresql.conf', qq{+ max_prepared_transactions = 20+ shared_preload_libraries = 'postgres_fdw'+ enable_csn_snapshot = on+ csn_snapshot_defer_time = 10+ postgres_fdw.use_csn_snapshots = true+ csn_time_shift = 0+ });+$node2->start;++# Create foreign servers+$node1->safe_psql('postgres', "+ CREATE EXTENSION postgres_fdw;+ CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node2->port."');+ CREATE USER MAPPING FOR PUBLIC SERVER remote;+");+$node2->safe_psql('postgres', "+ CREATE EXTENSION postgres_fdw;+ CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node1->port."');+ CREATE USER MAPPING FOR PUBLIC SERVER remote;+");++# Create sharded table+$node1->safe_psql('postgres', "+ CREATE TABLE dept1(name TEXT);+ CREATE FOREIGN TABLE dept2 (name TEXT) SERVER remote;+");+$node2->safe_psql('postgres', "+ CREATE TABLE dept2(name TEXT);+ CREATE FOREIGN TABLE dept1 (name TEXT) SERVER remote;+ CREATE TABLE results(success_tx int);+ INSERT INTO results (success_tx) VALUES (0);+");++# Fill the table+$node1->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Jonathan')");+$node1->safe_psql('postgres', "INSERT INTO dept2 (name) VALUES ('Hoshi')");+$node2->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Leonard')");+my $count1 = $node1->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a");+my $count2 = $node2->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a");+note("$count1, $count2");+is( (($count1 == 3) and ($count1 == $count2)), 1, 'Correct insert');++# Queries+my $q1 = File::Temp->new();+append_to_file($q1, q{+ BEGIN;+ SELECT count(*) AS cnt FROM dept1; \gset+ \if :cnt > 0+ INSERT INTO dept2 (SELECT * FROM dept1);+ DELETE FROM dept1;+ \else+ INSERT INTO dept1 (SELECT * FROM dept2);+ DELETE FROM dept2;+ \endif++ COMMIT;+});+my $q2 = File::Temp->new();+append_to_file($q2, q{+ SELECT count(*) AS cnt FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a; \gset+ \if :cnt = 3+ UPDATE results SET success_tx = success_tx + 1;+ \endif+});+my $transactions = 1000;+my $pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -t => $transactions, -f => $q1, 'postgres' );+my $pgb_handle2 = $node2->pgbench_async(-n, -c => 20, -t => $transactions, -f => $q2, 'postgres' );+$node1->pgbench_await($pgb_handle1);+$node2->pgbench_await($pgb_handle2);++$count2 = $node2->safe_psql('postgres', "SELECT success_tx FROM results");+note("$count2");+is( $count2, 20*$transactions, 'Correct READ COMMITTED updates');++$node1->stop();+$node2->stop();--
2.25.1
[text/x-patch] 0004-Clock-SI-implementation.patch (33.5K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/5-0004-Clock-SI-implementation.patch)
download | inline diff:
From 52425146d9484b1b19197a5b362ce9dfd2d69b23 Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Mon, 22 Mar 2021 16:37:39 +0500
Subject: [PATCH 4/4] Clock SI implementation
---
contrib/postgres_fdw/Makefile | 9 +
contrib/postgres_fdw/connection.c | 323 ++++++++++++++++--
contrib/postgres_fdw/postgres_fdw.c | 12 +
contrib/postgres_fdw/postgres_fdw.h | 2 +
.../postgres_fdw/t/001_bank_coordinator.pl | 264 ++++++++++++++
.../postgres_fdw/t/002_bank_participant.pl | 240 +++++++++++++
src/backend/access/transam/xact.c | 7 +
src/include/catalog/pg_proc.dat | 3 +
src/test/perl/PostgresNode.pm | 35 ++
9 files changed, 871 insertions(+), 24 deletions(-)
create mode 100644 contrib/postgres_fdw/t/001_bank_coordinator.pl
create mode 100644 contrib/postgres_fdw/t/002_bank_participant.pl
diff --git a/contrib/postgres_fdw/Makefile b/contrib/postgres_fdw/Makefileindex c1b0cad453..539b7c4c55 100644--- a/contrib/postgres_fdw/Makefile+++ b/contrib/postgres_fdw/Makefile@@ -29,3 +29,12 @@ top_builddir = ../..
include $(top_builddir)/src/Makefile.global
include $(top_srcdir)/contrib/contrib-global.mk
endif
++# Global makefile will do temp-install for 'check'. Since REGRESS is defined,+# PGXS (included from contrib-global.mk or directly) will care to add+# postgres_fdw to it as EXTRA_INSTALL and build pg_regress. It will also+# actually run pg_regress, so the only thing left is tap tests.+check: tapcheck++tapcheck: temp-install+ $(prove_check)diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.cindex ee0b4acf0b..0fadef60ac 100644--- a/contrib/postgres_fdw/connection.c+++ b/contrib/postgres_fdw/connection.c@@ -12,8 +12,10 @@
*/
#include "postgres.h"
+#include "access/csn_snapshot.h"
#include "access/htup_details.h"
#include "access/xact.h"
+#include "access/xlog.h" /* GetSystemIdentifier() */
#include "catalog/pg_user_mapping.h"
#include "commands/defrem.h"
#include "funcapi.h"
@@ -28,6 +30,8 @@
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
+#include "utils/snapmgr.h"+#include "utils/snapshot.h"
#include "utils/syscache.h"
/*
@@ -69,6 +73,21 @@ typedef struct ConnCacheEntry
*/
static HTAB *ConnectionHash = NULL;
+/*+ * FdwTransactionState+ *+ * Holds number of open remote transactions and shared state+ * needed for all connection entries.+ */+typedef struct FdwTransactionState+{+ char *gid;+ int nparticipants;+ CSN csn;+ bool two_phase_commit;+} FdwTransactionState;+static FdwTransactionState *fdwTransState;+
/* for assigning cursor numbers and prepared statement numbers */
static unsigned int cursor_number = 0;
static unsigned int prep_stmt_number = 0;
@@ -76,6 +95,9 @@ static unsigned int prep_stmt_number = 0;
/* tracks whether any work is needed in callback functions */
static bool xact_got_connection = false;
+/* counter of prepared tx made by this backend */+static int two_phase_xact_count = 0;+
/*
* SQL functions
*/
@@ -92,6 +114,7 @@ static void configure_remote_session(PGconn *conn);
static void do_sql_command(PGconn *conn, const char *sql);
static void begin_remote_xact(ConnCacheEntry *entry);
static void pgfdw_xact_callback(XactEvent event, void *arg);
+static void deallocate_prepared_stmts(ConnCacheEntry *entry);
static void pgfdw_subxact_callback(SubXactEvent event,
SubTransactionId mySubid,
SubTransactionId parentSubid,
@@ -148,6 +171,15 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
pgfdw_inval_callback, (Datum) 0);
}
+ /* allocate FdwTransactionState */+ if (fdwTransState == NULL)+ {+ MemoryContext oldcxt;+ oldcxt = MemoryContextSwitchTo(CacheMemoryContext);+ fdwTransState = palloc0(sizeof(FdwTransactionState));+ MemoryContextSwitchTo(oldcxt);+ }+
/* Set flag that we did GetConnection during the current transaction */
xact_got_connection = true;
@@ -531,7 +563,8 @@ configure_remote_session(PGconn *conn)
}
/*
- * Convenience subroutine to issue a non-data-returning SQL command to remote+ * Convenience subroutine to issue a non-data-returning SQL command or+ * statement to remote node.
*/
static void
do_sql_command(PGconn *conn, const char *sql)
@@ -541,7 +574,8 @@ do_sql_command(PGconn *conn, const char *sql)
if (!PQsendQuery(conn, sql))
pgfdw_report_error(ERROR, NULL, conn, false, sql);
res = pgfdw_get_result(conn, sql);
- if (PQresultStatus(res) != PGRES_COMMAND_OK)+ if (PQresultStatus(res) != PGRES_COMMAND_OK &&+ PQresultStatus(res) != PGRES_TUPLES_OK)
pgfdw_report_error(ERROR, res, conn, true, sql);
PQclear(res);
}
@@ -569,6 +603,10 @@ begin_remote_xact(ConnCacheEntry *entry)
elog(DEBUG3, "starting remote transaction on connection %p",
entry->conn);
+ if (UseCSNSnapshots && (/*!IsolationUsesXactSnapshot() ||*/+ IsolationIsSerializable()))+ elog(ERROR, "Global snapshots support only REPEATABLE READ");+
if (IsolationIsSerializable())
sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
else
@@ -577,6 +615,23 @@ begin_remote_xact(ConnCacheEntry *entry)
do_sql_command(entry->conn, sql);
entry->xact_depth = 1;
entry->changing_xact_state = false;
++ if (UseCSNSnapshots && IsolationUsesXactSnapshot())+ {+ char import_sql[128];++ /* Export our snapshot */+ if (fdwTransState->csn == 0)+ fdwTransState->csn = ExportCSNSnapshot();++ snprintf(import_sql, sizeof(import_sql),+ "SELECT pg_csn_snapshot_import("UINT64_FORMAT")",+ fdwTransState->csn);++ do_sql_command(entry->conn, import_sql);+ }++ fdwTransState->nparticipants += 1;
}
/*
@@ -784,6 +839,98 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
PG_END_TRY();
}
+/* Callback typedef for BroadcastStmt */+typedef bool (*BroadcastCmdResHandler) (PGresult *result, void *arg);++/*+ * Broadcast sql in parallel to all ConnectionHash entries+ */+static bool+BroadcastStmt(char const * sql, unsigned expectedStatus,+ BroadcastCmdResHandler handler, void *arg)+{+ HASH_SEQ_STATUS scan;+ ConnCacheEntry *entry;+ bool allOk = true;++ /* Broadcast sql */+ hash_seq_init(&scan, ConnectionHash);+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))+ {+ pgfdw_reject_incomplete_xact_state_change(entry);++ if (entry->xact_depth > 0 && entry->conn != NULL)+ {+ if (!PQsendQuery(entry->conn, sql))+ {+ PGresult *res = PQgetResult(entry->conn);++ elog(WARNING, "Failed to send command %s", sql);+ pgfdw_report_error(WARNING, res, entry->conn, true, sql);+ PQclear(res);+ }+ }+ }++ /* Collect responses */+ hash_seq_init(&scan, ConnectionHash);+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))+ {+ if (entry->xact_depth > 0 && entry->conn != NULL)+ {+ PGresult *result = PQgetResult(entry->conn);++ if (PQresultStatus(result) != expectedStatus ||+ (handler && !handler(result, arg)))+ {+ elog(WARNING,+ "Failed command %s: status=%d, expected status=%d",+ sql, PQresultStatus(result), expectedStatus);+ pgfdw_report_error(ERROR, result, entry->conn, true, sql);+ allOk = false;+ }+ PQclear(result);+ PQgetResult(entry->conn); /* consume NULL result */+ }+ }++ return allOk;+}++/* Wrapper for broadcasting commands */+static bool+BroadcastCmd(char const *sql)+{+ return BroadcastStmt(sql, PGRES_COMMAND_OK, NULL, NULL);+}++/* Wrapper for broadcasting statements */+static bool+BroadcastFunc(char const *sql)+{+ return BroadcastStmt(sql, PGRES_TUPLES_OK, NULL, NULL);+}++/* Callback for selecting maximal csn */+static bool+MaxCsnCB(PGresult *result, void *arg)+{+ char *resp;+ CSN *max_csn = (CSN *) arg;+ CSN csn = 0;++ resp = PQgetvalue(result, 0, 0);++ if (resp == NULL || (*resp) == '\0' ||+ sscanf(resp, UINT64_FORMAT, &csn) != 1)+ return false;++ if (*max_csn < csn)+ *max_csn = csn;++ return true;+}+
/*
* pgfdw_xact_callback --- cleanup at main-transaction end.
*
@@ -801,6 +948,104 @@ pgfdw_xact_callback(XactEvent event, void *arg)
if (!xact_got_connection)
return;
+ /* Handle possible two-phase commit */+ if (event == XACT_EVENT_PARALLEL_PRE_COMMIT || event == XACT_EVENT_PRE_COMMIT)+ {+ bool include_local_tx = false;++ /* Should we take into account this node? */+ if (TransactionIdIsValid(GetCurrentTransactionIdIfAny()))+ {+ include_local_tx = true;+ fdwTransState->nparticipants += 1;+ }++ /* Switch to 2PC mode there were more than one participant */+ if (UseCSNSnapshots && fdwTransState->nparticipants > 1)+ fdwTransState->two_phase_commit = true;++ /*+ * Data change to fast node on slow node, and slow node can see data change.+ * In order to implement above we should change code here.+ */+ if (fdwTransState->two_phase_commit)+ {+ CSN max_csn = InProgressCSN;+ CSN my_csn = InProgressCSN;+ bool res;+ char *sql;++ fdwTransState->gid = psprintf("pgfdw:%lld:%llu:%d:%u:%d:%d",+ (long long) GetCurrentTimestamp(),+ (long long) GetSystemIdentifier(),+ MyProcPid,+ GetCurrentTransactionIdIfAny(),+ ++two_phase_xact_count,+ fdwTransState->nparticipants);++ /* Broadcast PREPARE */+ sql = psprintf("PREPARE TRANSACTION '%s'", fdwTransState->gid);+ res = BroadcastCmd(sql);++ if (IsolationUsesXactSnapshot())+ {+ if (!res)+ goto error;++ /* Broadcast pg_csn_snapshot_prepare() */+ if (include_local_tx)+ my_csn = CSNSnapshotPrepareCurrent();++ sql = psprintf("SELECT pg_csn_snapshot_prepare('%s')",+ fdwTransState->gid);+ res = BroadcastStmt(sql, PGRES_TUPLES_OK, MaxCsnCB, &max_csn);+ if (!res)+ goto error;++ /* select maximal global csn */+ if (include_local_tx && my_csn > max_csn)+ max_csn = my_csn;++ /*+ * We should always notice local node to update the csn, so local can+ * see the change next transaction.+ */+ if (include_local_tx)+ CSNSnapshotAssignCurrent(max_csn);+ else+ /*+ * Read-only transactions haven't assigned xid csn. We only+ * increase the last csn value.+ */+ GenerateCSN(max_csn);++ sql = psprintf("SELECT pg_csn_snapshot_assign('%s',"UINT64_FORMAT")",+ fdwTransState->gid, max_csn);+ res = BroadcastFunc(sql);+ }++error:+ if (!res)+ {+ sql = psprintf("ABORT PREPARED '%s'", fdwTransState->gid);+ BroadcastCmd(sql);+ elog(ERROR, "Failed to PREPARE transaction on remote node");+ }++ /*+ * Do not fall down. Consequent COMMIT event will clean thing up.+ */+ return;+ }+ }++ /* COMMIT open transaction of we were doing 2PC */+ if (fdwTransState->two_phase_commit &&+ (event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_COMMIT))+ {+ BroadcastCmd(psprintf("COMMIT PREPARED '%s'", fdwTransState->gid));+ }+
/*
* Scan all connection cache entries to find open remote transactions, and
* close them.
@@ -808,8 +1053,6 @@ pgfdw_xact_callback(XactEvent event, void *arg)
hash_seq_init(&scan, ConnectionHash);
while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
{
- PGresult *res;-
/* Ignore cache entry if no open connection right now */
if (entry->conn == NULL)
continue;
@@ -826,6 +1069,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
{
case XACT_EVENT_PARALLEL_PRE_COMMIT:
case XACT_EVENT_PRE_COMMIT:
+ Assert(!fdwTransState->two_phase_commit);
/*
* If abort cleanup previously failed for this connection,
@@ -838,28 +1082,22 @@ pgfdw_xact_callback(XactEvent event, void *arg)
do_sql_command(entry->conn, "COMMIT TRANSACTION");
entry->changing_xact_state = false;
- /*- * If there were any errors in subtransactions, and we- * made prepared statements, do a DEALLOCATE ALL to make- * sure we get rid of all prepared statements. This is- * annoying and not terribly bulletproof, but it's- * probably not worth trying harder.- *- * DEALLOCATE ALL only exists in 8.3 and later, so this- * constrains how old a server postgres_fdw can- * communicate with. We intentionally ignore errors in- * the DEALLOCATE, so that we can hobble along to some- * extent with older servers (leaking prepared statements- * as we go; but we don't really support update operations- * pre-8.3 anyway).- */- if (entry->have_prep_stmt && entry->have_error)+ if (UseCSNSnapshots)
{
- res = PQexec(entry->conn, "DEALLOCATE ALL");- PQclear(res);+ CSN csn = InvalidCSN;+ PGresult *res;++ res = pgfdw_exec_query(entry->conn, "SELECT pg_current_csn()");+ if (PQresultStatus(res) == PGRES_TUPLES_OK)+ {+ sscanf(PQgetvalue(res, 0, 0), "%lu", &csn);++ if (csn != InvalidCSN)+ GenerateCSN(csn);+ }
}
- entry->have_prep_stmt = false;- entry->have_error = false;++ deallocate_prepared_stmts(entry);
break;
case XACT_EVENT_PRE_PREPARE:
@@ -878,6 +1116,11 @@ pgfdw_xact_callback(XactEvent event, void *arg)
break;
case XACT_EVENT_PARALLEL_COMMIT:
case XACT_EVENT_COMMIT:
+ if (fdwTransState->two_phase_commit)+ deallocate_prepared_stmts(entry);+ else /* Pre-commit should have closed the open transaction */+ elog(ERROR, "missed cleaning up connection during pre-commit");+ break;
case XACT_EVENT_PREPARE:
/* Pre-commit should have closed the open transaction */
elog(ERROR, "missed cleaning up connection during pre-commit");
@@ -975,6 +1218,38 @@ pgfdw_xact_callback(XactEvent event, void *arg)
/* Also reset cursor numbering for next transaction */
cursor_number = 0;
++ /* Reset fdwTransState */+ memset(fdwTransState, '\0', sizeof(FdwTransactionState));+}++/*+ * If there were any errors in subtransactions, and we+ * made prepared statements, do a DEALLOCATE ALL to make+ * sure we get rid of all prepared statements. This is+ * annoying and not terribly bulletproof, but it's+ * probably not worth trying harder.+ *+ * DEALLOCATE ALL only exists in 8.3 and later, so this+ * constrains how old a server postgres_fdw can+ * communicate with. We intentionally ignore errors in+ * the DEALLOCATE, so that we can hobble along to some+ * extent with older servers (leaking prepared statements+ * as we go; but we don't really support update operations+ * pre-8.3 anyway).+ */+static void+deallocate_prepared_stmts(ConnCacheEntry *entry)+{+ PGresult *res;++ if (entry->have_prep_stmt && entry->have_error)+ {+ res = PQexec(entry->conn, "DEALLOCATE ALL");+ PQclear(res);+ }+ entry->have_prep_stmt = false;+ entry->have_error = false;
}
/*
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.cindex 35b48575c5..f59f78a770 100644--- a/contrib/postgres_fdw/postgres_fdw.c+++ b/contrib/postgres_fdw/postgres_fdw.c@@ -312,6 +312,9 @@ typedef struct
List *already_used; /* expressions already dealt with */
} ec_member_foreign_arg;
+bool UseCSNSnapshots;+void _PG_init(void);+
/*
* SQL functions
*/
@@ -6813,3 +6816,12 @@ get_batch_size_option(Relation rel)
return batch_size;
}
++void+_PG_init(void)+{+ DefineCustomBoolVariable("postgres_fdw.use_csn_snapshots",+ "Use global snapshots for FDW transactions", NULL,+ &UseCSNSnapshots, false, PGC_USERSET, 0, NULL,+ NULL, NULL);+}
\ No newline at end of file
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.hindex 1f67b4d9fd..87fb9bf8c3 100644--- a/contrib/postgres_fdw/postgres_fdw.h+++ b/contrib/postgres_fdw/postgres_fdw.h@@ -211,4 +211,6 @@ extern const char *get_jointype_name(JoinType jointype);
extern bool is_builtin(Oid objectId);
extern bool is_shippable(Oid objectId, Oid classId, PgFdwRelationInfo *fpinfo);
+extern bool UseCSNSnapshots;+
#endif /* POSTGRES_FDW_H */
diff --git a/contrib/postgres_fdw/t/001_bank_coordinator.pl b/contrib/postgres_fdw/t/001_bank_coordinator.plnew file mode 100644index 0000000000..a32637f581--- /dev/null+++ b/contrib/postgres_fdw/t/001_bank_coordinator.pl@@ -0,0 +1,264 @@+use strict;+use warnings;++use PostgresNode;+use TestLib;+use Test::More tests => 3;++my $master = get_new_node("master");+$master->init;+$master->append_conf('postgresql.conf', qq(+ max_prepared_transactions = 30+ log_checkpoints = true+ postgres_fdw.use_csn_snapshots = on+ enable_csn_snapshot = on+ default_transaction_isolation = 'REPEATABLE READ'+));+$master->start;++my $shard1 = get_new_node("shard1");+$shard1->init;+$shard1->append_conf('postgresql.conf', qq(+ max_prepared_transactions = 30+ csn_snapshot_defer_time = 15+ enable_csn_snapshot = on+));+$shard1->start;++my $shard2 = get_new_node("shard2");+$shard2->init;+$shard2->append_conf('postgresql.conf', qq(+ max_prepared_transactions = 30+ csn_snapshot_defer_time = 15+ enable_csn_snapshot = on+));+$shard2->start;++###############################################################################+# Prepare nodes+###############################################################################++$master->safe_psql('postgres', qq[+ CREATE EXTENSION postgres_fdw;+ CREATE TABLE accounts(id integer primary key, amount integer);+ CREATE TABLE global_transactions(tx_time timestamp);+]);++foreach my $node ($shard1, $shard2)+{+ my $port = $node->port;+ my $host = $node->host;++ $node->safe_psql('postgres',+ "CREATE TABLE accounts(id integer primary key, amount integer)");++ $master->safe_psql('postgres', qq[+ CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw options(dbname 'postgres', host '$host', port '$port');+ CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts) server shard_$port options(table_name 'accounts');+ CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;+ ])+}++$shard1->safe_psql('postgres', qq[+ insert into accounts select 2*id-1, 0 from generate_series(1, 10010) as id;+ CREATE TABLE local_transactions(tx_time timestamp);+]);++$shard2->safe_psql('postgres', qq[+ insert into accounts select 2*id, 0 from generate_series(1, 10010) as id;+ CREATE TABLE local_transactions(tx_time timestamp);+]);++diag("master: @{[$master->connstr('postgres')]}");+diag("shard1: @{[$shard1->connstr('postgres')]}");+diag("shard2: @{[$shard2->connstr('postgres')]}");++###############################################################################+# pgbench scripts+###############################################################################++my $bank = File::Temp->new();+append_to_file($bank, q{+ \set id random(1, 20000)+ BEGIN;+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)+ INSERT into global_transactions SELECT now() FROM upd;+ UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);+ COMMIT;+});++my $bank1 = File::Temp->new();+append_to_file($bank1, q{+ \set id random(1, 10000)+ BEGIN;+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = (2*:id + 1) RETURNING *)+ INSERT into local_transactions SELECT now() FROM upd;+ UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 3);+ COMMIT;+});++my $bank2 = File::Temp->new();+append_to_file($bank2, q{+ \set id random(1, 10000)++ BEGIN;+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = 2*:id RETURNING *)+ INSERT into local_transactions SELECT now() FROM upd;+ UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 2);+ COMMIT;+});++###############################################################################+# Helpers+###############################################################################++sub count_and_delete_rows+{+ my ($node, $table) = @_;+ my $count;++ $count = $node->safe_psql('postgres',"select count(*) from $table");+ $node->safe_psql('postgres',"delete from $table");+ diag($node->name, ": completed $count transactions");+ return $count;+}++###############################################################################+# Concurrent global transactions+###############################################################################++my ($err, $rc);+my $started;+my $seconds = 30;+my $selects;+my $total = '0';+my $oldtotal = '0';+my $isolation_errors = 0;+++my $pgb_handle;++$pgb_handle = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );++$started = time();+$selects = 0;+while (time() - $started < $seconds)+{+ $total = $master->safe_psql('postgres', "select sum(amount) from accounts");+ if ( ($total ne $oldtotal) and ($total ne '') )+ {+ $isolation_errors++;+ $oldtotal = $total;+ diag("Isolation error. Total = $total");+ }+ if ($total ne '') { $selects++; }+}++$master->pgbench_await($pgb_handle);++# sanity check+diag("completed $selects selects");+die "no actual transactions happend" unless ( $selects > 0 &&+ count_and_delete_rows($master, 'global_transactions') > 0);++is($isolation_errors, 0, 'isolation between concurrent global transaction');++###############################################################################+# Concurrent global and local transactions+###############################################################################++my ($pgb_handle1, $pgb_handle2, $pgb_handle3);++# global txses+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );++# concurrent local+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );++$started = time();+$selects = 0;+$oldtotal = 0;+while (time() - $started < $seconds)+{+ $total = $master->safe_psql('postgres', "select sum(amount) from accounts");+ if ( ($total ne $oldtotal) and ($total ne '') )+ {+ $isolation_errors++;+ $oldtotal = $total;+ diag("Isolation error. Total = $total");+ }+ if ($total ne '') { $selects++; }+}++diag("selects = $selects");+$master->pgbench_await($pgb_handle1);+$shard1->pgbench_await($pgb_handle2);+$shard2->pgbench_await($pgb_handle3);++diag("completed $selects selects");+die "" unless ( $selects > 0 &&+ count_and_delete_rows($master, 'global_transactions') > 0 &&+ count_and_delete_rows($shard1, 'local_transactions') > 0 &&+ count_and_delete_rows($shard2, 'local_transactions') > 0);++is($isolation_errors, 0, 'isolation between concurrent global and local transactions');+++###############################################################################+# Snapshot stability+###############################################################################++my ($hashes, $hash1, $hash2);+my $stability_errors = 0;++# global txses+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );+# concurrent local+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );++$selects = 0;+$started = time();+while (time() - $started < $seconds)+{+ foreach my $node ($master, $shard1, $shard2)+ {+ ($hash1, $_, $hash2) = split "\n", $node->safe_psql('postgres', qq[+ begin isolation level repeatable read;+ select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;+ select pg_sleep(3);+ select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;+ commit;+ ]);++ if ($hash1 ne $hash2)+ {+ diag("oops");+ $stability_errors++;+ }+ elsif ($hash1 eq '' or $hash2 eq '')+ {+ die;+ }+ else+ {+ $selects++;+ }+ }+}++$master->pgbench_await($pgb_handle1);+$shard1->pgbench_await($pgb_handle2);+$shard2->pgbench_await($pgb_handle3);++die "" unless ( $selects > 0 &&+ count_and_delete_rows($master, 'global_transactions') > 0 &&+ count_and_delete_rows($shard1, 'local_transactions') > 0 &&+ count_and_delete_rows($shard2, 'local_transactions') > 0);++is($stability_errors, 0, 'snapshot is stable during concurrent global and local transactions');++$master->stop;+$shard1->stop;+$shard2->stop;diff --git a/contrib/postgres_fdw/t/002_bank_participant.pl b/contrib/postgres_fdw/t/002_bank_participant.plnew file mode 100644index 0000000000..e5a5a721c6--- /dev/null+++ b/contrib/postgres_fdw/t/002_bank_participant.pl@@ -0,0 +1,240 @@+use strict;+use warnings;++use PostgresNode;+use TestLib;+use Test::More tests => 3;++my $shard1 = get_new_node("shard1");+$shard1->init;+$shard1->append_conf('postgresql.conf', qq(+ max_prepared_transactions = 30+ postgres_fdw.use_csn_snapshots = on+ csn_snapshot_defer_time = 15+ enable_csn_snapshot = on+ default_transaction_isolation = 'REPEATABLE READ'+));+$shard1->start;++my $shard2 = get_new_node("shard2");+$shard2->init;+$shard2->append_conf('postgresql.conf', qq(+ max_prepared_transactions = 30+ postgres_fdw.use_csn_snapshots = on+ csn_snapshot_defer_time = 15+ enable_csn_snapshot = on+ default_transaction_isolation = 'REPEATABLE READ'+));+$shard2->start;++###############################################################################+# Prepare nodes+###############################################################################++my @shards = ($shard1, $shard2);++foreach my $node (@shards)+{+ $node->safe_psql('postgres', qq[+ CREATE EXTENSION postgres_fdw;+ CREATE TABLE accounts(id integer primary key, amount integer);+ CREATE TABLE accounts_local() inherits(accounts);+ CREATE TABLE global_transactions(tx_time timestamp);+ CREATE TABLE local_transactions(tx_time timestamp);+ ]);++ foreach my $neighbor (@shards)+ {+ next if ($neighbor eq $node);++ my $port = $neighbor->port;+ my $host = $neighbor->host;++ $node->safe_psql('postgres', qq[+ CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw+ options(dbname 'postgres', host '$host', port '$port');+ CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts)+ server shard_$port options(table_name 'accounts_local');+ CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;+ ]);+ }+}++$shard1->psql('postgres', "insert into accounts_local select 2*id-1, 0 from generate_series(1, 10010) as id;");+$shard2->psql('postgres', "insert into accounts_local select 2*id, 0 from generate_series(1, 10010) as id;");++###############################################################################+# pgbench scripts+###############################################################################++my $bank = File::Temp->new();+append_to_file($bank, q{+ \set id random(1, 20000)+ BEGIN;+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)+ INSERT into global_transactions SELECT now() FROM upd;+ UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);+ COMMIT;+});++###############################################################################+# Helpers+###############################################################################++sub count_and_delete_rows+{+ my ($node, $table) = @_;+ my $count;++ $count = $node->safe_psql('postgres',"select count(*) from $table");+ $node->safe_psql('postgres',"delete from $table");+ diag($node->name, ": completed $count transactions");+ return $count;+}++###############################################################################+# Concurrent global transactions+###############################################################################++my ($err, $rc);+my $started;+my $seconds = 30;+my $selects;+my $total = '0';+my $oldtotal = '0';+my $isolation_errors = 0;+my $i;+++my ($pgb_handle1, $pgb_handle2);++$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );++$started = time();+$selects = 0;+$i = 0;+while (time() - $started < $seconds)+{+ my $shard = $shard1;+ foreach my $shard (@shards)+ {+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");+ if ( ($total ne $oldtotal) and ($total ne '') )+ {+ $isolation_errors++;+ $oldtotal = $total;+ diag("$i: Isolation error. Total = $total");+ }+ if ($total ne '') { $selects++; }+ }+ $i++;+}++$shard1->pgbench_await($pgb_handle1);+$shard2->pgbench_await($pgb_handle2);++# sanity check+diag("completed $selects selects");+die "no actual transactions happend" unless ( $selects > 0 &&+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&+ count_and_delete_rows($shard2, 'global_transactions') > 0);++is($isolation_errors, 0, 'isolation between concurrent global transaction');++###############################################################################+# And do the same after soft restart+###############################################################################++$shard1->restart;+$shard2->restart;+$shard1->poll_query_until('postgres', "select 't'")+ or die "Timed out waiting for shard1 to became online";+$shard2->poll_query_until('postgres', "select 't'")+ or die "Timed out waiting for shard2 to became online";++$seconds = 15;+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );++$started = time();+$selects = 0;+$i = 0;++while (time() - $started < $seconds)+{+ my $shard = $shard1;+ foreach my $shard (@shards)+ {+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");+ if ( ($total ne $oldtotal) and ($total ne '') )+ {+ $isolation_errors++;+ $oldtotal = $total;+ diag("$i: Isolation error. Total = $total");+ }+ if ($total ne '') { $selects++; }+ }+ $i++;+}++$shard1->pgbench_await($pgb_handle1);+$shard2->pgbench_await($pgb_handle2);++# sanity check+diag("completed $selects selects");+die "no actual transactions happend" unless ( $selects > 0 &&+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&+ count_and_delete_rows($shard2, 'global_transactions') > 0);++is($isolation_errors, 0, 'isolation between concurrent global transaction after restart');++###############################################################################+# And do the same after hard restart+###############################################################################++$shard1->teardown_node;+$shard2->teardown_node;+$shard1->start;+$shard2->start;+$shard1->poll_query_until('postgres', "select 't'")+ or die "Timed out waiting for shard1 to became online";+$shard2->poll_query_until('postgres', "select 't'")+ or die "Timed out waiting for shard2 to became online";+++$seconds = 15;+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );++$started = time();+$selects = 0;+$i = 0;++while (time() - $started < $seconds)+{+ my $shard = $shard1;+ foreach my $shard (@shards)+ {+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");+ if ( ($total ne $oldtotal) and ($total ne '') )+ {+ $isolation_errors++;+ $oldtotal = $total;+ diag("$i: Isolation error. Total = $total");+ }+ if ($total ne '') { $selects++; }+ }+ $i++;+}++$shard1->pgbench_await($pgb_handle1);+$shard2->pgbench_await($pgb_handle2);++# sanity check+diag("completed $selects selects");+die "no actual transactions happend" unless ( $selects > 0 &&+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&+ count_and_delete_rows($shard2, 'global_transactions') > 0);++is($isolation_errors, 0, 'isolation between concurrent global transaction after hard restart');diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.cindex deabedbe37..d375eacc8f 100644--- a/src/backend/access/transam/xact.c+++ b/src/backend/access/transam/xact.c@@ -2103,6 +2103,13 @@ StartTransaction(void)
ShowTransactionState("StartTransaction");
}
+Datum+pg_current_csn(PG_FUNCTION_ARGS)+{+ CSN csn = GenerateCSN(InvalidCSN);++ PG_RETURN_INT64(csn);+}
/*
* CommitTransaction
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.datindex 8585464c5b..ed776e5166 100644--- a/src/include/catalog/pg_proc.dat+++ b/src/include/catalog/pg_proc.dat@@ -11424,5 +11424,8 @@
{ 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' },
+{ oid => '10005', descr => 'get current CSN',+ proname => 'pg_current_csn', provolatile => 'v', proparallel => 'u',+ prorettype => 'int8', proargtypes => '', prosrc => 'pg_current_csn' },
]
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pmindex 97e05993be..ea251be0b7 100644--- a/src/test/perl/PostgresNode.pm+++ b/src/test/perl/PostgresNode.pm@@ -2243,6 +2243,41 @@ sub pg_recvlogical_upto
}
}
+sub pgbench()+{+ my ($self, $node, @args) = @_;+ my $pgbench_handle = $self->pgbench_async($node, @args);+ $self->pgbench_await($pgbench_handle);+}++sub pgbench_async()+{+ my ($self, @args) = @_;++ my ($in, $out, $err, $rc);+ $in = '';+ $out = '';++ my @pgbench_command = (+ 'pgbench',+ -h => $self->host,+ -p => $self->port,+ @args+ );+ my $handle = IPC::Run::start(\@pgbench_command, $in, $out);+ return $handle;+}++sub pgbench_await()+{+ my ($self, $pgbench_handle) = @_;++ # During run some pgbench threads can exit (for example due to+ # serialization error). That will set non-zero returning code.+ # So don't check return code here and leave it to a caller.+ my $rc = IPC::Run::finish($pgbench_handle);+}+
=pod
=back
--
2.25.1
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: a.lepikhov@postgrespro.ru, tsunakawa.takay@fujitsu.com, masao.fujii@oss.nttdata.com, masahiko.sawada@2ndquadrant.com, amit.kapila16@gmail.com, movead.li@highgo.ca
Subject: Re: Global snapshots
In-Reply-To: <0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox