diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 5cb28cf..a441ca0 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1561,6 +1561,7 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, elog(ERROR, "cache lookup failed for relation %u", OIDOldHeap); relform = (Form_pg_class) GETSTRUCT(reltup); + Assert(TransactionIdIsNormal(frozenXid)); relform->relfrozenxid = frozenXid; relform->relminmxid = cutoffMulti; diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 97e8962..e1e4681 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -372,11 +372,19 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid) (arrayP->numProcs - index - 1) * sizeof(int)); arrayP->pgprocnos[arrayP->numProcs - 1] = -1; /* for debugging */ arrayP->numProcs--; + + pg_atomic_write_u32(&ProcGlobal->snapshot_cached, 0); + + ProcGlobal->cached_snapshot_valid = false; LWLockRelease(ProcArrayLock); return; } } + pg_atomic_write_u32(&ProcGlobal->snapshot_cached, 0); + + ProcGlobal->cached_snapshot_valid = false; + /* Ooops */ LWLockRelease(ProcArrayLock); @@ -420,6 +428,8 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid) if (LWLockConditionalAcquire(ProcArrayLock, LW_EXCLUSIVE)) { ProcArrayEndTransactionInternal(proc, pgxact, latestXid); + pg_atomic_write_u32(&ProcGlobal->snapshot_cached, 0); + ProcGlobal->cached_snapshot_valid = false; LWLockRelease(ProcArrayLock); } else @@ -568,6 +578,9 @@ ProcArrayGroupClearXid(PGPROC *proc, TransactionId latestXid) nextidx = pg_atomic_read_u32(&proc->procArrayGroupNext); } + pg_atomic_write_u32(&ProcGlobal->snapshot_cached, 0); + ProcGlobal->cached_snapshot_valid = false; + /* We're done with the lock now. */ LWLockRelease(ProcArrayLock); @@ -1553,6 +1566,8 @@ GetSnapshotData(Snapshot snapshot) errmsg("out of memory"))); } + snapshot->takenDuringRecovery = RecoveryInProgress(); + /* * It is sufficient to get shared lock on ProcArrayLock, even if we are * going to set MyPgXact->xmin. @@ -1567,12 +1582,39 @@ GetSnapshotData(Snapshot snapshot) /* initialize xmin calculation with xmax */ globalxmin = xmin = xmax; - snapshot->takenDuringRecovery = RecoveryInProgress(); + if (!snapshot->takenDuringRecovery && ProcGlobal->cached_snapshot_valid) + { + Snapshot csnap = &ProcGlobal->cached_snapshot; + TransactionId *saved_xip; + TransactionId *saved_subxip; + + saved_xip = snapshot->xip; + saved_subxip = snapshot->subxip; - if (!snapshot->takenDuringRecovery) + memcpy(snapshot, csnap, sizeof(SnapshotData)); + + snapshot->xip = saved_xip; + snapshot->subxip = saved_subxip; + + memcpy(snapshot->xip, csnap->xip, + sizeof(TransactionId) * csnap->xcnt); + memcpy(snapshot->subxip, csnap->subxip, + sizeof(TransactionId) * csnap->subxcnt); + globalxmin = ProcGlobal->cached_snapshot_globalxmin; + xmin = csnap->xmin; + + count = csnap->xcnt; + subcount = csnap->subxcnt; + suboverflowed = csnap->suboverflowed; + + Assert(TransactionIdIsValid(globalxmin)); + Assert(TransactionIdIsValid(xmin)); + } + else if (!snapshot->takenDuringRecovery) { int *pgprocnos = arrayP->pgprocnos; int numProcs; + uint32 snapshot_cached= 0; /* * Spin over procArray checking xid, xmin, and subxids. The goal is @@ -1587,14 +1629,11 @@ GetSnapshotData(Snapshot snapshot) TransactionId xid; /* - * Backend is doing logical decoding which manages xmin - * separately, check below. + * Ignore procs running LAZY VACUUM (which don't need to retain + * rows) and backends doing logical decoding (which manages xmin + * separately, check below). */ - if (pgxact->vacuumFlags & PROC_IN_LOGICAL_DECODING) - continue; - - /* Ignore procs running LAZY VACUUM */ - if (pgxact->vacuumFlags & PROC_IN_VACUUM) + if (pgxact->vacuumFlags & (PROC_IN_LOGICAL_DECODING | PROC_IN_VACUUM)) continue; /* Update globalxmin to be the smallest valid xmin */ @@ -1663,6 +1702,79 @@ GetSnapshotData(Snapshot snapshot) } } } + + /* + * Let only one of the caller cache the computed snapshot, and others + * can continue as before. + */ + if (pg_atomic_compare_exchange_u32(&ProcGlobal->snapshot_cached, + &snapshot_cached, 1)) + { + Snapshot csnap = &ProcGlobal->cached_snapshot; + TransactionId *saved_xip; + TransactionId *saved_subxip; + int curr_subcount= subcount; + + ProcGlobal->cached_snapshot_globalxmin = globalxmin; + + saved_xip = csnap->xip; + saved_subxip = csnap->subxip; + memcpy(csnap, snapshot, sizeof(SnapshotData)); + csnap->xip = saved_xip; + csnap->subxip = saved_subxip; + + /* not yet stored. Yuck */ + csnap->xmin = xmin; + + memcpy(csnap->xip, snapshot->xip, + sizeof(TransactionId) * count); + memcpy(csnap->subxip, snapshot->subxip, + sizeof(TransactionId) * subcount); + /* + * If the transaction has no XID assigned, we can skip it; it + * won't have sub-XIDs either. If the XID is >= xmax, we can also + * skip it; such transactions will be treated as running anyway + * (and any sub-XIDs will also be >= xmax). + */ + if (TransactionIdIsNormal(MyPgXact->xid) && NormalTransactionIdPrecedes(MyPgXact->xid, xmax)) + { + + /* Add my own XID and sub-XIDs to snapshot. */ + csnap->xip[count] = MyPgXact->xid; + csnap->xcnt = count + 1; + + if (!suboverflowed) + { + if (MyPgXact->overflowed) + suboverflowed = true; + else + { + int nxids = MyPgXact->nxids; + + if (nxids > 0) + { + memcpy(csnap->subxip + subcount, + (void *) MyProc->subxids.xids, + nxids * sizeof(TransactionId)); + curr_subcount += nxids; + } + } + } + + } + else + csnap->xcnt = count; + + csnap->subxcnt = curr_subcount; + csnap->suboverflowed = suboverflowed; + + /* + * The memory barrier has be to be placed here to ensure that + * cached_snapshot_valid is set only after snapshot is cached. + */ + pg_write_barrier(); + ProcGlobal->cached_snapshot_valid = true; + } } else { diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 6453b88..b8d0297 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -51,7 +51,7 @@ #include "storage/spin.h" #include "utils/timeout.h" #include "utils/timestamp.h" - +#include "port/atomics.h" /* GUC variables */ int DeadlockTimeout = 1000; @@ -114,6 +114,13 @@ ProcGlobalShmemSize(void) size = add_size(size, mul_size(NUM_AUXILIARY_PROCS, sizeof(PGXACT))); size = add_size(size, mul_size(max_prepared_xacts, sizeof(PGXACT))); + /* for the cached snapshot */ +#define PROCARRAY_MAXPROCS (MaxBackends + max_prepared_xacts) + size = add_size(size, mul_size(sizeof(TransactionId), PROCARRAY_MAXPROCS)); +#define TOTAL_MAX_CACHED_SUBXIDS \ + ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS) + size = add_size(size, mul_size(sizeof(TransactionId), TOTAL_MAX_CACHED_SUBXIDS)); + return size; } @@ -278,6 +285,13 @@ InitProcGlobal(void) /* Create ProcStructLock spinlock, too */ ProcStructLock = (slock_t *) ShmemAlloc(sizeof(slock_t)); SpinLockInit(ProcStructLock); + + /* cached snapshot */ + ProcGlobal->cached_snapshot_valid = false; + pg_atomic_write_u32(&ProcGlobal->snapshot_cached, 0); + ProcGlobal->cached_snapshot.xip = ShmemAlloc(PROCARRAY_MAXPROCS * sizeof(TransactionId)); + ProcGlobal->cached_snapshot.subxip = ShmemAlloc(TOTAL_MAX_CACHED_SUBXIDS * sizeof(TransactionId)); + ProcGlobal->cached_snapshot_globalxmin = InvalidTransactionId; } /* diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index dbcdd3f..73a424e 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -16,6 +16,7 @@ #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" @@ -234,6 +235,12 @@ typedef struct PROC_HDR int startupProcPid; /* Buffer id of the buffer that Startup process waits for pin on, or -1 */ int startupBufferPinWaitBufId; + + /* Cached snapshot */ + volatile bool cached_snapshot_valid; + pg_atomic_uint32 snapshot_cached; + SnapshotData cached_snapshot; + TransactionId cached_snapshot_globalxmin; } PROC_HDR; extern PROC_HDR *ProcGlobal;