global_snapshot_xmin support in procarray

(cherry picked from commit 18190aab39dc68c87dfc27e6a9a5442e2f280298)

Author: kelvich

contrib/postgres_fdw/t/001_bank_check.pl

@@ -68,8 +68,7 @@
 my $isolation_error = 0;
 
 my $pgb_path = catfile(dirname(__FILE__), "bank.pgb");
-$master->pgbench(-n, -c => 5, -t => 10, -f => "$pgb_path", 'postgres' );
-
+$master->pgbench(-n, -c => 20, -t => 30, -f => "$pgb_path", 'postgres' );
 my $pgb_handle = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => "$pgb_path", 'postgres' );
 
 my $started = time();

contrib/postgres_fdw/t/bank.pgb

@@ -2,5 +2,6 @@
 
 BEGIN;
 UPDATE accounts SET amount = amount - 1 WHERE id = :id;
+--select pg_sleep(0.5*random());
 UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);
 COMMIT;

src/backend/access/transam/global_snapshot.c

@@ -16,6 +16,7 @@
 #include "storage/s_lock.h"
 #include "storage/spin.h"
 #include "storage/lmgr.h"
+#include "storage/procarray.h"
 #include "storage/shmem.h"
 #include "storage/ipc.h"
 #include "access/xlogdefs.h"
@@ -60,8 +61,6 @@ typedef struct
 {
 	cid_t		cid;			/* last assigned CSN; used to provide unique
 								 * ascending CSNs */
-	TransactionId oldest_xid;	/* XID of oldest transaction visible by any
-								 * active transaction (local or global) */
 	long		time_shift;		/* correction to system time */
 	volatile slock_t lock;		/* spinlock to protect access to hash table  */
 	DtmTransStatus *trans_list_head;	/* L1 list of finished transactions
@@ -94,8 +93,6 @@ static bool DtmRecordCommits = 0;
 
 DtmCurrentTrans dtm_tx; // XXXX: make static
 
-static Snapshot DtmGetSnapshot(Snapshot snapshot);
-static TransactionId DtmGetOldestXmin(Relation rel, int flags);
 static bool DtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 static void DtmAdjustOldestXid(void);
 static bool DtmDetectGlobalDeadLock(PGPROC *proc);
@@ -109,9 +106,9 @@ static void DtmDeserializeTransactionState(void* ctx);
 static TransactionManager DtmTM = {
 	PgTransactionIdGetStatus,
 	PgTransactionIdSetTreeStatus,
-	DtmGetSnapshot,
+	PgGetSnapshotData,
 	PgGetNewTransactionId,
-	DtmGetOldestXmin,
+	PgGetOldestXmin,
 	PgTransactionIdIsInProgress,
 	PgGetGlobalTransactionId,
 	DtmXidInMVCCSnapshot,
@@ -331,16 +328,16 @@ static void
 DtmAdjustOldestXid()
 {
 	DtmTransStatus *ts,
-				*prev = NULL;
+				   *prev = NULL;
+	timestamp_t		cutoff_time;
+	TransactionId   oldest_xid = InvalidTransactionId;
+	int				total = 0,
+					deleted = 0;
 
-	timestamp_t cutoff_time = dtm_get_current_time() - DtmVacuumDelay * USEC;
-	int total = 0, deleted = 0;
+	cutoff_time = dtm_get_current_time() - DtmVacuumDelay * USEC;
 
 	SpinLockAcquire(&local->lock);
 
-	for (ts = local->trans_list_head; ts != NULL; ts = ts->next)
-		total++;
-
 	for (ts = local->trans_list_head; ts != NULL && ts->cid < cutoff_time; prev = ts, ts = ts->next)
 	{
 		if (prev != NULL)
@@ -353,53 +350,24 @@ DtmAdjustOldestXid()
 	if (prev != NULL)
 		local->trans_list_head = prev;
 
-	if (ts != NULL)
-		local->oldest_xid = ts->xid;
-	else
-		local->oldest_xid = InvalidTransactionId;
+	if (local->trans_list_head)
+		oldest_xid = local->trans_list_head->xid;
 
-	SpinLockRelease(&local->lock);
-
-	// elog(LOG, "DtmAdjustOldestXid total=%d, deleted=%d, xid=%d, prev=%p, ts=%p", total, deleted, local->oldest_xid, prev, ts);
-}
-
-Snapshot
-DtmGetSnapshot(Snapshot snapshot)
-{
-	snapshot = PgGetSnapshotData(snapshot);
-	// RecentGlobalDataXmin = RecentGlobalXmin = DtmAdjustOldestXid(RecentGlobalDataXmin);
-	SpinLockAcquire(&local->lock);
-
-	if (TransactionIdIsValid(local->oldest_xid) &&
-		TransactionIdPrecedes(local->oldest_xid, RecentGlobalXmin))
-		RecentGlobalXmin = local->oldest_xid;
-
-	if (TransactionIdIsValid(local->oldest_xid) &&
-		TransactionIdPrecedes(local->oldest_xid, RecentGlobalDataXmin))
-		RecentGlobalDataXmin = local->oldest_xid;
+	for (ts = local->trans_list_head; ts != NULL; ts = ts->next)
+	{
+		if (TransactionIdPrecedes(ts->xid, oldest_xid))
+			oldest_xid = ts->xid;
+		total++;
+	}
 
 	SpinLockRelease(&local->lock);
-	return snapshot;
-}
-
-TransactionId
-DtmGetOldestXmin(Relation rel, int flags)
-{
-	TransactionId xmin = PgGetOldestXmin(rel, flags);
 
-	// xmin = DtmAdjustOldestXid(xmin);
+	ProcArraySetGlobalSnapshotXmin(oldest_xid);
 
-	SpinLockAcquire(&local->lock);
-
-	if (TransactionIdIsValid(local->oldest_xid) &&
-		TransactionIdPrecedes(local->oldest_xid, xmin))
-		xmin = local->oldest_xid;
-
-	SpinLockRelease(&local->lock);
-
-	return xmin;
+	// elog(LOG, "DtmAdjustOldestXid total=%d, deleted=%d, xid=%d, prev=%p, ts=%p", total, deleted, oldest_xid, prev, ts);
 }
 
+
 /*
  * Check tuple bisibility based on CSN of current transaction.
  * If there is no niformation about transaction with this XID, then use standard PostgreSQL visibility rules.
@@ -489,7 +457,6 @@ DtmInitialize()
 	if (!found)
 	{
 		local->time_shift = 0;
-		local->oldest_xid = FirstNormalTransactionId;
 		local->cid = dtm_get_current_time();
 		local->trans_list_head = NULL;
 		local->trans_list_tail = &local->trans_list_head;

src/backend/storage/ipc/procarray.c

@@ -93,6 +93,9 @@ typedef struct ProcArrayStruct
 	/* oldest catalog xmin of any replication slot */
 	TransactionId replication_slot_catalog_xmin;
 
+	/* xmin of oldest active global snapshot */
+	TransactionId global_snapshot_xmin;
+
 	/* indexes into allPgXact[], has PROCARRAY_MAXPROCS entries */
 	int			pgprocnos[FLEXIBLE_ARRAY_MEMBER];
 } ProcArrayStruct;
@@ -247,6 +250,7 @@ CreateSharedProcArray(void)
 		procArray->lastOverflowedXid = InvalidTransactionId;
 		procArray->replication_slot_xmin = InvalidTransactionId;
 		procArray->replication_slot_catalog_xmin = InvalidTransactionId;
+		procArray->global_snapshot_xmin = InvalidTransactionId;
 	}
 
 	allProcs = ProcGlobal->allProcs;
@@ -1407,6 +1411,7 @@ PgGetOldestXmin(Relation rel, int flags)
 
 	volatile TransactionId replication_slot_xmin = InvalidTransactionId;
 	volatile TransactionId replication_slot_catalog_xmin = InvalidTransactionId;
+	volatile TransactionId global_snapshot_xmin = InvalidTransactionId;
 
 	/*
 	 * If we're not computing a relation specific limit, or if a shared
@@ -1468,6 +1473,7 @@ PgGetOldestXmin(Relation rel, int flags)
 	/* fetch into volatile var while ProcArrayLock is held */
 	replication_slot_xmin = procArray->replication_slot_xmin;
 	replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+	global_snapshot_xmin = procArray->global_snapshot_xmin;
 
 	if (RecoveryInProgress())
 	{
@@ -1509,6 +1515,10 @@ PgGetOldestXmin(Relation rel, int flags)
 			result = FirstNormalTransactionId;
 	}
 
+	if (TransactionIdIsValid(global_snapshot_xmin) &&
+		NormalTransactionIdPrecedes(global_snapshot_xmin, result))
+		result = global_snapshot_xmin;
+
 	/*
 	 * Check whether there are replication slots requiring an older xmin.
 	 */
@@ -1610,6 +1620,7 @@ PgGetSnapshotData(Snapshot snapshot)
 	bool		suboverflowed = false;
 	volatile TransactionId replication_slot_xmin = InvalidTransactionId;
 	volatile TransactionId replication_slot_catalog_xmin = InvalidTransactionId;
+	volatile TransactionId global_snapshot_xmin = InvalidTransactionId;
 
 	Assert(snapshot != NULL);
 
@@ -1801,6 +1812,7 @@ PgGetSnapshotData(Snapshot snapshot)
 	/* fetch into volatile var while ProcArrayLock is held */
 	replication_slot_xmin = procArray->replication_slot_xmin;
 	replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+	global_snapshot_xmin = procArray->global_snapshot_xmin;
 
 	if (!TransactionIdIsValid(MyPgXact->xmin))
 		MyPgXact->xmin = TransactionXmin = xmin;
@@ -1821,6 +1833,10 @@ PgGetSnapshotData(Snapshot snapshot)
 	else
 		RecentGlobalXmin = globalxmin - vacuum_defer_cleanup_age;
 
+	if (TransactionIdIsValid(global_snapshot_xmin) &&
+		TransactionIdPrecedes(global_snapshot_xmin, RecentGlobalXmin))
+		RecentGlobalXmin = global_snapshot_xmin;
+
 	/* Check whether there's a replication slot requiring an older xmin. */
 	if (TransactionIdIsValid(replication_slot_xmin) &&
 		NormalTransactionIdPrecedes(replication_slot_xmin, RecentGlobalXmin))
@@ -3099,6 +3115,16 @@ ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
 	LWLockRelease(ProcArrayLock);
 }
 
+/*
+ * ProcArraySetGlobalSnapshotXmin
+ */
+void
+ProcArraySetGlobalSnapshotXmin(TransactionId xmin)
+{
+	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	procArray->global_snapshot_xmin = xmin;
+	LWLockRelease(ProcArrayLock);
+}
 
 #define XidCacheRemove(i) \
 	do { \

src/include/storage/procarray.h

@@ -127,5 +127,6 @@ extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
 
 extern void ResumePgXact(PGXACT *pgxact);
 extern void SuspendPgXact(PGXACT *pgxact);
+extern void ProcArraySetGlobalSnapshotXmin(TransactionId xmin);
 
 #endif							/* PROCARRAY_H */