A transaction that has not assigned an XID clears its advertised xmin at
transaction end without ProcArrayLock. ProcArrayInstallImportedXmin() can
therefore verify the source under a shared lock, then lose the source xmin
before it installs the importer xmin. A concurrent horizon scan can miss
both.
Track each PGPROC exported xmin with an atomic three-state protocol. An
importer marks the xmin referenced before installing it. If the importer
wins the handoff, the source clears its xmin under ProcArrayLock exclusive;
if transaction end wins, the import fails. Snapshots that were never
imported retain the existing lock-free cleanup path.
---
src/backend/access/transam/twophase.c | 1 +
src/backend/storage/ipc/procarray.c | 86 ++++++++++++++++++++++++---
src/backend/storage/lmgr/proc.c | 4 ++
src/backend/utils/time/snapmgr.c | 13 ++++
src/include/storage/proc.h | 21 +++++++
5 files changed, 117 insertions(+), 8 deletions(-)
diff --git a/src/backend/access/transam/twophase.c
b/src/backend/access/transam/twophase.c
index fa3bc50ec48..a64b1a41d0c 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -479,6 +479,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact,
FullTransactionId fxid,
proc->waitLock = NULL;
dlist_node_init(&proc->waitLink);
proc->waitProcLock = NULL;
+ pg_atomic_init_u32(&proc->xminExportState, PROC_XMIN_EXPORT_ENDING);
pg_atomic_init_u64(&proc->waitStart, 0);
for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
dlist_init(&proc->myProcLocks[i]);
diff --git a/src/backend/storage/ipc/procarray.c
b/src/backend/storage/ipc/procarray.c
index 66b394e2b4e..a2556314325 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -687,15 +687,39 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId
latestXid)
}
else
{
+ uint32 xmin_export_state;
+ bool clear_xmin_exclusively = false;
+
/*
- * If we have no XID, we don't need to lock, since we won't
affect
- * anyone else's calculation of a snapshot. We might change
their
- * estimate of global xmin, but that's OK.
+ * If we have no XID, we ordinarily don't need to lock, since
we won't
+ * affect anyone else's calculation of a snapshot. Atomically
close
+ * an exported xmin to new importers. If an importer acquired a
+ * reference first, clear under an exclusive lock to serialize
with
+ * both that importer and horizon calculations.
*/
Assert(!TransactionIdIsValid(proc->xid));
Assert(proc->subxidStatus.count == 0);
Assert(!proc->subxidStatus.overflowed);
+ xmin_export_state = pg_atomic_read_u32(&proc->xminExportState);
+ Assert(xmin_export_state == PROC_XMIN_EXPORT_ENDING ||
+ xmin_export_state == PROC_XMIN_EXPORT_EXPORTED ||
+ xmin_export_state == PROC_XMIN_EXPORT_REFERENCED);
+
+ if (xmin_export_state != PROC_XMIN_EXPORT_ENDING)
+ {
+ xmin_export_state =
+ pg_atomic_exchange_u32(&proc->xminExportState,
+
PROC_XMIN_EXPORT_ENDING);
+ Assert(xmin_export_state == PROC_XMIN_EXPORT_EXPORTED ||
+ xmin_export_state ==
PROC_XMIN_EXPORT_REFERENCED);
+ clear_xmin_exclusively =
+ (xmin_export_state ==
PROC_XMIN_EXPORT_REFERENCED);
+ }
+
+ if (clear_xmin_exclusively)
+ LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
proc->vxid.lxid = InvalidLocalTransactionId;
proc->xmin = InvalidTransactionId;
@@ -706,13 +730,20 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId
latestXid)
/* avoid unnecessarily dirtying shared cachelines */
if (proc->statusFlags & PROC_VACUUM_STATE_MASK)
{
- Assert(!LWLockHeldByMe(ProcArrayLock));
- LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+ if (!clear_xmin_exclusively)
+ {
+ Assert(!LWLockHeldByMe(ProcArrayLock));
+ LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+ }
Assert(proc->statusFlags ==
ProcGlobal->statusFlags[proc->pgxactoff]);
proc->statusFlags &= ~PROC_VACUUM_STATE_MASK;
ProcGlobal->statusFlags[proc->pgxactoff] =
proc->statusFlags;
- LWLockRelease(ProcArrayLock);
+ if (!clear_xmin_exclusively)
+ LWLockRelease(ProcArrayLock);
}
+
+ if (clear_xmin_exclusively)
+ LWLockRelease(ProcArrayLock);
}
}
@@ -738,6 +769,7 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId
latestXid)
proc->xid = InvalidTransactionId;
proc->vxid.lxid = InvalidLocalTransactionId;
proc->xmin = InvalidTransactionId;
+ pg_atomic_write_u32(&proc->xminExportState, PROC_XMIN_EXPORT_ENDING);
/* be sure this is cleared in abort */
proc->delayChkptFlags = 0;
@@ -2498,7 +2530,10 @@ ProcArrayInstallImportedXmin(TransactionId xmin,
if (!sourcevxid)
return false;
- /* Get lock so source xact can't end while we're doing this */
+ /*
+ * Stabilize the proc array while locating the source. A source without
+ * an XID can still begin ending until we acquire an xmin reference
below.
+ */
LWLockAcquire(ProcArrayLock, LW_SHARED);
/*
@@ -2512,6 +2547,7 @@ ProcArrayInstallImportedXmin(TransactionId xmin,
PGPROC *proc = &allProcs[pgprocno];
int statusFlags =
ProcGlobal->statusFlags[index];
TransactionId xid;
+ uint32 xmin_export_state;
/* Ignore procs running LAZY VACUUM */
if (statusFlags & PROC_IN_VACUUM)
@@ -2535,8 +2571,42 @@ ProcArrayInstallImportedXmin(TransactionId xmin,
INJECTION_POINT("install-imported-xmin-before-reference", NULL);
/*
- * Likewise, let's just make real sure its xmin does cover us.
+ * Acquire a reference to the exported xmin. This atomic
transition
+ * arbitrates with lock-free transaction end: either we change
+ * EXPORTED to REFERENCED first, forcing the source to clear
under
+ * ProcArrayLock exclusive, or the source changes it to ENDING
first
+ * and this import fails.
*/
+ xmin_export_state = pg_atomic_read_u32(&proc->xminExportState);
+ while (xmin_export_state == PROC_XMIN_EXPORT_EXPORTED)
+ {
+ if
(pg_atomic_compare_exchange_u32(&proc->xminExportState,
+
&xmin_export_state,
+
PROC_XMIN_EXPORT_REFERENCED))
+ {
+ xmin_export_state = PROC_XMIN_EXPORT_REFERENCED;
+ break;
+ }
+ }
+
+ if (xmin_export_state != PROC_XMIN_EXPORT_REFERENCED)
+ {
+ Assert(xmin_export_state == PROC_XMIN_EXPORT_ENDING);
+ continue;
+ }
+
+ /*
+ * The source could have ended and reused its PGPROC between the
+ * initial VXID match and the state transition above. Recheck
its
+ * identity after acquiring the reference. We intentionally
leave a
+ * conservatively acquired reference in place on failure.
+ */
+ if (proc->vxid.procNumber != sourcevxid->procNumber ||
+ proc->vxid.lxid != sourcevxid->localTransactionId ||
+ proc->databaseId != MyDatabaseId)
+ continue;
+
+ /* Lastly, make real sure its xmin does cover us. */
xid = UINT32_ACCESS_ONCE(proc->xmin);
if (!TransactionIdIsNormal(xid) ||
!TransactionIdPrecedesOrEquals(xid, xmin))
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 9d6e69175a5..0af8774d585 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -372,6 +372,8 @@ ProcGlobalShmemInit(void *arg)
*/
pg_atomic_init_u32(&(proc->procArrayGroupNext),
INVALID_PROC_NUMBER);
pg_atomic_init_u32(&(proc->clogGroupNext), INVALID_PROC_NUMBER);
+ pg_atomic_init_u32(&(proc->xminExportState),
+ PROC_XMIN_EXPORT_ENDING);
pg_atomic_init_u64(&(proc->waitStart), 0);
}
@@ -477,6 +479,7 @@ InitProcess(void)
MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
MyProc->xid = InvalidTransactionId;
MyProc->xmin = InvalidTransactionId;
+ pg_atomic_write_u32(&MyProc->xminExportState, PROC_XMIN_EXPORT_ENDING);
MyProc->pid = MyProcPid;
MyProc->vxid.procNumber = MyProcNumber;
MyProc->vxid.lxid = InvalidLocalTransactionId;
@@ -682,6 +685,7 @@ InitAuxiliaryProcess(void)
MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
MyProc->xid = InvalidTransactionId;
MyProc->xmin = InvalidTransactionId;
+ pg_atomic_write_u32(&MyProc->xminExportState, PROC_XMIN_EXPORT_ENDING);
MyProc->vxid.procNumber = INVALID_PROC_NUMBER;
MyProc->vxid.lxid = InvalidLocalTransactionId;
MyProc->databaseId = InvalidOid;
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index bc98a4361bf..3760dff7558 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -1122,6 +1122,7 @@ ExportSnapshot(Snapshot snapshot)
StringInfoData buf;
FILE *f;
MemoryContext oldcxt;
+ uint32 xmin_export_state;
char path[MAXPGPATH];
char pathtmp[MAXPGPATH];
@@ -1264,6 +1265,18 @@ ExportSnapshot(Snapshot snapshot)
(errcode_for_file_access(),
errmsg("could not write to file \"%s\": %m",
pathtmp)));
+ /*
+ * Make our xmin available for import before publishing the file.
Preserve
+ * REFERENCED if an earlier snapshot from this transaction was already
+ * imported.
+ */
+ xmin_export_state = PROC_XMIN_EXPORT_ENDING;
+ if (!pg_atomic_compare_exchange_u32(&MyProc->xminExportState,
+
&xmin_export_state,
+
PROC_XMIN_EXPORT_EXPORTED))
+ Assert(xmin_export_state == PROC_XMIN_EXPORT_EXPORTED ||
+ xmin_export_state == PROC_XMIN_EXPORT_REFERENCED);
+
/*
* Now that we have written everything into a .tmp file, rename the file
* to remove the .tmp suffix.
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 03a1a466fa8..ee8ee3862a4 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -81,6 +81,25 @@ struct XidCache
*/
#define PROC_XMIN_FLAGS (PROC_IN_VACUUM | PROC_IN_SAFE_IC)
+/*
+ * States for PGPROC.xminExportState:
+ *
+ * ENDING --ExportSnapshot()--> EXPORTED
+ * EXPORTED --snapshot import--> REFERENCED
+ * EXPORTED/REFERENCED --transaction end--> ENDING
+ *
+ * ENDING is also the quiescent state before the current transaction exports
+ * a snapshot. Transaction end may clear xmin lock-free when it changes
+ * EXPORTED to ENDING, but must use ProcArrayLock exclusive when it changes
+ * REFERENCED to ENDING.
+ */
+typedef enum ProcXminExportState
+{
+ PROC_XMIN_EXPORT_ENDING = 0,
+ PROC_XMIN_EXPORT_EXPORTED,
+ PROC_XMIN_EXPORT_REFERENCED
+} ProcXminExportState;
+
/*
* We allow a limited number of "weak" relation locks (AccessShareLock,
* RowShareLock, RowExclusiveLock) to be recorded in the PGPROC structure
@@ -250,6 +269,8 @@ typedef struct PGPROC
* vacuum must
not remove tuples deleted by
* xid >= xmin
! */
+ pg_atomic_uint32 xminExportState; /* one of ProcXminExportState */
+
XidCacheStatus subxidStatus; /* mirrored with
*
ProcGlobal->subxidStates[i] */
struct XidCache subxids; /* cache for subtransaction XIDs */
--
2.43.0
