From b4d5c70fd0db9964088f8d8766bda07a51054055 Mon Sep 17 00:00:00 2001 From: Shinya Kato Date: Wed, 10 Jun 2026 09:30:46 +0900 Subject: [PATCH v6 1/2] Add infrastructure to identify what holds back the xid horizon Introduce GetXidHorizonBlocker() and GetXidHorizonBlockers() in procarray.c. Given an xid horizon, these search the proc array and the replication slots for what is preventing it from advancing: active transactions, idle-in-transaction sessions, prepared transactions, hot standby feedback (held either in a walsender's PGPROC or, when a physical slot is in use, in the slot itself), and logical replication slots. A blocker is described by the new XidHorizonBlocker struct, and the candidates are ranked by XidHorizonBlockerType so that the single highest-priority (root-cause) blocker can be reported. A companion GetPreparedTransactionGid() in twophase.c resolves the GID of a blocking prepared transaction. This is infrastructure with no in-tree caller yet. A following patch uses it to report the blocker in VACUUM's log output, and the same API can underpin a future SQL-callable view that lists every blocker. Author: Shinya Kato Reviewed-by: wenhui qiu Reviewed-by: Fujii Masao Reviewed-by: Sami Imseih Reviewed-by: Dilip Kumar Reviewed-by: Japin Li Discussion: https://postgr.es/m/CAOzEurSgy-gDtwFmEbj5+R9PL0_G3qYB6nnzJtNStyuf87VSVg@mail.gmail.com --- src/backend/access/transam/twophase.c | 39 +++ src/backend/storage/ipc/procarray.c | 336 ++++++++++++++++++++++++++ src/include/access/twophase.h | 1 + src/include/storage/procarray.h | 46 ++++ src/tools/pgindent/typedefs.list | 3 + 5 files changed, 425 insertions(+) diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 1035e8b3fc7..69d1d82b342 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2822,6 +2822,45 @@ LookupGXactBySubid(Oid subid) return found; } +/* + * GetPreparedTransactionGid + * Get the GID for the prepared transaction with the given XID. + * + * Returns true when a matching prepared transaction is found. gid will be + * set to an empty string when no match is found. + */ +bool +GetPreparedTransactionGid(TransactionId xid, char gid[GIDSIZE]) +{ + bool found = false; + + Assert(TransactionIdIsValid(xid)); + + gid[0] = '\0'; + + if (max_prepared_xacts == 0 || TwoPhaseState == NULL) + return false; + + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (int i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + if (!gxact->valid) + continue; + + if (!TransactionIdEquals(XidFromFullTransactionId(gxact->fxid), xid)) + continue; + + strlcpy(gid, gxact->gid, GIDSIZE); + found = true; + break; + } + LWLockRelease(TwoPhaseStateLock); + + return found; +} + /* * TwoPhaseGetOldestXidInCommit * Return the oldest transaction ID from prepared transactions that are diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index f540bb6b23f..2c1f7410c05 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -58,11 +58,13 @@ #include "pgstat.h" #include "postmaster/bgworker.h" #include "port/pg_lfind.h" +#include "replication/slot.h" #include "storage/proc.h" #include "storage/procarray.h" #include "storage/procsignal.h" #include "storage/subsystems.h" #include "utils/acl.h" +#include "utils/backend_status.h" #include "utils/builtins.h" #include "utils/injection_point.h" #include "utils/lsyscache.h" @@ -282,6 +284,28 @@ typedef enum KAXCompressReason KAX_STARTUP_PROCESS_IDLE, /* startup process is about to sleep */ } KAXCompressReason; +/* + * A candidate blocker collected during the ProcArray/replication-slot scan. + * + * This is deliberately small. Like the other ProcArray scanners we size the + * result array for the worst case (one entry per PGPROC plus one per + * replication slot), so a fat per-entry struct would waste a lot of memory at + * high max_connections. The human-readable name of a blocker (prepared-xact + * GID, standby application_name or slot name) is comparatively large but only + * needed for the single blocker that is actually reported, so it is resolved + * later by FillXidHorizonBlocker() rather than stored for every candidate. + */ +typedef struct XidHorizonBlockerCandidate +{ + XidHorizonBlockerType type; + TransactionId xid; /* the blocking xid/xmin */ + int pid; /* backend pid (0 for prepared xacts and slots) */ + ProcNumber proc_number; /* walsender proc number for hot standby + * feedback; INVALID_PROC_NUMBER otherwise */ + int slot_index; /* replication_slots[] index for slot blockers; + * -1 otherwise */ +} XidHorizonBlockerCandidate; + static PGPROC *allProcs; /* @@ -1999,6 +2023,318 @@ GetReplicationHorizons(TransactionId *xmin, TransactionId *catalog_xmin) *catalog_xmin = horizons.slot_catalog_xmin; } +/* + * Return XidHorizonBlockerType for a backend whose xid matches the horizon. + */ +static inline XidHorizonBlockerType +XidHorizonBlockerTypeForBackend(const PGPROC *proc) +{ + if (proc->wait_event_info == WAIT_EVENT_CLIENT_READ) + return XHB_IDLE_IN_TRANSACTION; + return XHB_ACTIVE_TRANSACTION; +} + +/* + * Return XidHorizonBlockerType for a backend whose xmin matches the horizon. + */ +static inline XidHorizonBlockerType +XidHorizonBlockerTypeForXminBackend(const PGPROC *proc) +{ + if (proc->wait_event_info == WAIT_EVENT_CLIENT_READ) + return XHB_XMIN_IDLE_IN_TRANSACTION; + return XHB_XMIN_ACTIVE_TRANSACTION; +} + +/* + * Find the blockers that are holding back the given xid horizon. + * + * This function searches for what is preventing the given horizon from being + * advanced to allow removal of dead tuples. It checks: + * 1. Active transactions (running statements) + * 2. Idle-in-transaction sessions + * 3. Prepared transactions + * 4. Hot standby feedback + * 5. Replication slots (physical or logical) + * + * Hot standby feedback deserves a note, because where the standby's xmin is + * stored depends on whether the connection uses a replication slot (see + * ProcessStandbyHSFeedbackMessage): + * + * - Without a slot, the xmin is held in the walsender's PGPROC and is found by + * the ProcArray scan below as XHB_HOT_STANDBY_FEEDBACK. + * - With a physical slot, the xmin is held in the slot (the walsender's PGPROC + * xmin is reset to invalid), so the ProcArray scan does not see it. The slot + * scan finds it instead: if a standby is currently connected (the slot is + * active) it is still reported as XHB_HOT_STANDBY_FEEDBACK, otherwise the + * persisted reservation is reported as XHB_PHYSICAL_REPLICATION_SLOT. + * + * Logical slots reserve catalog_xmin and are reported as + * XHB_LOGICAL_REPLICATION_SLOT. + * + * Because the horizon was computed earlier, the original blocker may have + * already committed by the time this function runs. The result is therefore + * best-effort: it may return a different blocker, or no blocker at all. + * + * Returns a palloc'd array of candidate blockers and stores the number of + * entries in *nblockers. The blocker names are not resolved here; the caller + * does that for the blocker it reports (see FillXidHorizonBlocker). The array + * may be empty if no blocker is found. + */ +static XidHorizonBlockerCandidate * +GetXidHorizonBlockers(TransactionId horizon, int *nblockers) +{ + ProcArrayStruct *arrayP = procArray; + TransactionId *other_xids = ProcGlobal->xids; + XidHorizonBlockerCandidate *result; + int count = 0; + int max_blockers; + + Assert(TransactionIdIsValid(horizon)); + Assert(nblockers != NULL); + + /* + * Size the result array for the worst case (one entry per PGPROC plus one + * per replication slot) and allocate it before acquiring ProcArrayLock, so + * the scan below never has to allocate while holding the lock. The other + * ProcArray scanners such as GetCurrentVirtualXIDs() size their result + * space the same way. Only 0-2 entries are returned in practice, and each + * entry is small because the blocker name is resolved later, not here. + */ + max_blockers = arrayP->maxProcs + max_replication_slots; + result = palloc_array(XidHorizonBlockerCandidate, max_blockers); + + LWLockAcquire(ProcArrayLock, LW_SHARED); + + for (int index = 0; index < arrayP->numProcs; index++) + { + int pgprocno = arrayP->pgprocnos[index]; + PGPROC *proc = &allProcs[pgprocno]; + int8 statusFlags = ProcGlobal->statusFlags[index]; + TransactionId proc_xid; + TransactionId proc_xmin; + XidHorizonBlockerCandidate *dst = NULL; + + /* + * Skip over backends either vacuuming (which is ok with rows being + * removed, as long as pg_subtrans is not truncated), doing logical + * decoding (which manages xmin separately, check below), or myself. + */ + if (statusFlags & (PROC_IN_VACUUM | PROC_IN_LOGICAL_DECODING) || + proc == MyProc) + continue; + + /* Fetch xid just once - see GetNewTransactionId */ + proc_xid = UINT32_ACCESS_ONCE(other_xids[index]); + proc_xmin = UINT32_ACCESS_ONCE(proc->xmin); + + /* + * Candidates are collected in ProcArray order; callers can reorder if + * needed. Only the blocker type differs between the cases below; the + * common fields are filled in once afterwards. + */ + if (TransactionIdEquals(proc_xid, horizon)) + { + /* This proc's xid matches the horizon (the root cause) */ + dst = &result[count++]; + if (proc->pid == 0) + dst->type = XHB_PREPARED_TRANSACTION; + else + dst->type = XidHorizonBlockerTypeForBackend(proc); + } + else if (TransactionIdEquals(proc_xmin, horizon)) + { + /* This proc's xmin matches the horizon (held back by the above) */ + dst = &result[count++]; + if (statusFlags & PROC_AFFECTS_ALL_HORIZONS) + dst->type = XHB_HOT_STANDBY_FEEDBACK; + else + dst->type = XidHorizonBlockerTypeForXminBackend(proc); + } + + if (dst) + { + dst->pid = proc->pid; + dst->xid = horizon; + dst->proc_number = pgprocno; + dst->slot_index = -1; + } + } + + LWLockRelease(ProcArrayLock); + + /* + * Also check replication slots. + * + * A physical slot reserves xmin on behalf of a standby using hot standby + * feedback. If a standby is currently connected we attribute the + * reservation to that feedback (and record the walsender pid so its + * application_name can be looked up below); otherwise it is a persisted + * physical-slot reservation with no connected standby. A logical slot + * reserves catalog_xmin for logical decoding. + */ + if (max_replication_slots > 0) + { + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + + for (int i = 0; i < max_replication_slots; i++) + { + ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i]; + TransactionId slot_xmin; + TransactionId slot_catalog_xmin; + ProcNumber active_proc; + XidHorizonBlockerCandidate *dst; + + if (!s->in_use) + continue; + + SpinLockAcquire(&s->mutex); + slot_xmin = s->data.xmin; + slot_catalog_xmin = s->data.catalog_xmin; + active_proc = s->active_proc; + SpinLockRelease(&s->mutex); + + if (!TransactionIdEquals(slot_xmin, horizon) && + !TransactionIdEquals(slot_catalog_xmin, horizon)) + continue; + + dst = &result[count++]; + dst->xid = TransactionIdIsValid(slot_xmin) ? + slot_xmin : + slot_catalog_xmin; + dst->slot_index = i; + + if (SlotIsPhysical(s) && active_proc != INVALID_PROC_NUMBER) + { + /* Connected standby: report as hot standby feedback. */ + dst->type = XHB_HOT_STANDBY_FEEDBACK; + dst->pid = GetPGProcByNumber(active_proc)->pid; + dst->proc_number = active_proc; + } + else + { + /* + * A physical slot with no connected standby, or a logical + * slot. The slot name is resolved later from slot_index. + */ + dst->type = SlotIsPhysical(s) ? + XHB_PHYSICAL_REPLICATION_SLOT : + XHB_LOGICAL_REPLICATION_SLOT; + dst->pid = 0; + dst->proc_number = INVALID_PROC_NUMBER; + } + } + + LWLockRelease(ReplicationSlotControlLock); + } + + *nblockers = count; + return result; +} + +/* + * Resolve a scanned candidate into a fully-populated blocker. + * + * The scan deliberately leaves the (comparatively large) blocker name out of + * every candidate; it is filled in here for the single blocker that is + * actually reported, after all scan locks have been released: + * + * - prepared transaction: look up the GID from the xid; + * - hot standby feedback: look up the standby's application_name from the + * walsender's backend status entry. Matching on proc_number rather than pid + * avoids being fooled by pid reuse, and the lookup is a bsearch over a + * process-local snapshot; the st_procpid check guards against the proc being + * reused since the scan; + * - replication slot: copy the slot name (best-effort: empty if the slot has + * since been dropped). + */ +static void +FillXidHorizonBlocker(const XidHorizonBlockerCandidate *cand, + XidHorizonBlocker *blocker) +{ + blocker->type = cand->type; + blocker->xid = cand->xid; + blocker->pid = cand->pid; + blocker->proc_number = cand->proc_number; + blocker->name[0] = '\0'; + + switch (cand->type) + { + case XHB_PREPARED_TRANSACTION: + GetPreparedTransactionGid(cand->xid, blocker->name); + break; + + case XHB_HOT_STANDBY_FEEDBACK: + { + PgBackendStatus *beentry; + + beentry = pgstat_get_beentry_by_proc_number(cand->proc_number); + if (beentry != NULL && beentry->st_procpid == cand->pid && + beentry->st_appname != NULL && beentry->st_appname[0] != '\0') + strlcpy(blocker->name, beentry->st_appname, + sizeof(blocker->name)); + break; + } + + case XHB_PHYSICAL_REPLICATION_SLOT: + case XHB_LOGICAL_REPLICATION_SLOT: + { + NameData slotname; + + if (ReplicationSlotName(cand->slot_index, &slotname)) + strlcpy(blocker->name, NameStr(slotname), + sizeof(blocker->name)); + break; + } + + default: + break; + } +} + +/* + * Get the highest-priority blocker holding back the xid horizon. + * + * Returns true and stores the blocker in *blocker if any are found. + */ +bool +GetXidHorizonBlocker(TransactionId horizon, XidHorizonBlocker *blocker) +{ + XidHorizonBlockerCandidate *blockers; + XidHorizonBlockerCandidate *best = NULL; + int nblockers; + + Assert(TransactionIdIsValid(horizon)); + Assert(blocker != NULL); + + blockers = GetXidHorizonBlockers(horizon, &nblockers); + for (int i = 0; i < nblockers; i++) + { + if (best == NULL || blockers[i].type < best->type) + { + best = &blockers[i]; + + /* + * xid-match types are the highest priority (the root cause holding + * the horizon), so nothing can outrank them; stop once we find one. + */ + if (best->type <= XHB_PREPARED_TRANSACTION) + break; + } + } + + /* + * Resolve the name of the selected blocker only. The expensive GID / + * application_name / slot-name lookups thus run once for the reported + * blocker instead of for every candidate collected above. + */ + if (best != NULL) + FillXidHorizonBlocker(best, blocker); + + pfree(blockers); + + return (best != NULL); +} + /* * GetMaxSnapshotXidCount -- get max size for snapshot XID array * diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 1d2ff42c9b7..fc7294a4e25 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -70,6 +70,7 @@ extern void TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid_res, int szgid); extern bool LookupGXactBySubid(Oid subid); +extern bool GetPreparedTransactionGid(TransactionId xid, char gid[GIDSIZE]); extern TransactionId TwoPhaseGetOldestXidInCommit(void); #endif /* TWOPHASE_H */ diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index d718a5b542f..ece8a12b7af 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -14,10 +14,53 @@ #ifndef PROCARRAY_H #define PROCARRAY_H +#include "access/xact.h" +#include "storage/lock.h" +#include "storage/procnumber.h" #include "storage/standby.h" #include "utils/relcache.h" #include "utils/snapshot.h" +/* + * Type of blocker that is holding back the xid horizon. + * Listed in priority order from highest to lowest. Blockers whose xid + * matches the horizon (the root cause) are listed before blockers whose + * xmin matches (held back by the root cause). Within each group, active + * transactions are listed first because they are the most actionable for + * the DBA (the running query can be identified and cancelled). + */ +typedef enum XidHorizonBlockerType +{ + XHB_NONE = 0, + /* xid-match types (horizon == proc's xid) */ + XHB_ACTIVE_TRANSACTION, /* backend running a statement */ + XHB_IDLE_IN_TRANSACTION, /* backend idle in transaction */ + XHB_PREPARED_TRANSACTION, /* prepared (two-phase) transaction */ + /* xmin-match types (horizon == proc's xmin or slot's xmin) */ + XHB_XMIN_ACTIVE_TRANSACTION, /* backend running a statement */ + XHB_XMIN_IDLE_IN_TRANSACTION, /* backend idle in transaction */ + XHB_HOT_STANDBY_FEEDBACK, /* connected standby with hot_standby_feedback */ + XHB_PHYSICAL_REPLICATION_SLOT, /* physical slot reserving xmin (no + * connected standby) */ + XHB_LOGICAL_REPLICATION_SLOT, /* logical replication slot */ +} XidHorizonBlockerType; + +/* + * Information about a blocker that is holding back the xid horizon. + */ +typedef struct XidHorizonBlocker +{ + XidHorizonBlockerType type; + TransactionId xid; /* the blocking xid/xmin */ + int pid; /* backend pid (0 for prepared xacts and + * slots) */ + ProcNumber proc_number; /* backend's proc number, used to look up its + * application_name; INVALID_PROC_NUMBER when + * there is no associated backend */ + /* large enough for prepared-txn GID or replication slot name */ + char name[Max(GIDSIZE, NAMEDATALEN)]; +} XidHorizonBlocker; + extern void ProcArrayAdd(PGPROC *proc); extern void ProcArrayRemove(PGPROC *proc, TransactionId latestXid); @@ -98,4 +141,7 @@ extern void ProcArraySetReplicationSlotXmin(TransactionId xmin, extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin, TransactionId *catalog_xmin); +extern bool GetXidHorizonBlocker(TransactionId horizon, + XidHorizonBlocker *blocker); + #endif /* PROCARRAY_H */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 8cf40c87043..356e730efb0 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3544,6 +3544,9 @@ XactLockTableWaitInfo XidBoundsViolation XidCacheStatus XidCommitStatus +XidHorizonBlocker +XidHorizonBlockerCandidate +XidHorizonBlockerType XidStatus XmlExpr XmlExprOp -- 2.47.3