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, XLogRecPtr two_phase_at)
 
void FreeSnapshotBuilder (SnapBuild *cache)
 
void SnapBuildSnapDecRefcount (Snapshot snap)
 
Snapshot SnapBuildInitialSnapshot (SnapBuild *builder)
 
const char * SnapBuildExportSnapshot (SnapBuild *snapstate)
 
void SnapBuildClearExportedSnapshot (void)
 
void SnapBuildResetExportedSnapshotState (void)
 
SnapBuildState SnapBuildCurrentState (SnapBuild *snapstate)
 
Snapshot SnapBuildGetOrBuildSnapshot (SnapBuild *builder, TransactionId xid)
 
bool SnapBuildXactNeedsSkip (SnapBuild *snapstate, XLogRecPtr ptr)
 
XLogRecPtr SnapBuildGetTwoPhaseAt (SnapBuild *builder)
 
void SnapBuildSetTwoPhaseAt (SnapBuild *builder, 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 1 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:19
@ SNAPBUILD_START
Definition: snapbuild.h:23
@ SNAPBUILD_BUILDING_SNAPSHOT
Definition: snapbuild.h:29
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:39
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46

Function Documentation

◆ AllocateSnapshotBuilder()

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

Definition at line 280 of file snapbuild.c.

285 {
286  MemoryContext context;
287  MemoryContext oldcontext;
288  SnapBuild *builder;
289 
290  /* allocate memory in own context, to have better accountability */
292  "snapshot builder context",
294  oldcontext = MemoryContextSwitchTo(context);
295 
296  builder = palloc0(sizeof(SnapBuild));
297 
298  builder->state = SNAPBUILD_START;
299  builder->context = context;
300  builder->reorder = reorder;
301  /* Other struct members initialized by zeroing via palloc0 above */
302 
303  builder->committed.xcnt = 0;
304  builder->committed.xcnt_space = 128; /* arbitrary number */
305  builder->committed.xip =
306  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
307  builder->committed.includes_all_transactions = true;
308 
309  builder->initial_xmin_horizon = xmin_horizon;
310  builder->start_decoding_at = start_lsn;
311  builder->building_full_snapshot = need_full_snapshot;
312  builder->two_phase_at = two_phase_at;
313 
314  MemoryContextSwitchTo(oldcontext);
315 
316  return builder;
317 }
uint32 TransactionId
Definition: c.h:587
void * palloc0(Size size)
Definition: mcxt.c:1099
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
#define AllocSetContextCreate
Definition: memutils.h:173
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:197
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165
SnapBuildState state
Definition: snapbuild.c:150
TransactionId initial_xmin_horizon
Definition: snapbuild.c:182
struct SnapBuild::@16 committed
TransactionId * xip
Definition: snapbuild.c:242
XLogRecPtr two_phase_at
Definition: snapbuild.c:176
bool building_full_snapshot
Definition: snapbuild.c:185
size_t xcnt
Definition: snapbuild.c:216
size_t xcnt_space
Definition: snapbuild.c:219
bool includes_all_transactions
Definition: snapbuild.c:226
MemoryContext context
Definition: snapbuild.c:153
ReorderBuffer * reorder
Definition: snapbuild.c:200

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::two_phase_at, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by StartupDecodingContext().

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 1921 of file snapbuild.c.

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

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().

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild cache)

Definition at line 323 of file snapbuild.c.

324 {
325  MemoryContext context = builder->context;
326 
327  /* free snapshot explicitly, that contains some error checking */
328  if (builder->snapshot != NULL)
329  {
330  SnapBuildSnapDecRefcount(builder->snapshot);
331  builder->snapshot = NULL;
332  }
333 
334  /* other resources are deallocated via memory context reset */
335  MemoryContextDelete(context);
336 }
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:218
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:418

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

Referenced by FreeDecodingContext().

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 683 of file snapbuild.c.

684 {
685  ResourceOwner tmpResOwner;
686 
687  /* nothing exported, that is the usual case */
688  if (!ExportInProgress)
689  return;
690 
691  if (!IsTransactionState())
692  elog(ERROR, "clearing exported snapshot in wrong transaction state");
693 
694  /*
695  * AbortCurrentTransaction() takes care of resetting the snapshot state,
696  * so remember SavedResourceOwnerDuringExport.
697  */
698  tmpResOwner = SavedResourceOwnerDuringExport;
699 
700  /* make sure nothing could have ever happened */
702 
703  CurrentResourceOwner = tmpResOwner;
704 }
#define ERROR
Definition: elog.h:33
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:250
static bool ExportInProgress
Definition: snapbuild.c:251
bool IsTransactionState(void)
Definition: xact.c:374
void AbortCurrentTransaction(void)
Definition: xact.c:3293

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

Referenced by exec_replication_command().

◆ SnapBuildCommitTxn()

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

Definition at line 937 of file snapbuild.c.

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

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

Referenced by DecodeCommit().

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild snapstate)

Definition at line 367 of file snapbuild.c.

368 {
369  return builder->state;
370 }

References SnapBuild::state.

Referenced by DecodePrepare(), DecodingContextReady(), heap2_decode(), heap_decode(), logicalmsg_decode(), ReorderBufferCanStartStreaming(), and xact_decode().

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild snapstate)

Definition at line 622 of file snapbuild.c.

623 {
624  Snapshot snap;
625  char *snapname;
626 
628  elog(ERROR, "cannot export a snapshot from within a transaction");
629 
631  elog(ERROR, "can only export one snapshot at a time");
632 
634  ExportInProgress = true;
635 
637 
638  /* There doesn't seem to a nice API to set these */
640  XactReadOnly = true;
641 
642  snap = SnapBuildInitialSnapshot(builder);
643 
644  /*
645  * now that we've built a plain snapshot, make it active and use the
646  * normal mechanisms for exporting it
647  */
648  snapname = ExportSnapshot(snap);
649 
650  ereport(LOG,
651  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
652  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
653  snap->xcnt,
654  snapname, snap->xcnt)));
655  return snapname;
656 }
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:1014
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:530
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1123
uint32 xcnt
Definition: snapshot.h:169
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4784
bool XactReadOnly
Definition: xact.c:81
void StartTransactionCommand(void)
Definition: xact.c:2925
int XactIsoLevel
Definition: xact.c:78
#define XACT_REPEATABLE_READ
Definition: xact.h:38

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

Referenced by CreateReplicationSlot().

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder,
TransactionId  xid 
)

Definition at line 662 of file snapbuild.c.

663 {
664  Assert(builder->state == SNAPBUILD_CONSISTENT);
665 
666  /* only build a new snapshot if we don't have a prebuilt one */
667  if (builder->snapshot == NULL)
668  {
669  builder->snapshot = SnapBuildBuildSnapshot(builder);
670  /* increase refcount for the snapshot builder */
672  }
673 
674  return builder->snapshot;
675 }

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

Referenced by logicalmsg_decode().

◆ SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt ( SnapBuild builder)

Definition at line 376 of file snapbuild.c.

377 {
378  return builder->two_phase_at;
379 }

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 530 of file snapbuild.c.

531 {
532  Snapshot snap;
533  TransactionId xid;
534  TransactionId *newxip;
535  int newxcnt = 0;
536 
539 
540  if (builder->state != SNAPBUILD_CONSISTENT)
541  elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
542 
543  if (!builder->committed.includes_all_transactions)
544  elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
545 
546  /* so we don't overwrite the existing value */
548  elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
549 
550  snap = SnapBuildBuildSnapshot(builder);
551 
552  /*
553  * We know that snap->xmin is alive, enforced by the logical xmin
554  * mechanism. Due to that we can do this without locks, we're only
555  * changing our own value.
556  */
557 #ifdef USE_ASSERT_CHECKING
558  {
559  TransactionId safeXid;
560 
561  LWLockAcquire(ProcArrayLock, LW_SHARED);
562  safeXid = GetOldestSafeDecodingTransactionId(false);
563  LWLockRelease(ProcArrayLock);
564 
565  Assert(TransactionIdPrecedesOrEquals(safeXid, snap->xmin));
566  }
567 #endif
568 
569  MyProc->xmin = snap->xmin;
570 
571  /* allocate in transaction context */
572  newxip = (TransactionId *)
574 
575  /*
576  * snapbuild.c builds transactions in an "inverted" manner, which means it
577  * stores committed transactions in ->xip, not ones in progress. Build a
578  * classical snapshot by marking all non-committed transactions as
579  * in-progress. This can be expensive.
580  */
581  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
582  {
583  void *test;
584 
585  /*
586  * Check whether transaction committed using the decoding snapshot
587  * meaning of ->xip.
588  */
589  test = bsearch(&xid, snap->xip, snap->xcnt,
590  sizeof(TransactionId), xidComparator);
591 
592  if (test == NULL)
593  {
594  if (newxcnt >= GetMaxSnapshotXidCount())
595  ereport(ERROR,
597  errmsg("initial slot snapshot too large")));
598 
599  newxip[newxcnt++] = xid;
600  }
601 
603  }
604 
605  /* adjust remaining snapshot fields as needed */
607  snap->xcnt = newxcnt;
608  snap->xip = newxip;
609 
610  return snap;
611 }
int errcode(int sqlerrcode)
Definition: elog.c:693
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1196
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1800
@ LW_SHARED
Definition: lwlock.h:105
void * palloc(Size size)
Definition: mcxt.c:1068
#define ERRCODE_T_R_SERIALIZATION_FAILURE
Definition: pgbench.c:79
static void test(void)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2977
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:2085
bool FirstSnapshotSet
Definition: snapmgr.c:149
@ SNAPSHOT_MVCC
Definition: snapshot.h:50
PGPROC * MyProc
Definition: proc.c:68
TransactionId xmin
Definition: proc.h:176
TransactionId xmin
Definition: snapshot.h:157
TransactionId xmax
Definition: snapshot.h:158
SnapshotType snapshot_type
Definition: snapshot.h:144
TransactionId * xip
Definition: snapshot.h:168
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:136

References Assert(), SnapBuild::committed, elog, ereport, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, 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().

◆ SnapBuildProcessChange()

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

Definition at line 722 of file snapbuild.c.

723 {
724  /*
725  * We can't handle data in transactions if we haven't built a snapshot
726  * yet, so don't store them.
727  */
728  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
729  return false;
730 
731  /*
732  * No point in keeping track of changes in transactions that we don't have
733  * enough information about to decode. This means that they started before
734  * we got into the SNAPBUILD_FULL_SNAPSHOT state.
735  */
736  if (builder->state < SNAPBUILD_CONSISTENT &&
737  TransactionIdPrecedes(xid, builder->next_phase_at))
738  return false;
739 
740  /*
741  * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
742  * be needed to decode the change we're currently processing.
743  */
744  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
745  {
746  /* only build a new snapshot if we don't have a prebuilt one */
747  if (builder->snapshot == NULL)
748  {
749  builder->snapshot = SnapBuildBuildSnapshot(builder);
750  /* increase refcount for the snapshot builder */
752  }
753 
754  /*
755  * Increase refcount for the transaction we're handing the snapshot
756  * out to.
757  */
759  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
760  builder->snapshot);
761  }
762 
763  return true;
764 }

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

Referenced by heap2_decode(), heap_decode(), and logicalmsg_decode().

◆ SnapBuildProcessNewCid()

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

Definition at line 772 of file snapbuild.c.

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

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 heap2_decode().

◆ SnapBuildProcessRunningXacts()

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

Definition at line 1104 of file snapbuild.c.

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

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 standby_decode().

◆ SnapBuildResetExportedSnapshotState()

void SnapBuildResetExportedSnapshotState ( void  )

Definition at line 710 of file snapbuild.c.

711 {
713  ExportInProgress = false;
714 }

References ExportInProgress, and SavedResourceOwnerDuringExport.

Referenced by AbortTransaction().

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1479 of file snapbuild.c.

1480 {
1481  if (builder->state < SNAPBUILD_CONSISTENT)
1482  SnapBuildRestore(builder, lsn);
1483  else
1484  SnapBuildSerialize(builder, lsn);
1485 }
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1704

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

Referenced by xlog_decode().

◆ SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 385 of file snapbuild.c.

386 {
387  builder->two_phase_at = ptr;
388 }

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 418 of file snapbuild.c.

419 {
420  /* make sure we don't get passed an external snapshot */
422 
423  /* make sure nobody modified our snapshot */
424  Assert(snap->curcid == FirstCommandId);
425  Assert(!snap->suboverflowed);
426  Assert(!snap->takenDuringRecovery);
427 
428  Assert(snap->regd_count == 0);
429 
430  Assert(snap->active_count > 0);
431 
432  /* slightly more likely, so it's checked even without casserts */
433  if (snap->copied)
434  elog(ERROR, "cannot free a copied snapshot");
435 
436  snap->active_count--;
437  if (snap->active_count == 0)
438  SnapBuildFreeSnapshot(snap);
439 }
#define FirstCommandId
Definition: c.h:603
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:342
@ SNAPSHOT_HISTORIC_MVCC
Definition: snapshot.h:109
bool copied
Definition: snapshot.h:185
uint32 regd_count
Definition: snapshot.h:205
uint32 active_count
Definition: snapshot.h:204
CommandId curcid
Definition: snapshot.h:187
bool suboverflowed
Definition: snapshot.h:182
bool takenDuringRecovery
Definition: snapshot.h:184

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().

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild snapstate,
XLogRecPtr  ptr 
)

Definition at line 394 of file snapbuild.c.

395 {
396  return ptr < builder->start_decoding_at;
397 }

References SnapBuild::start_decoding_at.

Referenced by DecodeTXNNeedSkip(), logicalmsg_decode(), and ReorderBufferCanStartStreaming().