70 #define UINT32_ACCESS_ONCE(var) ((uint32)(*((volatile uint32 *)&(var))))
312 #ifdef XIDCACHE_DEBUG
315 static long xc_by_recent_xmin = 0;
316 static long xc_by_known_xact = 0;
317 static long xc_by_my_xact = 0;
318 static long xc_by_latest_xid = 0;
319 static long xc_by_main_xid = 0;
320 static long xc_by_child_xid = 0;
321 static long xc_by_known_assigned = 0;
322 static long xc_no_overflow = 0;
323 static long xc_slow_answer = 0;
325 #define xc_by_recent_xmin_inc() (xc_by_recent_xmin++)
326 #define xc_by_known_xact_inc() (xc_by_known_xact++)
327 #define xc_by_my_xact_inc() (xc_by_my_xact++)
328 #define xc_by_latest_xid_inc() (xc_by_latest_xid++)
329 #define xc_by_main_xid_inc() (xc_by_main_xid++)
330 #define xc_by_child_xid_inc() (xc_by_child_xid++)
331 #define xc_by_known_assigned_inc() (xc_by_known_assigned++)
332 #define xc_no_overflow_inc() (xc_no_overflow++)
333 #define xc_slow_answer_inc() (xc_slow_answer++)
335 static void DisplayXidCache(
void);
338 #define xc_by_recent_xmin_inc() ((void) 0)
339 #define xc_by_known_xact_inc() ((void) 0)
340 #define xc_by_my_xact_inc() ((void) 0)
341 #define xc_by_latest_xid_inc() ((void) 0)
342 #define xc_by_main_xid_inc() ((void) 0)
343 #define xc_by_child_xid_inc() ((void) 0)
344 #define xc_by_known_assigned_inc() ((void) 0)
345 #define xc_no_overflow_inc() ((void) 0)
346 #define xc_slow_answer_inc() ((void) 0)
352 bool exclusive_lock);
387 #define PROCARRAY_MAXPROCS (MaxBackends + max_prepared_xacts)
405 #define TOTAL_MAX_CACHED_SUBXIDS \
406 ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS)
493 (
errcode(ERRCODE_TOO_MANY_CONNECTIONS),
494 errmsg(
"sorry, too many clients already")));
577 #ifdef XIDCACHE_DEBUG
589 Assert(myoff >= 0 && myoff < arrayP->numProcs);
619 movecount = arrayP->
numProcs - myoff - 1;
846 while (extraWaits-- > 0)
1119 "recovery snapshots are now enabled");
1123 "recovery snapshot waiting for non-overflowed snapshot or "
1124 "until oldest active xid on standby is at least %u (now %u)",
1178 xids[nxids++] = xid;
1186 elog(
ERROR,
"KnownAssignedXids is not empty");
1205 for (
i = 0;
i < nxids;
i++)
1210 "found duplicated transaction %u for KnownAssignedXids insertion",
1294 "recovery snapshot waiting for non-overflowed snapshot or "
1295 "until oldest active xid on standby is at least %u (now %u)",
1336 for (
i = 0;
i < nsubxids;
i++)
1451 (
errcode(ERRCODE_OUT_OF_MEMORY),
1452 errmsg(
"out of memory")));
1464 latestCompletedXid =
1476 for (
int pgxactoff = 0; pgxactoff < numProcs; pgxactoff++)
1484 if (pgxactoff == mypgxactoff)
1513 pxids = other_subxidstates[pgxactoff].
count;
1515 pgprocno = arrayP->
pgprocnos[pgxactoff];
1517 for (
j = pxids - 1;
j >= 0;
j--)
1537 if (other_subxidstates[pgxactoff].overflowed)
1538 xids[nxids++] = pxid;
1623 bool result =
false;
2007 rel->
rd_rel->relkind == RELKIND_RELATION ||
2008 rel->
rd_rel->relkind == RELKIND_MATVIEW ||
2009 rel->
rd_rel->relkind == RELKIND_TOASTVALUE);
2154 uint64 curXactCompletionCount;
2194 snapshot->
copied =
false;
2243 bool suboverflowed =
false;
2248 uint64 curXactCompletionCount;
2253 Assert(snapshot != NULL);
2266 if (snapshot->
xip == NULL)
2274 if (snapshot->
xip == NULL)
2276 (
errcode(ERRCODE_OUT_OF_MEMORY),
2277 errmsg(
"out of memory")));
2281 if (snapshot->
subxip == NULL)
2283 (
errcode(ERRCODE_OUT_OF_MEMORY),
2284 errmsg(
"out of memory")));
2301 myxid = other_xids[mypgxactoff];
2333 for (
int pgxactoff = 0; pgxactoff < numProcs; pgxactoff++)
2353 if (pgxactoff == mypgxactoff)
2376 statusFlags = allStatusFlags[pgxactoff];
2404 if (subxidStates[pgxactoff].overflowed)
2405 suboverflowed =
true;
2408 int nsubxids = subxidStates[pgxactoff].
count;
2412 int pgprocno = pgprocnos[pgxactoff];
2417 memcpy(snapshot->
subxip + subcount,
2420 subcount += nsubxids;
2461 suboverflowed =
true;
2494 def_vis_xid_data = xmin;
2507 def_vis_xid = def_vis_xid_data;
2567 snapshot->
xmin = xmin;
2568 snapshot->
xmax = xmax;
2569 snapshot->
xcnt = count;
2582 snapshot->
copied =
false;
2603 bool result =
false;
2681 bool result =
false;
2779 if (CurrentRunningXacts->
xids == NULL)
2786 if (CurrentRunningXacts->
xids == NULL)
2788 (
errcode(ERRCODE_OUT_OF_MEMORY),
2789 errmsg(
"out of memory")));
2792 xids = CurrentRunningXacts->
xids;
2794 count = subcount = 0;
2795 suboverflowed =
false;
2804 latestCompletedXid =
2832 oldestRunningXid = xid;
2835 suboverflowed =
true;
2845 xids[count++] = xid;
2875 subcount += nsubxids;
2895 CurrentRunningXacts->
xcnt = count - subcount;
2896 CurrentRunningXacts->
subxcnt = subcount;
2908 return CurrentRunningXacts;
2962 oldestRunningXid = xid;
2972 return oldestRunningXid;
3042 if (!recovery_in_progress)
3060 oldestSafeXid = xid;
3066 return oldestSafeXid;
3116 vxids[count++] = vxid;
3138 bool result =
false;
3159 for (
i = 0;
i < nvxids;
i++)
3221 if (proc->
pid == pid)
3262 if (other_xids[
index] == xid)
3314 bool allDbs,
int excludeVacuum,
3337 if (excludeVacuum & statusFlags)
3359 vxids[count++] = vxid;
3424 (
errcode(ERRCODE_OUT_OF_MEMORY),
3425 errmsg(
"out of memory")));
3460 vxids[count++] = vxid;
3487 bool conflictPending)
3581 return count >= min;
3706 if (proc->
roleId == roleid)
3743 #define MAXAUTOVACPIDS 10
3748 for (tries = 0; tries < 50; tries++)
3756 *nbackends = *nprepared = 0;
3780 autovac_pids[nautovacs++] = proc->
pid;
3796 (
void)
kill(autovac_pids[
index], SIGTERM);
3846 (
errcode(ERRCODE_OBJECT_IN_USE),
3847 errmsg(
"database \"%s\" is being used by prepared transactions",
3850 "There are %d prepared transactions using the database.",
3878 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
3879 errmsg(
"must be a superuser to terminate superuser process")));
3885 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
3886 errmsg(
"permission denied to terminate process"),
3887 errdetail(
"Only roles with privileges of the role whose process is being terminated or with privileges of the \"%s\" role may terminate this process.",
3888 "pg_signal_backend")));
3910 (void)
kill(-pid, SIGTERM);
3912 (void)
kill(pid, SIGTERM);
3928 bool already_locked)
3932 if (!already_locked)
3938 if (!already_locked)
3941 elog(
DEBUG1,
"xmin required by slots: data %u, catalog %u",
3942 xmin, catalog_xmin);
3960 if (catalog_xmin != NULL)
4006 for (
i = nxids - 1;
i >= 0;
i--)
4016 mysubxidstat->
count--;
4029 if (j < 0 && !MyProc->subxidStatus.overflowed)
4030 elog(
WARNING,
"did not find subXID %u in MyProc", anxid);
4039 mysubxidstat->
count--;
4045 if (j < 0 && !MyProc->subxidStatus.overflowed)
4046 elog(
WARNING,
"did not find subXID %u in MyProc", xid);
4057 #ifdef XIDCACHE_DEBUG
4063 DisplayXidCache(
void)
4066 "XidCache: xmin: %ld, known: %ld, myxact: %ld, latest: %ld, mainxid: %ld, childxid: %ld, knownassigned: %ld, nooflo: %ld, slow: %ld\n",
4073 xc_by_known_assigned,
4142 state->definitely_needed))
4281 return state->maybe_needed;
4345 if (retreat_by == 0)
4384 + (
int32) (xid - rel_xid));
4488 Assert(next_expected_xid == xid);
4706 static unsigned int transactionEndsCounter;
4710 #define KAX_COMPRESS_FREQUENCY 128
4711 #define KAX_COMPRESS_IDLE_INTERVAL 1000
4719 nelements = head - tail;
4750 if (nelements < 2 * pArray->numKnownAssignedXids)
4760 if (lastCompressTs != 0)
4780 for (
i = tail;
i < head;
i++)
4814 bool exclusive_lock)
4830 if (to_xid >= from_xid)
4831 nxids = to_xid - from_xid + 1;
4835 next_xid = from_xid;
4850 Assert(head >= 0 && head <= pArray->maxKnownAssignedXids);
4851 Assert(tail >= 0 && tail < pArray->maxKnownAssignedXids);
4862 elog(
ERROR,
"out-of-order XID insertion in KnownAssignedXids");
4879 elog(
ERROR,
"too many KnownAssignedXids");
4883 next_xid = from_xid;
4884 for (
i = 0;
i < nxids;
i++)
4931 int result_index = -1;
4954 while (first <= last)
4959 mid_index = (first + last) / 2;
4964 result_index = mid_index;
4968 last = mid_index - 1;
4970 first = mid_index + 1;
4973 if (result_index < 0)
4990 if (result_index == tail)
5064 for (
i = 0;
i < nsubxids;
i++)
5103 for (
i = tail;
i < head;
i++)
5126 for (
i = tail;
i < head;
i++)
5192 for (
i = tail;
i < head;
i++)
5216 xarray[count++] = knownXid;
5242 for (
i = tail;
i < head;
i++)
5277 for (
i = tail;
i < head;
i++)
5286 elog(trace_level,
"%d KnownAssignedXids (num=%d tail=%d head=%d) %s",
bool has_privs_of_role(Oid member, Oid role)
static bool pg_atomic_compare_exchange_u32(volatile pg_atomic_uint32 *ptr, uint32 *expected, uint32 newval)
#define pg_read_barrier()
#define pg_write_barrier()
static void pg_atomic_write_u32(volatile pg_atomic_uint32 *ptr, uint32 val)
static uint32 pg_atomic_read_u32(volatile pg_atomic_uint32 *ptr)
static uint32 pg_atomic_exchange_u32(volatile pg_atomic_uint32 *ptr, uint32 newval)
TimestampTz GetCurrentTimestamp(void)
#define FLEXIBLE_ARRAY_MEMBER
#define PG_USED_FOR_ASSERTS_ONLY
#define OidIsValid(objectId)
bool IsCatalogRelation(Relation relation)
char * get_database_name(Oid dbid)
elog(ERROR, "%s: %s", p2, msg)
int errdetail(const char *fmt,...)
int errdetail_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
int trace_recovery(int trace_level)
#define ereport(elevel,...)
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend_int(List *list, int datum)
#define VirtualTransactionIdIsValid(vxid)
#define InvalidLocalTransactionId
#define VirtualTransactionIdEquals(vxid1, vxid2)
#define GET_VXID_FROM_PGPROC(vxid, proc)
bool LWLockHeldByMe(LWLock *lock)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
void LWLockRelease(LWLock *lock)
bool LWLockConditionalAcquire(LWLock *lock, LWLockMode mode)
void pfree(void *pointer)
#define AmStartupProcess()
#define IsBootstrapProcessingMode()
#define CHECK_FOR_INTERRUPTS()
static bool pg_lfind32(uint32 key, uint32 *base, uint32 nelem)
#define qsort(a, b, c, d)
void PGSemaphoreUnlock(PGSemaphore sema)
void PGSemaphoreLock(PGSemaphore sema)
#define PROC_IN_LOGICAL_DECODING
#define NUM_AUXILIARY_PROCS
#define PROC_AFFECTS_ALL_HORIZONS
#define PROC_VACUUM_STATE_MASK
#define PROC_IS_AUTOVACUUM
@ KAX_STARTUP_PROCESS_IDLE
static GlobalVisState GlobalVisDataRels
bool GlobalVisTestIsRemovableFullXid(GlobalVisState *state, FullTransactionId fxid)
TransactionId GetOldestNonRemovableTransactionId(Relation rel)
static void GetSnapshotDataInitOldSnapshot(Snapshot snapshot)
VirtualTransactionId * GetVirtualXIDsDelayingChkpt(int *nvxids, int type)
#define TOTAL_MAX_CACHED_SUBXIDS
static GlobalVisState GlobalVisSharedRels
void ProcArrayGetReplicationSlotXmin(TransactionId *xmin, TransactionId *catalog_xmin)
static GlobalVisState GlobalVisCatalogRels
static void TransactionIdRetreatSafely(TransactionId *xid, int retreat_by, FullTransactionId rel)
bool GlobalVisTestIsRemovableXid(GlobalVisState *state, TransactionId xid)
bool GlobalVisCheckRemovableFullXid(Relation rel, FullTransactionId fxid)
static void KnownAssignedXidsCompress(KAXCompressReason reason, bool haveLock)
pid_t SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode, bool conflictPending)
Size ProcArrayShmemSize(void)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
void XidCacheRemoveRunningXids(TransactionId xid, int nxids, const TransactionId *xids, TransactionId latestXid)
bool TransactionIdIsActive(TransactionId xid)
static FullTransactionId FullXidRelativeTo(FullTransactionId rel, TransactionId xid)
bool MinimumActiveBackends(int min)
void TerminateOtherDBBackends(Oid databaseId)
#define xc_no_overflow_inc()
static TransactionId standbySnapshotPendingXmin
void ExpireAllKnownAssignedTransactionIds(void)
#define UINT32_ACCESS_ONCE(var)
VirtualTransactionId * GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid)
RunningTransactions GetRunningTransactionData(void)
TransactionId GetOldestActiveTransactionId(void)
static void KnownAssignedXidsRemoveTree(TransactionId xid, int nsubxids, TransactionId *subxids)
static int KnownAssignedXidsGetAndSetXmin(TransactionId *xarray, TransactionId *xmin, TransactionId xmax)
#define xc_by_recent_xmin_inc()
void ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
void RecordKnownAssignedTransactionIds(TransactionId xid)
static int KnownAssignedXidsGet(TransactionId *xarray, TransactionId xmax)
TransactionId GetOldestTransactionIdConsideredRunning(void)
static TransactionId latestObservedXid
static ProcArrayStruct * procArray
int GetMaxSnapshotSubxidCount(void)
int CountDBConnections(Oid databaseid)
static GlobalVisState GlobalVisTempRels
#define xc_by_my_xact_inc()
#define xc_by_known_assigned_inc()
struct ProcArrayStruct ProcArrayStruct
void CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending)
#define PROCARRAY_MAXPROCS
void GetReplicationHorizons(TransactionId *xmin, TransactionId *catalog_xmin)
static bool GlobalVisTestShouldUpdate(GlobalVisState *state)
static void ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid)
static void KnownAssignedXidsRemovePreceding(TransactionId removeXid)
void ProcArrayAdd(PGPROC *proc)
struct ComputeXidHorizonsResult ComputeXidHorizonsResult
TransactionId GlobalVisTestNonRemovableHorizon(GlobalVisState *state)
static TransactionId * KnownAssignedXids
#define xc_by_child_xid_inc()
pid_t CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode)
Snapshot GetSnapshotData(Snapshot snapshot)
static bool * KnownAssignedXidsValid
bool HaveVirtualXIDsDelayingChkpt(VirtualTransactionId *vxids, int nvxids, int type)
static void KnownAssignedXidsRemove(TransactionId xid)
void KnownAssignedTransactionIdsIdleMaintenance(void)
static void GlobalVisUpdateApply(ComputeXidHorizonsResult *horizons)
int GetMaxSnapshotXidCount(void)
GlobalVisState * GlobalVisTestFor(Relation rel)
int CountDBBackends(Oid databaseid)
bool GlobalVisCheckRemovableXid(Relation rel, TransactionId xid)
bool ProcArrayInstallRestoredXmin(TransactionId xmin, PGPROC *proc)
#define KAX_COMPRESS_FREQUENCY
void CreateSharedProcArray(void)
static TransactionId KnownAssignedXidsGetOldestXmin(void)
void ProcArrayApplyRecoveryInfo(RunningTransactions running)
void ProcArrayClearTransaction(PGPROC *proc)
VirtualTransactionId * GetCurrentVirtualXIDs(TransactionId limitXmin, bool excludeXmin0, bool allDbs, int excludeVacuum, int *nvxids)
int CountUserBackends(Oid roleid)
PGPROC * BackendPidGetProc(int pid)
static TransactionId ComputeXidHorizonsResultLastXmin
static void GlobalVisUpdate(void)
#define xc_slow_answer_inc()
static void KnownAssignedXidsDisplay(int trace_level)
#define xc_by_main_xid_inc()
PGPROC * BackendPidGetProcWithLock(int pid)
static void MaintainLatestCompletedXidRecovery(TransactionId latestXid)
static void ComputeXidHorizons(ComputeXidHorizonsResult *h)
void ProcArrayApplyXidAssignment(TransactionId topxid, int nsubxids, TransactionId *subxids)
static bool KnownAssignedXidExists(TransactionId xid)
bool CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared)
int BackendXidGetPid(TransactionId xid)
#define xc_by_latest_xid_inc()
bool IsBackendPid(int pid)
#define xc_by_known_xact_inc()
static bool KnownAssignedXidsSearch(TransactionId xid, bool remove)
static void KnownAssignedXidsReset(void)
FullTransactionId GlobalVisTestNonRemovableFullHorizon(GlobalVisState *state)
static GlobalVisHorizonKind GlobalVisHorizonKindForRel(Relation rel)
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
void ProcArrayInitRecovery(TransactionId initializedUptoXID)
void ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
#define KAX_COMPRESS_IDLE_INTERVAL
static void MaintainLatestCompletedXid(TransactionId latestXid)
static void ProcArrayGroupClearXid(PGPROC *proc, TransactionId latestXid)
void ExpireTreeKnownAssignedTransactionIds(TransactionId xid, int nsubxids, TransactionId *subxids, TransactionId max_xid)
static TransactionId cachedXidIsNotInProgress
bool ProcArrayInstallImportedXmin(TransactionId xmin, VirtualTransactionId *sourcevxid)
static bool GetSnapshotDataReuse(Snapshot snapshot)
static void KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid, bool exclusive_lock)
bool TransactionIdIsInProgress(TransactionId xid)
void ExpireOldKnownAssignedTransactionIds(TransactionId xid)
int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
#define RELATION_IS_LOCAL(relation)
#define RelationIsAccessibleInLogicalDecoding(relation)
Size add_size(Size s1, Size s2)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Size mul_size(Size s1, Size s2)
void pg_usleep(long microsec)
void MaintainOldSnapshotTimeMapping(TimestampTz whenTaken, TransactionId xmin)
TimestampTz GetSnapshotCurrentTimestamp(void)
TransactionId TransactionXmin
static bool OldSnapshotThresholdActive(void)
#define SpinLockInit(lock)
#define SpinLockRelease(lock)
#define SpinLockAcquire(lock)
int vacuum_defer_cleanup_age
void StandbyReleaseOldLocks(TransactionId oldxid)
void appendStringInfo(StringInfo str, const char *fmt,...)
void initStringInfo(StringInfo str)
TransactionId slot_catalog_xmin
TransactionId data_oldest_nonremovable
TransactionId temp_oldest_nonremovable
TransactionId shared_oldest_nonremovable
TransactionId oldest_considered_running
FullTransactionId latest_completed
TransactionId catalog_oldest_nonremovable
TransactionId shared_oldest_nonremovable_raw
FullTransactionId definitely_needed
FullTransactionId maybe_needed
bool procArrayGroupMember
pg_atomic_uint32 procArrayGroupNext
bool recoveryConflictPending