PostgreSQL Source Code  git master
snapbuild.h File Reference
#include "access/xlogdefs.h"
#include "utils/snapmgr.h"
Include dependency graph for snapbuild.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Typedefs

typedef struct SnapBuild SnapBuild
 

Enumerations

enum  SnapBuildState { SNAPBUILD_START = -1, SNAPBUILD_BUILDING_SNAPSHOT = 0, SNAPBUILD_FULL_SNAPSHOT = 1, SNAPBUILD_CONSISTENT = 2 }
 

Functions

void CheckPointSnapBuild (void)
 
SnapBuildAllocateSnapshotBuilder (struct ReorderBuffer *cache, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot)
 
void FreeSnapshotBuilder (SnapBuild *cache)
 
void SnapBuildSnapDecRefcount (Snapshot snap)
 
Snapshot SnapBuildInitialSnapshot (SnapBuild *builder)
 
const char * SnapBuildExportSnapshot (SnapBuild *snapstate)
 
void SnapBuildClearExportedSnapshot (void)
 
SnapBuildState SnapBuildCurrentState (SnapBuild *snapstate)
 
Snapshot SnapBuildGetOrBuildSnapshot (SnapBuild *builder, TransactionId xid)
 
bool SnapBuildXactNeedsSkip (SnapBuild *snapstate, XLogRecPtr ptr)
 
void SnapBuildCommitTxn (SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
 
bool SnapBuildProcessChange (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
 
void SnapBuildProcessNewCid (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, struct xl_heap_new_cid *cid)
 
void SnapBuildProcessRunningXacts (SnapBuild *builder, XLogRecPtr lsn, struct xl_running_xacts *running)
 
void SnapBuildSerializationPoint (SnapBuild *builder, XLogRecPtr lsn)
 

Typedef Documentation

◆ SnapBuild

typedef struct SnapBuild SnapBuild

Definition at line 51 of file snapbuild.h.

Enumeration Type Documentation

◆ SnapBuildState

Enumerator
SNAPBUILD_START 
SNAPBUILD_BUILDING_SNAPSHOT 
SNAPBUILD_FULL_SNAPSHOT 
SNAPBUILD_CONSISTENT 

Definition at line 18 of file snapbuild.h.

19 {
20  /*
21  * Initial state, we can't do much yet.
22  */
23  SNAPBUILD_START = -1,
24 
25  /*
26  * Collecting committed transactions, to build the initial catalog
27  * snapshot.
28  */
30 
31  /*
32  * We have collected enough information to decode tuples in transactions
33  * that started after this.
34  *
35  * Once we reached this we start to collect changes. We cannot apply them
36  * yet, because they might be based on transactions that were still
37  * running when FULL_SNAPSHOT was reached.
38  */
40 
41  /*
42  * Found a point after SNAPBUILD_FULL_SNAPSHOT where all transactions that
43  * were running at that point finished. Till we reach that we hold off
44  * calling any commit callbacks.
45  */
SnapBuildState
Definition: snapbuild.h:18

Function Documentation

◆ AllocateSnapshotBuilder()

SnapBuild* AllocateSnapshotBuilder ( struct ReorderBuffer cache,
TransactionId  xmin_horizon,
XLogRecPtr  start_lsn,
bool  need_full_snapshot 
)

Definition at line 310 of file snapbuild.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, SnapBuild::building_full_snapshot, SnapBuild::committed, SnapBuild::context, CurrentMemoryContext, SnapBuild::includes_all_transactions, SnapBuild::initial_xmin_horizon, MemoryContextSwitchTo(), palloc0(), SnapBuild::reorder, SNAPBUILD_START, SnapBuild::start_decoding_at, SnapBuild::state, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by StartupDecodingContext().

314 {
315  MemoryContext context;
316  MemoryContext oldcontext;
317  SnapBuild *builder;
318 
319  /* allocate memory in own context, to have better accountability */
321  "snapshot builder context",
323  oldcontext = MemoryContextSwitchTo(context);
324 
325  builder = palloc0(sizeof(SnapBuild));
326 
327  builder->state = SNAPBUILD_START;
328  builder->context = context;
329  builder->reorder = reorder;
330  /* Other struct members initialized by zeroing via palloc0 above */
331 
332  builder->committed.xcnt = 0;
333  builder->committed.xcnt_space = 128; /* arbitrary number */
334  builder->committed.xip =
335  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
336  builder->committed.includes_all_transactions = true;
337 
338  builder->initial_xmin_horizon = xmin_horizon;
339  builder->start_decoding_at = start_lsn;
340  builder->building_full_snapshot = need_full_snapshot;
341 
342  MemoryContextSwitchTo(oldcontext);
343 
344  return builder;
345 }
#define AllocSetContextCreate
Definition: memutils.h:170
uint32 TransactionId
Definition: c.h:521
bool building_full_snapshot
Definition: snapbuild.c:174
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
size_t xcnt_space
Definition: snapbuild.c:221
size_t xcnt
Definition: snapbuild.c:218
SnapBuildState state
Definition: snapbuild.c:150
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:192
ReorderBuffer * reorder
Definition: snapbuild.c:189
TransactionId initial_xmin_horizon
Definition: snapbuild.c:171
TransactionId * xip
Definition: snapbuild.c:244
bool includes_all_transactions
Definition: snapbuild.c:228
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
void * palloc0(Size size)
Definition: mcxt.c:981
MemoryContext context
Definition: snapbuild.c:153
struct SnapBuild::@23 committed
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 1933 of file snapbuild.c.

References AllocateDir(), dirent::d_name, DEBUG1, elog, ereport, errcode_for_file_access(), errmsg(), FreeDir(), GetRedoRecPtr(), InvalidXLogRecPtr, LOG, lstat, MAXPGPATH, ReadDir(), ReplicationSlotsComputeLogicalRestartLSN(), S_ISREG, snprintf, and stat::st_mode.

Referenced by CheckPointGuts().

1934 {
1935  XLogRecPtr cutoff;
1936  XLogRecPtr redo;
1937  DIR *snap_dir;
1938  struct dirent *snap_de;
1939  char path[MAXPGPATH + 21];
1940 
1941  /*
1942  * We start off with a minimum of the last redo pointer. No new
1943  * replication slot will start before that, so that's a safe upper bound
1944  * for removal.
1945  */
1946  redo = GetRedoRecPtr();
1947 
1948  /* now check for the restart ptrs from existing slots */
1950 
1951  /* don't start earlier than the restart lsn */
1952  if (redo < cutoff)
1953  cutoff = redo;
1954 
1955  snap_dir = AllocateDir("pg_logical/snapshots");
1956  while ((snap_de = ReadDir(snap_dir, "pg_logical/snapshots")) != NULL)
1957  {
1958  uint32 hi;
1959  uint32 lo;
1960  XLogRecPtr lsn;
1961  struct stat statbuf;
1962 
1963  if (strcmp(snap_de->d_name, ".") == 0 ||
1964  strcmp(snap_de->d_name, "..") == 0)
1965  continue;
1966 
1967  snprintf(path, sizeof(path), "pg_logical/snapshots/%s", snap_de->d_name);
1968 
1969  if (lstat(path, &statbuf) == 0 && !S_ISREG(statbuf.st_mode))
1970  {
1971  elog(DEBUG1, "only regular files expected: %s", path);
1972  continue;
1973  }
1974 
1975  /*
1976  * temporary filenames from SnapBuildSerialize() include the LSN and
1977  * everything but are postfixed by .$pid.tmp. We can just remove them
1978  * the same as other files because there can be none that are
1979  * currently being written that are older than cutoff.
1980  *
1981  * We just log a message if a file doesn't fit the pattern, it's
1982  * probably some editors lock/state file or similar...
1983  */
1984  if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
1985  {
1986  ereport(LOG,
1987  (errmsg("could not parse file name \"%s\"", path)));
1988  continue;
1989  }
1990 
1991  lsn = ((uint64) hi) << 32 | lo;
1992 
1993  /* check whether we still need it */
1994  if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
1995  {
1996  elog(DEBUG1, "removing snapbuild snapshot %s", path);
1997 
1998  /*
1999  * It's not particularly harmful, though strange, if we can't
2000  * remove the file here. Don't prevent the checkpoint from
2001  * completing, that'd be a cure worse than the disease.
2002  */
2003  if (unlink(path) < 0)
2004  {
2005  ereport(LOG,
2007  errmsg("could not remove file \"%s\": %m",
2008  path)));
2009  continue;
2010  }
2011  }
2012  }
2013  FreeDir(snap_dir);
2014 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
#define LOG
Definition: elog.h:26
Definition: dirent.h:9
Definition: dirent.c:25
#define MAXPGPATH
int errcode_for_file_access(void)
Definition: elog.c:633
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:869
unsigned int uint32
Definition: c.h:375
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2583
#define S_ISREG(m)
Definition: win32_port.h:319
#define ereport(elevel,...)
Definition: elog.h:144
uint64 XLogRecPtr
Definition: xlogdefs.h:21
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2649
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8368
#define lstat(path, sb)
Definition: win32_port.h:276
int errmsg(const char *fmt,...)
Definition: elog.c:821
#define elog(elevel,...)
Definition: elog.h:214
char d_name[MAX_PATH]
Definition: dirent.h:15
#define snprintf
Definition: port.h:215
int FreeDir(DIR *dir)
Definition: fd.c:2701

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild cache)

Definition at line 351 of file snapbuild.c.

References SnapBuild::context, MemoryContextDelete(), SnapBuildSnapDecRefcount(), and SnapBuild::snapshot.

Referenced by FreeDecodingContext().

352 {
353  MemoryContext context = builder->context;
354 
355  /* free snapshot explicitly, that contains some error checking */
356  if (builder->snapshot != NULL)
357  {
358  SnapBuildSnapDecRefcount(builder->snapshot);
359  builder->snapshot = NULL;
360  }
361 
362  /* other resources are deallocated via memory context reset */
363  MemoryContextDelete(context);
364 }
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:212
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:428

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 693 of file snapbuild.c.

References AbortCurrentTransaction(), CurrentResourceOwner, elog, ERROR, ExportInProgress, IsTransactionState(), and SavedResourceOwnerDuringExport.

Referenced by exec_replication_command().

694 {
695  /* nothing exported, that is the usual case */
696  if (!ExportInProgress)
697  return;
698 
699  if (!IsTransactionState())
700  elog(ERROR, "clearing exported snapshot in wrong transaction state");
701 
702  /* make sure nothing could have ever happened */
704 
707  ExportInProgress = false;
708 }
void AbortCurrentTransaction(void)
Definition: xact.c:3211
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:252
#define ERROR
Definition: elog.h:43
static bool ExportInProgress
Definition: snapbuild.c:253
bool IsTransactionState(void)
Definition: xact.c:371
#define elog(elevel,...)
Definition: elog.h:214

◆ SnapBuildCommitTxn()

void SnapBuildCommitTxn ( SnapBuild builder,
XLogRecPtr  lsn,
TransactionId  xid,
int  nsubxacts,
TransactionId subxacts 
)

Definition at line 924 of file snapbuild.c.

References Assert, SnapBuild::building_full_snapshot, SnapBuild::committed, DEBUG1, DEBUG2, elog, SnapBuild::includes_all_transactions, NormalTransactionIdFollows, SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), ReorderBufferXidHasCatalogChanges(), SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildAddCommittedTxn(), SnapBuildBuildSnapshot(), SnapBuildDistributeNewCatalogSnapshot(), SnapBuildNextPhaseAt(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdAdvance, TransactionIdFollowsOrEquals(), TransactionIdIsValid, TransactionIdPrecedes(), and SnapBuild::xmax.

Referenced by DecodeCommit().

926 {
927  int nxact;
928 
929  bool needs_snapshot = false;
930  bool needs_timetravel = false;
931  bool sub_needs_timetravel = false;
932 
933  TransactionId xmax = xid;
934 
935  /*
936  * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
937  * will they be part of a snapshot. So we don't need to record anything.
938  */
939  if (builder->state == SNAPBUILD_START ||
940  (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
942  {
943  /* ensure that only commits after this are getting replayed */
944  if (builder->start_decoding_at <= lsn)
945  builder->start_decoding_at = lsn + 1;
946  return;
947  }
948 
949  if (builder->state < SNAPBUILD_CONSISTENT)
950  {
951  /* ensure that only commits after this are getting replayed */
952  if (builder->start_decoding_at <= lsn)
953  builder->start_decoding_at = lsn + 1;
954 
955  /*
956  * If building an exportable snapshot, force xid to be tracked, even
957  * if the transaction didn't modify the catalog.
958  */
959  if (builder->building_full_snapshot)
960  {
961  needs_timetravel = true;
962  }
963  }
964 
965  for (nxact = 0; nxact < nsubxacts; nxact++)
966  {
967  TransactionId subxid = subxacts[nxact];
968 
969  /*
970  * Add subtransaction to base snapshot if catalog modifying, we don't
971  * distinguish to toplevel transactions there.
972  */
973  if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid))
974  {
975  sub_needs_timetravel = true;
976  needs_snapshot = true;
977 
978  elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
979  xid, subxid);
980 
981  SnapBuildAddCommittedTxn(builder, subxid);
982 
983  if (NormalTransactionIdFollows(subxid, xmax))
984  xmax = subxid;
985  }
986 
987  /*
988  * If we're forcing timetravel we also need visibility information
989  * about subtransaction, so keep track of subtransaction's state, even
990  * if not catalog modifying. Don't need to distribute a snapshot in
991  * that case.
992  */
993  else if (needs_timetravel)
994  {
995  SnapBuildAddCommittedTxn(builder, subxid);
996  if (NormalTransactionIdFollows(subxid, xmax))
997  xmax = subxid;
998  }
999  }
1000 
1001  /* if top-level modified catalog, it'll need a snapshot */
1002  if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
1003  {
1004  elog(DEBUG2, "found top level transaction %u, with catalog changes",
1005  xid);
1006  needs_snapshot = true;
1007  needs_timetravel = true;
1008  SnapBuildAddCommittedTxn(builder, xid);
1009  }
1010  else if (sub_needs_timetravel)
1011  {
1012  /* track toplevel txn as well, subxact alone isn't meaningful */
1013  SnapBuildAddCommittedTxn(builder, xid);
1014  }
1015  else if (needs_timetravel)
1016  {
1017  elog(DEBUG2, "forced transaction %u to do timetravel", xid);
1018 
1019  SnapBuildAddCommittedTxn(builder, xid);
1020  }
1021 
1022  if (!needs_timetravel)
1023  {
1024  /* record that we cannot export a general snapshot anymore */
1025  builder->committed.includes_all_transactions = false;
1026  }
1027 
1028  Assert(!needs_snapshot || needs_timetravel);
1029 
1030  /*
1031  * Adjust xmax of the snapshot builder, we only do that for committed,
1032  * catalog modifying, transactions, everything else isn't interesting for
1033  * us since we'll never look at the respective rows.
1034  */
1035  if (needs_timetravel &&
1036  (!TransactionIdIsValid(builder->xmax) ||
1037  TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1038  {
1039  builder->xmax = xmax;
1040  TransactionIdAdvance(builder->xmax);
1041  }
1042 
1043  /* if there's any reason to build a historic snapshot, do so now */
1044  if (needs_snapshot)
1045  {
1046  /*
1047  * If we haven't built a complete snapshot yet there's no need to hand
1048  * it out, it wouldn't (and couldn't) be used anyway.
1049  */
1050  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
1051  return;
1052 
1053  /*
1054  * Decrease the snapshot builder's refcount of the old snapshot, note
1055  * that it still will be used if it has been handed out to the
1056  * reorderbuffer earlier.
1057  */
1058  if (builder->snapshot)
1060 
1061  builder->snapshot = SnapBuildBuildSnapshot(builder);
1062 
1063  /* we might need to execute invalidations, add snapshot */
1064  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1065  {
1067  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1068  builder->snapshot);
1069  }
1070 
1071  /* refcount of the snapshot builder for the new snapshot */
1073 
1074  /* add a new catalog snapshot to all currently running transactions */
1076  }
1077 }
#define TransactionIdAdvance(dest)
Definition: transam.h:91
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
#define DEBUG1
Definition: elog.h:25
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:854
uint32 TransactionId
Definition: c.h:521
Snapshot snapshot
Definition: snapbuild.c:179
bool building_full_snapshot
Definition: snapbuild.c:174
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:349
SnapBuildState state
Definition: snapbuild.c:150
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:808
#define DEBUG2
Definition: elog.h:24
ReorderBuffer * reorder
Definition: snapbuild.c:189
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:460
bool includes_all_transactions
Definition: snapbuild.c:228
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
TransactionId xmax
Definition: snapbuild.c:159
#define Assert(condition)
Definition: c.h:746
#define NormalTransactionIdFollows(id1, id2)
Definition: transam.h:152
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:428
#define elog(elevel,...)
Definition: elog.h:214
struct SnapBuild::@23 committed
#define TransactionIdIsValid(xid)
Definition: transam.h:41
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:416
static TransactionId SnapBuildNextPhaseAt(SnapBuild *builder)
Definition: snapbuild.c:280
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild snapstate)

Definition at line 395 of file snapbuild.c.

References SnapBuild::state.

Referenced by DecodeHeap2Op(), DecodeHeapOp(), DecodeLogicalMsgOp(), DecodeXactOp(), DecodingContextReady(), and ReorderBufferCanStartStreaming().

396 {
397  return builder->state;
398 }

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild snapstate)

Definition at line 632 of file snapbuild.c.

References CurrentResourceOwner, elog, ereport, errmsg_plural(), ERROR, ExportInProgress, ExportSnapshot(), IsTransactionOrTransactionBlock(), LOG, SnapBuildInitialSnapshot(), StartTransactionCommand(), XACT_REPEATABLE_READ, XactIsoLevel, XactReadOnly, and SnapshotData::xcnt.

Referenced by CreateReplicationSlot().

633 {
634  Snapshot snap;
635  char *snapname;
636 
638  elog(ERROR, "cannot export a snapshot from within a transaction");
639 
641  elog(ERROR, "can only export one snapshot at a time");
642 
644  ExportInProgress = true;
645 
647 
648  /* There doesn't seem to a nice API to set these */
650  XactReadOnly = true;
651 
652  snap = SnapBuildInitialSnapshot(builder);
653 
654  /*
655  * now that we've built a plain snapshot, make it active and use the
656  * normal mechanisms for exporting it
657  */
658  snapname = ExportSnapshot(snap);
659 
660  ereport(LOG,
661  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
662  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
663  snap->xcnt,
664  snapname, snap->xcnt)));
665  return snapname;
666 }
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:931
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
#define XACT_REPEATABLE_READ
Definition: xact.h:38
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4702
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1106
#define LOG
Definition: elog.h:26
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:252
#define ERROR
Definition: elog.h:43
static bool ExportInProgress
Definition: snapbuild.c:253
#define ereport(elevel,...)
Definition: elog.h:144
bool XactReadOnly
Definition: xact.c:78
void StartTransactionCommand(void)
Definition: xact.c:2846
int XactIsoLevel
Definition: xact.c:75
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:540
uint32 xcnt
Definition: snapshot.h:169
#define elog(elevel,...)
Definition: elog.h:214

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder,
TransactionId  xid 
)

Definition at line 672 of file snapbuild.c.

References Assert, SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, and SnapBuild::state.

Referenced by DecodeLogicalMsgOp().

673 {
674  Assert(builder->state == SNAPBUILD_CONSISTENT);
675 
676  /* only build a new snapshot if we don't have a prebuilt one */
677  if (builder->snapshot == NULL)
678  {
679  builder->snapshot = SnapBuildBuildSnapshot(builder);
680  /* increase refcount for the snapshot builder */
682  }
683 
684  return builder->snapshot;
685 }
Snapshot snapshot
Definition: snapbuild.c:179
SnapBuildState state
Definition: snapbuild.c:150
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:460
#define Assert(condition)
Definition: c.h:746
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:416

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 540 of file snapbuild.c.

References Assert, SnapBuild::committed, elog, ereport, errcode(), errmsg(), ERROR, FirstSnapshotSet, GetMaxSnapshotXidCount(), GetOldestSafeDecodingTransactionId(), SnapBuild::includes_all_transactions, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyProc, NormalTransactionIdPrecedes, palloc(), SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SNAPSHOT_MVCC, SnapshotData::snapshot_type, SnapBuild::state, test(), TransactionIdAdvance, TransactionIdIsValid, TransactionIdPrecedesOrEquals(), XACT_REPEATABLE_READ, XactIsoLevel, SnapshotData::xcnt, xidComparator(), SnapshotData::xip, SnapshotData::xmax, PGPROC::xmin, and SnapshotData::xmin.

Referenced by CreateReplicationSlot(), and SnapBuildExportSnapshot().

541 {
542  Snapshot snap;
543  TransactionId xid;
544  TransactionId *newxip;
545  int newxcnt = 0;
546 
549 
550  if (builder->state != SNAPBUILD_CONSISTENT)
551  elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
552 
553  if (!builder->committed.includes_all_transactions)
554  elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
555 
556  /* so we don't overwrite the existing value */
558  elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
559 
560  snap = SnapBuildBuildSnapshot(builder);
561 
562  /*
563  * We know that snap->xmin is alive, enforced by the logical xmin
564  * mechanism. Due to that we can do this without locks, we're only
565  * changing our own value.
566  */
567 #ifdef USE_ASSERT_CHECKING
568  {
569  TransactionId safeXid;
570 
571  LWLockAcquire(ProcArrayLock, LW_SHARED);
572  safeXid = GetOldestSafeDecodingTransactionId(false);
573  LWLockRelease(ProcArrayLock);
574 
575  Assert(TransactionIdPrecedesOrEquals(safeXid, snap->xmin));
576  }
577 #endif
578 
579  MyProc->xmin = snap->xmin;
580 
581  /* allocate in transaction context */
582  newxip = (TransactionId *)
584 
585  /*
586  * snapbuild.c builds transactions in an "inverted" manner, which means it
587  * stores committed transactions in ->xip, not ones in progress. Build a
588  * classical snapshot by marking all non-committed transactions as
589  * in-progress. This can be expensive.
590  */
591  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
592  {
593  void *test;
594 
595  /*
596  * Check whether transaction committed using the decoding snapshot
597  * meaning of ->xip.
598  */
599  test = bsearch(&xid, snap->xip, snap->xcnt,
600  sizeof(TransactionId), xidComparator);
601 
602  if (test == NULL)
603  {
604  if (newxcnt >= GetMaxSnapshotXidCount())
605  ereport(ERROR,
606  (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
607  errmsg("initial slot snapshot too large")));
608 
609  newxip[newxcnt++] = xid;
610  }
611 
613  }
614 
615  /* adjust remaining snapshot fields as needed */
617  snap->xcnt = newxcnt;
618  snap->xip = newxip;
619 
620  return snap;
621 }
#define TransactionIdAdvance(dest)
Definition: transam.h:91
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2794
static void test(void)
uint32 TransactionId
Definition: c.h:521
PGPROC * MyProc
Definition: proc.c:67
#define XACT_REPEATABLE_READ
Definition: xact.h:38
int errcode(int sqlerrcode)
Definition: elog.c:610
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
SnapBuildState state
Definition: snapbuild.c:150
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
#define ERROR
Definition: elog.h:43
bool FirstSnapshotSet
Definition: snapmgr.c:149
TransactionId xmin
Definition: proc.h:129
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:460
bool includes_all_transactions
Definition: snapbuild.c:228
SnapshotType snapshot_type
Definition: snapshot.h:144
TransactionId xmax
Definition: snapshot.h:158
TransactionId xmin
Definition: snapshot.h:157
TransactionId * xip
Definition: snapshot.h:168
#define ereport(elevel,...)
Definition: elog.h:144
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:1927
#define Assert(condition)
Definition: c.h:746
int XactIsoLevel
Definition: xact.c:75
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
uint32 xcnt
Definition: snapshot.h:169
void * palloc(Size size)
Definition: mcxt.c:950
int errmsg(const char *fmt,...)
Definition: elog.c:821
#define elog(elevel,...)
Definition: elog.h:214
struct SnapBuild::@23 committed
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:139

◆ SnapBuildProcessChange()

bool SnapBuildProcessChange ( SnapBuild builder,
TransactionId  xid,
XLogRecPtr  lsn 
)

Definition at line 716 of file snapbuild.c.

References SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SnapBuildBuildSnapshot(), SnapBuildNextPhaseAt(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::state, and TransactionIdPrecedes().

Referenced by DecodeHeap2Op(), DecodeHeapOp(), and DecodeLogicalMsgOp().

717 {
718  /*
719  * We can't handle data in transactions if we haven't built a snapshot
720  * yet, so don't store them.
721  */
722  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
723  return false;
724 
725  /*
726  * No point in keeping track of changes in transactions that we don't have
727  * enough information about to decode. This means that they started before
728  * we got into the SNAPBUILD_FULL_SNAPSHOT state.
729  */
730  if (builder->state < SNAPBUILD_CONSISTENT &&
732  return false;
733 
734  /*
735  * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
736  * be needed to decode the change we're currently processing.
737  */
738  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
739  {
740  /* only build a new snapshot if we don't have a prebuilt one */
741  if (builder->snapshot == NULL)
742  {
743  builder->snapshot = SnapBuildBuildSnapshot(builder);
744  /* increase refcount for the snapshot builder */
746  }
747 
748  /*
749  * Increase refcount for the transaction we're handing the snapshot
750  * out to.
751  */
753  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
754  builder->snapshot);
755  }
756 
757  return true;
758 }
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
Snapshot snapshot
Definition: snapbuild.c:179
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
SnapBuildState state
Definition: snapbuild.c:150
ReorderBuffer * reorder
Definition: snapbuild.c:189
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:460
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:416
static TransactionId SnapBuildNextPhaseAt(SnapBuild *builder)
Definition: snapbuild.c:280

◆ SnapBuildProcessNewCid()

void SnapBuildProcessNewCid ( SnapBuild builder,
TransactionId  xid,
XLogRecPtr  lsn,
struct xl_heap_new_cid cid 
)

Definition at line 766 of file snapbuild.c.

References xl_heap_new_cid::cmax, xl_heap_new_cid::cmin, xl_heap_new_cid::combocid, elog, ERROR, InvalidCommandId, Max, SnapBuild::reorder, ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), ReorderBufferXidSetCatalogChanges(), xl_heap_new_cid::target_node, xl_heap_new_cid::target_tid, and xl_heap_new_cid::top_xid.

Referenced by DecodeHeap2Op().

768 {
769  CommandId cid;
770 
771  /*
772  * we only log new_cid's if a catalog tuple was modified, so mark the
773  * transaction as containing catalog modifications
774  */
775  ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
776 
777  ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
778  xlrec->target_node, xlrec->target_tid,
779  xlrec->cmin, xlrec->cmax,
780  xlrec->combocid);
781 
782  /* figure out new command id */
783  if (xlrec->cmin != InvalidCommandId &&
784  xlrec->cmax != InvalidCommandId)
785  cid = Max(xlrec->cmin, xlrec->cmax);
786  else if (xlrec->cmax != InvalidCommandId)
787  cid = xlrec->cmax;
788  else if (xlrec->cmin != InvalidCommandId)
789  cid = xlrec->cmin;
790  else
791  {
792  cid = InvalidCommandId; /* silence compiler */
793  elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
794  }
795 
796  ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
797 }
uint32 CommandId
Definition: c.h:535
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileNode node, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
#define ERROR
Definition: elog.h:43
ReorderBuffer * reorder
Definition: snapbuild.c:189
#define InvalidCommandId
Definition: c.h:538
#define Max(x, y)
Definition: c.h:922
#define elog(elevel,...)
Definition: elog.h:214
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)

◆ SnapBuildProcessRunningXacts()

void SnapBuildProcessRunningXacts ( SnapBuild builder,
XLogRecPtr  lsn,
struct xl_running_xacts running 
)

Definition at line 1091 of file snapbuild.c.

References ReorderBuffer::current_restart_decoding_lsn, DEBUG3, elog, InvalidTransactionId, InvalidXLogRecPtr, SnapBuild::last_serialized_snapshot, LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), xl_running_xacts::oldestRunningXid, SnapBuild::reorder, ReorderBufferGetOldestTXN(), ReorderBufferGetOldestXmin(), ReorderBufferTXN::restart_decoding_lsn, SNAPBUILD_CONSISTENT, SnapBuildFindSnapshot(), SnapBuildPurgeCommittedTxn(), SnapBuildSerialize(), SnapBuild::state, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by DecodeStandbyOp().

1092 {
1093  ReorderBufferTXN *txn;
1094  TransactionId xmin;
1095 
1096  /*
1097  * If we're not consistent yet, inspect the record to see whether it
1098  * allows to get closer to being consistent. If we are consistent, dump
1099  * our snapshot so others or we, after a restart, can use it.
1100  */
1101  if (builder->state < SNAPBUILD_CONSISTENT)
1102  {
1103  /* returns false if there's no point in performing cleanup just yet */
1104  if (!SnapBuildFindSnapshot(builder, lsn, running))
1105  return;
1106  }
1107  else
1108  SnapBuildSerialize(builder, lsn);
1109 
1110  /*
1111  * Update range of interesting xids based on the running xacts
1112  * information. We don't increase ->xmax using it, because once we are in
1113  * a consistent state we can do that ourselves and much more efficiently
1114  * so, because we only need to do it for catalog transactions since we
1115  * only ever look at those.
1116  *
1117  * NB: We only increase xmax when a catalog modifying transaction commits
1118  * (see SnapBuildCommitTxn). Because of this, xmax can be lower than
1119  * xmin, which looks odd but is correct and actually more efficient, since
1120  * we hit fast paths in heapam_visibility.c.
1121  */
1122  builder->xmin = running->oldestRunningXid;
1123 
1124  /* Remove transactions we don't need to keep track off anymore */
1125  SnapBuildPurgeCommittedTxn(builder);
1126 
1127  /*
1128  * Advance the xmin limit for the current replication slot, to allow
1129  * vacuum to clean up the tuples this slot has been protecting.
1130  *
1131  * The reorderbuffer might have an xmin among the currently running
1132  * snapshots; use it if so. If not, we need only consider the snapshots
1133  * we'll produce later, which can't be less than the oldest running xid in
1134  * the record we're reading now.
1135  */
1136  xmin = ReorderBufferGetOldestXmin(builder->reorder);
1137  if (xmin == InvalidTransactionId)
1138  xmin = running->oldestRunningXid;
1139  elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1140  builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
1141  LogicalIncreaseXminForSlot(lsn, xmin);
1142 
1143  /*
1144  * Also tell the slot where we can restart decoding from. We don't want to
1145  * do that after every commit because changing that implies an fsync of
1146  * the logical slot's state file, so we only do it every time we see a
1147  * running xacts record.
1148  *
1149  * Do so by looking for the oldest in progress transaction (determined by
1150  * the first LSN of any of its relevant records). Every transaction
1151  * remembers the last location we stored the snapshot to disk before its
1152  * beginning. That point is where we can restart from.
1153  */
1154 
1155  /*
1156  * Can't know about a serialized snapshot's location if we're not
1157  * consistent.
1158  */
1159  if (builder->state < SNAPBUILD_CONSISTENT)
1160  return;
1161 
1162  txn = ReorderBufferGetOldestTXN(builder->reorder);
1163 
1164  /*
1165  * oldest ongoing txn might have started when we didn't yet serialize
1166  * anything because we hadn't reached a consistent state yet.
1167  */
1168  if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
1170 
1171  /*
1172  * No in-progress transaction, can reuse the last serialized snapshot if
1173  * we have one.
1174  */
1175  else if (txn == NULL &&
1179  builder->last_serialized_snapshot);
1180 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
uint32 TransactionId
Definition: c.h:521
#define DEBUG3
Definition: elog.h:23
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1295
XLogRecPtr current_restart_decoding_lsn
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1193
SnapBuildState state
Definition: snapbuild.c:150
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1481
ReorderBuffer * reorder
Definition: snapbuild.c:189
#define InvalidTransactionId
Definition: transam.h:31
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:184
TransactionId xmax
Definition: snapbuild.c:159
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1238
TransactionId xmin
Definition: snapbuild.c:156
#define elog(elevel,...)
Definition: elog.h:214
XLogRecPtr restart_decoding_lsn
TransactionId oldestRunningXid
Definition: standbydefs.h:53
static void SnapBuildPurgeCommittedTxn(SnapBuild *builder)
Definition: snapbuild.c:883

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1468 of file snapbuild.c.

References SNAPBUILD_CONSISTENT, SnapBuildRestore(), SnapBuildSerialize(), and SnapBuild::state.

Referenced by DecodeXLogOp().

1469 {
1470  if (builder->state < SNAPBUILD_CONSISTENT)
1471  SnapBuildRestore(builder, lsn);
1472  else
1473  SnapBuildSerialize(builder, lsn);
1474 }
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1690
SnapBuildState state
Definition: snapbuild.c:150
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1481

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 428 of file snapbuild.c.

References SnapshotData::active_count, Assert, SnapshotData::copied, SnapshotData::curcid, elog, ERROR, FirstCommandId, SnapshotData::regd_count, SnapBuildFreeSnapshot(), SNAPSHOT_HISTORIC_MVCC, SnapshotData::snapshot_type, SnapshotData::suboverflowed, and SnapshotData::takenDuringRecovery.

Referenced by FreeSnapshotBuilder(), ReorderBufferCleanupTXN(), ReorderBufferFreeSnap(), ReorderBufferTransferSnapToParent(), SnapBuildCommitTxn(), and SnapBuildRestore().

429 {
430  /* make sure we don't get passed an external snapshot */
432 
433  /* make sure nobody modified our snapshot */
434  Assert(snap->curcid == FirstCommandId);
435  Assert(!snap->suboverflowed);
436  Assert(!snap->takenDuringRecovery);
437 
438  Assert(snap->regd_count == 0);
439 
440  Assert(snap->active_count > 0);
441 
442  /* slightly more likely, so it's checked even without casserts */
443  if (snap->copied)
444  elog(ERROR, "cannot free a copied snapshot");
445 
446  snap->active_count--;
447  if (snap->active_count == 0)
448  SnapBuildFreeSnapshot(snap);
449 }
bool copied
Definition: snapshot.h:185
bool suboverflowed
Definition: snapshot.h:182
uint32 regd_count
Definition: snapshot.h:205
#define FirstCommandId
Definition: c.h:537
#define ERROR
Definition: elog.h:43
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:370
SnapshotType snapshot_type
Definition: snapshot.h:144
CommandId curcid
Definition: snapshot.h:187
#define Assert(condition)
Definition: c.h:746
bool takenDuringRecovery
Definition: snapshot.h:184
#define elog(elevel,...)
Definition: elog.h:214
uint32 active_count
Definition: snapshot.h:204

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild snapstate,
XLogRecPtr  ptr 
)

Definition at line 404 of file snapbuild.c.

References SnapBuild::start_decoding_at.

Referenced by DecodeCommit(), DecodeLogicalMsgOp(), and ReorderBufferCanStartStreaming().

405 {
406  return ptr < builder->start_decoding_at;
407 }