PostgreSQL Source Code  git master
snapbuild.c File Reference
#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/block.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/standby.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/snapshot.h"
Include dependency graph for snapbuild.c:

Go to the source code of this file.

Data Structures

struct  SnapBuild
 
struct  SnapBuildOnDisk
 

Macros

#define SnapBuildOnDiskConstantSize   offsetof(SnapBuildOnDisk, builder)
 
#define SnapBuildOnDiskNotChecksummedSize   offsetof(SnapBuildOnDisk, version)
 
#define SNAPBUILD_MAGIC   0x51A1E001
 
#define SNAPBUILD_VERSION   4
 

Typedefs

typedef struct SnapBuildOnDisk SnapBuildOnDisk
 

Functions

static void SnapBuildPurgeCommittedTxn (SnapBuild *builder)
 
static Snapshot SnapBuildBuildSnapshot (SnapBuild *builder)
 
static void SnapBuildFreeSnapshot (Snapshot snap)
 
static void SnapBuildSnapIncRefcount (Snapshot snap)
 
static void SnapBuildDistributeNewCatalogSnapshot (SnapBuild *builder, XLogRecPtr lsn)
 
static bool SnapBuildFindSnapshot (SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
 
static void SnapBuildWaitSnapshot (xl_running_xacts *running, TransactionId cutoff)
 
static void SnapBuildSerialize (SnapBuild *builder, XLogRecPtr lsn)
 
static bool SnapBuildRestore (SnapBuild *builder, XLogRecPtr lsn)
 
SnapBuildAllocateSnapshotBuilder (ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
 
void FreeSnapshotBuilder (SnapBuild *builder)
 
SnapBuildState SnapBuildCurrentState (SnapBuild *builder)
 
XLogRecPtr SnapBuildGetTwoPhaseAt (SnapBuild *builder)
 
void SnapBuildSetTwoPhaseAt (SnapBuild *builder, XLogRecPtr ptr)
 
bool SnapBuildXactNeedsSkip (SnapBuild *builder, XLogRecPtr ptr)
 
void SnapBuildSnapDecRefcount (Snapshot snap)
 
Snapshot SnapBuildInitialSnapshot (SnapBuild *builder)
 
const char * SnapBuildExportSnapshot (SnapBuild *builder)
 
Snapshot SnapBuildGetOrBuildSnapshot (SnapBuild *builder, TransactionId xid)
 
void SnapBuildClearExportedSnapshot (void)
 
void SnapBuildResetExportedSnapshotState (void)
 
bool SnapBuildProcessChange (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
 
void SnapBuildProcessNewCid (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
 
static void SnapBuildAddCommittedTxn (SnapBuild *builder, TransactionId xid)
 
void SnapBuildCommitTxn (SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
 
void SnapBuildProcessRunningXacts (SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
 
void SnapBuildSerializationPoint (SnapBuild *builder, XLogRecPtr lsn)
 
void CheckPointSnapBuild (void)
 

Variables

static ResourceOwner SavedResourceOwnerDuringExport = NULL
 
static bool ExportInProgress = false
 

Macro Definition Documentation

◆ SNAPBUILD_MAGIC

#define SNAPBUILD_MAGIC   0x51A1E001

Definition at line 1470 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SNAPBUILD_VERSION

#define SNAPBUILD_VERSION   4

Definition at line 1471 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SnapBuildOnDiskConstantSize

#define SnapBuildOnDiskConstantSize   offsetof(SnapBuildOnDisk, builder)

Definition at line 1465 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize   offsetof(SnapBuildOnDisk, version)

Definition at line 1467 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

Typedef Documentation

◆ SnapBuildOnDisk

Function Documentation

◆ AllocateSnapshotBuilder()

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

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

Referenced by StartupDecodingContext().

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 }
#define AllocSetContextCreate
Definition: memutils.h:173
uint32 TransactionId
Definition: c.h:587
bool building_full_snapshot
Definition: snapbuild.c:185
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
size_t xcnt_space
Definition: snapbuild.c:219
size_t xcnt
Definition: snapbuild.c:216
SnapBuildState state
Definition: snapbuild.c:150
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:195
ReorderBuffer * reorder
Definition: snapbuild.c:200
TransactionId initial_xmin_horizon
Definition: snapbuild.c:182
TransactionId * xip
Definition: snapbuild.c:242
bool includes_all_transactions
Definition: snapbuild.c:226
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
XLogRecPtr two_phase_at
Definition: snapbuild.c:176
void * palloc0(Size size)
Definition: mcxt.c:1093
MemoryContext context
Definition: snapbuild.c:153
struct SnapBuild::@16 committed
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 1922 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().

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

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild builder)

Definition at line 323 of file snapbuild.c.

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

Referenced by FreeDecodingContext().

324 {
325  MemoryContext context = builder->context;
326 
327  /* free snapshot explicitly, that contains some error checking */
328  if (builder->snapshot != NULL)
329  {
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
Snapshot snapshot
Definition: snapbuild.c:190
MemoryContext context
Definition: snapbuild.c:153
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:418

◆ SnapBuildAddCommittedTxn()

static void SnapBuildAddCommittedTxn ( SnapBuild builder,
TransactionId  xid 
)
static

Definition at line 867 of file snapbuild.c.

References Assert, SnapBuild::committed, DEBUG1, elog, repalloc(), TransactionIdIsValid, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by SnapBuildCommitTxn().

868 {
870 
871  if (builder->committed.xcnt == builder->committed.xcnt_space)
872  {
873  builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
874 
875  elog(DEBUG1, "increasing space for committed transactions to %u",
876  (uint32) builder->committed.xcnt_space);
877 
878  builder->committed.xip = repalloc(builder->committed.xip,
879  builder->committed.xcnt_space * sizeof(TransactionId));
880  }
881 
882  /*
883  * TODO: It might make sense to keep the array sorted here instead of
884  * doing it every time we build a new snapshot. On the other hand this
885  * gets called repeatedly when a transaction with subtransactions commits.
886  */
887  builder->committed.xip[builder->committed.xcnt++] = xid;
888 }
#define DEBUG1
Definition: elog.h:25
uint32 TransactionId
Definition: c.h:587
size_t xcnt_space
Definition: snapbuild.c:219
size_t xcnt
Definition: snapbuild.c:216
TransactionId * xip
Definition: snapbuild.c:242
unsigned int uint32
Definition: c.h:441
#define Assert(condition)
Definition: c.h:804
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1182
struct SnapBuild::@16 committed
#define elog(elevel,...)
Definition: elog.h:232
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ SnapBuildBuildSnapshot()

static Snapshot SnapBuildBuildSnapshot ( SnapBuild builder)
static

Definition at line 450 of file snapbuild.c.

References SnapshotData::active_count, Assert, SnapBuild::committed, SnapBuild::context, SnapshotData::copied, SnapshotData::curcid, FirstCommandId, MemoryContextAllocZero(), qsort, SnapshotData::regd_count, SNAPBUILD_FULL_SNAPSHOT, SnapBuild::snapshot, SNAPSHOT_HISTORIC_MVCC, SnapshotData::snapshot_type, SnapshotData::snapXactCompletionCount, SnapBuild::state, SnapshotData::suboverflowed, SnapshotData::subxcnt, SnapshotData::subxip, SnapshotData::takenDuringRecovery, TransactionIdIsNormal, SnapshotData::xcnt, SnapBuild::xcnt, xidComparator(), SnapshotData::xip, SnapBuild::xip, SnapshotData::xmax, SnapBuild::xmax, SnapBuild::xmin, and SnapshotData::xmin.

Referenced by SnapBuildCommitTxn(), SnapBuildGetOrBuildSnapshot(), SnapBuildInitialSnapshot(), SnapBuildProcessChange(), and SnapBuildRestore().

451 {
452  Snapshot snapshot;
453  Size ssize;
454 
455  Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
456 
457  ssize = sizeof(SnapshotData)
458  + sizeof(TransactionId) * builder->committed.xcnt
459  + sizeof(TransactionId) * 1 /* toplevel xid */ ;
460 
461  snapshot = MemoryContextAllocZero(builder->context, ssize);
462 
464 
465  /*
466  * We misuse the original meaning of SnapshotData's xip and subxip fields
467  * to make the more fitting for our needs.
468  *
469  * In the 'xip' array we store transactions that have to be treated as
470  * committed. Since we will only ever look at tuples from transactions
471  * that have modified the catalog it's more efficient to store those few
472  * that exist between xmin and xmax (frequently there are none).
473  *
474  * Snapshots that are used in transactions that have modified the catalog
475  * also use the 'subxip' array to store their toplevel xid and all the
476  * subtransaction xids so we can recognize when we need to treat rows as
477  * visible that are not in xip but still need to be visible. Subxip only
478  * gets filled when the transaction is copied into the context of a
479  * catalog modifying transaction since we otherwise share a snapshot
480  * between transactions. As long as a txn hasn't modified the catalog it
481  * doesn't need to treat any uncommitted rows as visible, so there is no
482  * need for those xids.
483  *
484  * Both arrays are qsort'ed so that we can use bsearch() on them.
485  */
486  Assert(TransactionIdIsNormal(builder->xmin));
487  Assert(TransactionIdIsNormal(builder->xmax));
488 
489  snapshot->xmin = builder->xmin;
490  snapshot->xmax = builder->xmax;
491 
492  /* store all transactions to be treated as committed by this snapshot */
493  snapshot->xip =
494  (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
495  snapshot->xcnt = builder->committed.xcnt;
496  memcpy(snapshot->xip,
497  builder->committed.xip,
498  builder->committed.xcnt * sizeof(TransactionId));
499 
500  /* sort so we can bsearch() */
501  qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
502 
503  /*
504  * Initially, subxip is empty, i.e. it's a snapshot to be used by
505  * transactions that don't modify the catalog. Will be filled by
506  * ReorderBufferCopySnap() if necessary.
507  */
508  snapshot->subxcnt = 0;
509  snapshot->subxip = NULL;
510 
511  snapshot->suboverflowed = false;
512  snapshot->takenDuringRecovery = false;
513  snapshot->copied = false;
514  snapshot->curcid = FirstCommandId;
515  snapshot->active_count = 0;
516  snapshot->regd_count = 0;
517  snapshot->snapXactCompletionCount = 0;
518 
519  return snapshot;
520 }
uint64 snapXactCompletionCount
Definition: snapshot.h:216
uint32 TransactionId
Definition: c.h:587
bool copied
Definition: snapshot.h:185
bool suboverflowed
Definition: snapshot.h:182
size_t xcnt
Definition: snapbuild.c:216
uint32 regd_count
Definition: snapshot.h:205
#define FirstCommandId
Definition: c.h:603
SnapBuildState state
Definition: snapbuild.c:150
TransactionId * xip
Definition: snapbuild.c:242
struct SnapshotData SnapshotData
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
TransactionId xmax
Definition: snapbuild.c:159
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:906
CommandId curcid
Definition: snapshot.h:187
#define Assert(condition)
Definition: c.h:804
bool takenDuringRecovery
Definition: snapshot.h:184
size_t Size
Definition: c.h:540
MemoryContext context
Definition: snapbuild.c:153
struct SnapBuild::@16 committed
uint32 xcnt
Definition: snapshot.h:169
TransactionId xmin
Definition: snapbuild.c:156
#define qsort(a, b, c, d)
Definition: port.h:505
#define TransactionIdIsNormal(xid)
Definition: transam.h:42
TransactionId * subxip
Definition: snapshot.h:180
uint32 active_count
Definition: snapshot.h:204
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:136
int32 subxcnt
Definition: snapshot.h:181

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 683 of file snapbuild.c.

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

Referenced by exec_replication_command().

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 }
void AbortCurrentTransaction(void)
Definition: xact.c:3230
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:250
#define ERROR
Definition: elog.h:46
static bool ExportInProgress
Definition: snapbuild.c:251
bool IsTransactionState(void)
Definition: xact.c:372
#define elog(elevel,...)
Definition: elog.h:232

◆ SnapBuildCommitTxn()

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

Definition at line 937 of file snapbuild.c.

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

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 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:867
uint32 TransactionId
Definition: c.h:587
TransactionId next_phase_at
Definition: snapbuild.c:207
Snapshot snapshot
Definition: snapbuild.c:190
bool building_full_snapshot
Definition: snapbuild.c:185
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:814
#define DEBUG2
Definition: elog.h:24
ReorderBuffer * reorder
Definition: snapbuild.c:200
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:450
bool includes_all_transactions
Definition: snapbuild.c:226
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:804
#define NormalTransactionIdFollows(id1, id2)
Definition: transam.h:152
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:418
struct SnapBuild::@16 committed
#define elog(elevel,...)
Definition: elog.h:232
#define TransactionIdIsValid(xid)
Definition: transam.h:41
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:406
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild builder)

Definition at line 367 of file snapbuild.c.

References SnapBuild::state.

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

368 {
369  return builder->state;
370 }
SnapBuildState state
Definition: snapbuild.c:150

◆ SnapBuildDistributeNewCatalogSnapshot()

static void SnapBuildDistributeNewCatalogSnapshot ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 814 of file snapbuild.c.

References Assert, dlist_iter::cur, DEBUG2, dlist_container, dlist_foreach, elog, LSN_FORMAT_ARGS, rbtxn_prepared, rbtxn_skip_prepared, SnapBuild::reorder, ReorderBufferAddSnapshot(), ReorderBufferXidHasBaseSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, ReorderBuffer::toplevel_by_lsn, TransactionIdIsValid, and ReorderBufferTXN::xid.

Referenced by SnapBuildCommitTxn().

815 {
816  dlist_iter txn_i;
817  ReorderBufferTXN *txn;
818 
819  /*
820  * Iterate through all toplevel transactions. This can include
821  * subtransactions which we just don't yet know to be that, but that's
822  * fine, they will just get an unnecessary snapshot queued.
823  */
824  dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
825  {
826  txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
827 
829 
830  /*
831  * If we don't have a base snapshot yet, there are no changes in this
832  * transaction which in turn implies we don't yet need a snapshot at
833  * all. We'll add a snapshot when the first change gets queued.
834  *
835  * NB: This works correctly even for subtransactions because
836  * ReorderBufferAssignChild() takes care to transfer the base snapshot
837  * to the top-level transaction, and while iterating the changequeue
838  * we'll get the change from the subtxn.
839  */
840  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
841  continue;
842 
843  /*
844  * We don't need to add snapshot to prepared transactions as they
845  * should not see the new catalog contents.
846  */
847  if (rbtxn_prepared(txn) || rbtxn_skip_prepared(txn))
848  continue;
849 
850  elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
851  txn->xid, LSN_FORMAT_ARGS(lsn));
852 
853  /*
854  * increase the snapshot's refcount for the transaction we are handing
855  * it out to
856  */
858  ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
859  builder->snapshot);
860  }
861 }
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
#define rbtxn_prepared(txn)
Snapshot snapshot
Definition: snapbuild.c:190
#define dlist_foreach(iter, lhead)
Definition: ilist.h:526
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define dlist_container(type, membername, ptr)
Definition: ilist.h:496
#define DEBUG2
Definition: elog.h:24
ReorderBuffer * reorder
Definition: snapbuild.c:200
dlist_head toplevel_by_lsn
TransactionId xid
dlist_node * cur
Definition: ilist.h:161
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
#define Assert(condition)
Definition: c.h:804
#define rbtxn_skip_prepared(txn)
#define elog(elevel,...)
Definition: elog.h:232
#define TransactionIdIsValid(xid)
Definition: transam.h:41
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:406

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild builder)

Definition at line 622 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().

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:1019
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
#define XACT_REPEATABLE_READ
Definition: xact.h:38
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4721
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1121
#define LOG
Definition: elog.h:26
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:250
#define ERROR
Definition: elog.h:46
static bool ExportInProgress
Definition: snapbuild.c:251
#define ereport(elevel,...)
Definition: elog.h:157
bool XactReadOnly
Definition: xact.c:79
void StartTransactionCommand(void)
Definition: xact.c:2858
int XactIsoLevel
Definition: xact.c:76
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:530
uint32 xcnt
Definition: snapshot.h:169
#define elog(elevel,...)
Definition: elog.h:232

◆ SnapBuildFindSnapshot()

static bool SnapBuildFindSnapshot ( SnapBuild builder,
XLogRecPtr  lsn,
xl_running_xacts running 
)
static

Definition at line 1206 of file snapbuild.c.

References Assert, SnapBuild::building_full_snapshot, DEBUG1, ereport, errdetail(), errdetail_internal(), errmsg(), errmsg_internal(), SnapBuild::initial_xmin_horizon, InvalidTransactionId, InvalidXLogRecPtr, LOG, LSN_FORMAT_ARGS, SnapBuild::next_phase_at, xl_running_xacts::nextXid, NormalTransactionIdPrecedes, xl_running_xacts::oldestRunningXid, SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildRestore(), SnapBuildWaitSnapshot(), SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdIsNormal, TransactionIdPrecedesOrEquals(), xl_running_xacts::xcnt, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildProcessRunningXacts().

1207 {
1208  /* ---
1209  * Build catalog decoding snapshot incrementally using information about
1210  * the currently running transactions. There are several ways to do that:
1211  *
1212  * a) There were no running transactions when the xl_running_xacts record
1213  * was inserted, jump to CONSISTENT immediately. We might find such a
1214  * state while waiting on c)'s sub-states.
1215  *
1216  * b) This (in a previous run) or another decoding slot serialized a
1217  * snapshot to disk that we can use. Can't use this method for the
1218  * initial snapshot when slot is being created and needs full snapshot
1219  * for export or direct use, as that snapshot will only contain catalog
1220  * modifying transactions.
1221  *
1222  * c) First incrementally build a snapshot for catalog tuples
1223  * (BUILDING_SNAPSHOT), that requires all, already in-progress,
1224  * transactions to finish. Every transaction starting after that
1225  * (FULL_SNAPSHOT state), has enough information to be decoded. But
1226  * for older running transactions no viable snapshot exists yet, so
1227  * CONSISTENT will only be reached once all of those have finished.
1228  * ---
1229  */
1230 
1231  /*
1232  * xl_running_xact record is older than what we can use, we might not have
1233  * all necessary catalog rows anymore.
1234  */
1237  builder->initial_xmin_horizon))
1238  {
1239  ereport(DEBUG1,
1240  (errmsg_internal("skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1241  LSN_FORMAT_ARGS(lsn)),
1242  errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
1243  builder->initial_xmin_horizon, running->oldestRunningXid)));
1244 
1245 
1246  SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
1247 
1248  return true;
1249  }
1250 
1251  /*
1252  * a) No transaction were running, we can jump to consistent.
1253  *
1254  * This is not affected by races around xl_running_xacts, because we can
1255  * miss transaction commits, but currently not transactions starting.
1256  *
1257  * NB: We might have already started to incrementally assemble a snapshot,
1258  * so we need to be careful to deal with that.
1259  */
1260  if (running->oldestRunningXid == running->nextXid)
1261  {
1262  if (builder->start_decoding_at == InvalidXLogRecPtr ||
1263  builder->start_decoding_at <= lsn)
1264  /* can decode everything after this */
1265  builder->start_decoding_at = lsn + 1;
1266 
1267  /* As no transactions were running xmin/xmax can be trivially set. */
1268  builder->xmin = running->nextXid; /* < are finished */
1269  builder->xmax = running->nextXid; /* >= are running */
1270 
1271  /* so we can safely use the faster comparisons */
1272  Assert(TransactionIdIsNormal(builder->xmin));
1273  Assert(TransactionIdIsNormal(builder->xmax));
1274 
1275  builder->state = SNAPBUILD_CONSISTENT;
1277 
1278  ereport(LOG,
1279  (errmsg("logical decoding found consistent point at %X/%X",
1280  LSN_FORMAT_ARGS(lsn)),
1281  errdetail("There are no running transactions.")));
1282 
1283  return false;
1284  }
1285  /* b) valid on disk state and not building full snapshot */
1286  else if (!builder->building_full_snapshot &&
1287  SnapBuildRestore(builder, lsn))
1288  {
1289  /* there won't be any state to cleanup */
1290  return false;
1291  }
1292 
1293  /*
1294  * c) transition from START to BUILDING_SNAPSHOT.
1295  *
1296  * In START state, and a xl_running_xacts record with running xacts is
1297  * encountered. In that case, switch to BUILDING_SNAPSHOT state, and
1298  * record xl_running_xacts->nextXid. Once all running xacts have finished
1299  * (i.e. they're all >= nextXid), we have a complete catalog snapshot. It
1300  * might look that we could use xl_running_xact's ->xids information to
1301  * get there quicker, but that is problematic because transactions marked
1302  * as running, might already have inserted their commit record - it's
1303  * infeasible to change that with locking.
1304  */
1305  else if (builder->state == SNAPBUILD_START)
1306  {
1308  builder->next_phase_at = running->nextXid;
1309 
1310  /*
1311  * Start with an xmin/xmax that's correct for future, when all the
1312  * currently running transactions have finished. We'll update both
1313  * while waiting for the pending transactions to finish.
1314  */
1315  builder->xmin = running->nextXid; /* < are finished */
1316  builder->xmax = running->nextXid; /* >= are running */
1317 
1318  /* so we can safely use the faster comparisons */
1319  Assert(TransactionIdIsNormal(builder->xmin));
1320  Assert(TransactionIdIsNormal(builder->xmax));
1321 
1322  ereport(LOG,
1323  (errmsg("logical decoding found initial starting point at %X/%X",
1324  LSN_FORMAT_ARGS(lsn)),
1325  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1326  running->xcnt, running->nextXid)));
1327 
1328  SnapBuildWaitSnapshot(running, running->nextXid);
1329  }
1330 
1331  /*
1332  * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
1333  *
1334  * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
1335  * is >= than nextXid from when we switched to BUILDING_SNAPSHOT. This
1336  * means all transactions starting afterwards have enough information to
1337  * be decoded. Switch to FULL_SNAPSHOT.
1338  */
1339  else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
1341  running->oldestRunningXid))
1342  {
1343  builder->state = SNAPBUILD_FULL_SNAPSHOT;
1344  builder->next_phase_at = running->nextXid;
1345 
1346  ereport(LOG,
1347  (errmsg("logical decoding found initial consistent point at %X/%X",
1348  LSN_FORMAT_ARGS(lsn)),
1349  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1350  running->xcnt, running->nextXid)));
1351 
1352  SnapBuildWaitSnapshot(running, running->nextXid);
1353  }
1354 
1355  /*
1356  * c) transition from FULL_SNAPSHOT to CONSISTENT.
1357  *
1358  * In FULL_SNAPSHOT state (see d) ), and this xl_running_xacts'
1359  * oldestRunningXid is >= than nextXid from when we switched to
1360  * FULL_SNAPSHOT. This means all transactions that are currently in
1361  * progress have a catalog snapshot, and all their changes have been
1362  * collected. Switch to CONSISTENT.
1363  */
1364  else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
1366  running->oldestRunningXid))
1367  {
1368  builder->state = SNAPBUILD_CONSISTENT;
1370 
1371  ereport(LOG,
1372  (errmsg("logical decoding found consistent point at %X/%X",
1373  LSN_FORMAT_ARGS(lsn)),
1374  errdetail("There are no old transactions anymore.")));
1375  }
1376 
1377  /*
1378  * We already started to track running xacts and need to wait for all
1379  * in-progress ones to finish. We fall through to the normal processing of
1380  * records so incremental cleanup can be performed.
1381  */
1382  return true;
1383 
1384 }
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1705
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
TransactionId next_phase_at
Definition: snapbuild.c:207
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
Definition: snapbuild.c:1398
bool building_full_snapshot
Definition: snapbuild.c:185
#define LOG
Definition: elog.h:26
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1069
SnapBuildState state
Definition: snapbuild.c:150
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
TransactionId initial_xmin_horizon
Definition: snapbuild.c:182
int errdetail(const char *fmt,...)
Definition: elog.c:1042
#define InvalidTransactionId
Definition: transam.h:31
TransactionId xmax
Definition: snapbuild.c:159
#define ereport(elevel,...)
Definition: elog.h:157
int errmsg_internal(const char *fmt,...)
Definition: elog.c:996
#define Assert(condition)
Definition: c.h:804
TransactionId nextXid
Definition: standbydefs.h:52
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
int errmsg(const char *fmt,...)
Definition: elog.c:909
TransactionId xmin
Definition: snapbuild.c:156
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define TransactionIdIsNormal(xid)
Definition: transam.h:42
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

◆ SnapBuildFreeSnapshot()

static void SnapBuildFreeSnapshot ( Snapshot  snap)
static

Definition at line 342 of file snapbuild.c.

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

Referenced by SnapBuildSnapDecRefcount().

343 {
344  /* make sure we don't get passed an external snapshot */
346 
347  /* make sure nobody modified our snapshot */
348  Assert(snap->curcid == FirstCommandId);
349  Assert(!snap->suboverflowed);
350  Assert(!snap->takenDuringRecovery);
351  Assert(snap->regd_count == 0);
352 
353  /* slightly more likely, so it's checked even without c-asserts */
354  if (snap->copied)
355  elog(ERROR, "cannot free a copied snapshot");
356 
357  if (snap->active_count)
358  elog(ERROR, "cannot free an active snapshot");
359 
360  pfree(snap);
361 }
bool copied
Definition: snapshot.h:185
bool suboverflowed
Definition: snapshot.h:182
uint32 regd_count
Definition: snapshot.h:205
#define FirstCommandId
Definition: c.h:603
void pfree(void *pointer)
Definition: mcxt.c:1169
#define ERROR
Definition: elog.h:46
SnapshotType snapshot_type
Definition: snapshot.h:144
CommandId curcid
Definition: snapshot.h:187
#define Assert(condition)
Definition: c.h:804
bool takenDuringRecovery
Definition: snapshot.h:184
#define elog(elevel,...)
Definition: elog.h:232
uint32 active_count
Definition: snapshot.h:204

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder,
TransactionId  xid 
)

Definition at line 662 of file snapbuild.c.

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

Referenced by DecodeLogicalMsgOp().

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 }
Snapshot snapshot
Definition: snapbuild.c:190
SnapBuildState state
Definition: snapbuild.c:150
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:450
#define Assert(condition)
Definition: c.h:804
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:406

◆ SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt ( SnapBuild builder)

Definition at line 376 of file snapbuild.c.

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

377 {
378  return builder->two_phase_at;
379 }
XLogRecPtr two_phase_at
Definition: snapbuild.c:176

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 530 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().

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,
596  (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
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 }
#define TransactionIdAdvance(dest)
Definition: transam.h:91
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2947
static void test(void)
uint32 TransactionId
Definition: c.h:587
PGPROC * MyProc
Definition: proc.c:68
#define XACT_REPEATABLE_READ
Definition: xact.h:38
int errcode(int sqlerrcode)
Definition: elog.c:698
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
SnapBuildState state
Definition: snapbuild.c:150
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
#define ERROR
Definition: elog.h:46
bool FirstSnapshotSet
Definition: snapmgr.c:149
TransactionId xmin
Definition: proc.h:138
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:450
bool includes_all_transactions
Definition: snapbuild.c:226
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:157
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:2069
#define Assert(condition)
Definition: c.h:804
int XactIsoLevel
Definition: xact.c:76
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1199
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
struct SnapBuild::@16 committed
uint32 xcnt
Definition: snapshot.h:169
void * palloc(Size size)
Definition: mcxt.c:1062
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define elog(elevel,...)
Definition: elog.h:232
#define TransactionIdIsValid(xid)
Definition: transam.h:41
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:136

◆ SnapBuildProcessChange()

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

Definition at line 722 of file snapbuild.c.

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

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

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 }
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
TransactionId next_phase_at
Definition: snapbuild.c:207
Snapshot snapshot
Definition: snapbuild.c:190
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
SnapBuildState state
Definition: snapbuild.c:150
ReorderBuffer * reorder
Definition: snapbuild.c:200
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:450
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:406

◆ SnapBuildProcessNewCid()

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

Definition at line 772 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().

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 }
uint32 CommandId
Definition: c.h:601
CommandId combocid
Definition: heapam_xlog.h:370
CommandId cmax
Definition: heapam_xlog.h:369
ItemPointerData target_tid
Definition: heapam_xlog.h:376
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileNode node, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
#define ERROR
Definition: elog.h:46
ReorderBuffer * reorder
Definition: snapbuild.c:200
RelFileNode target_node
Definition: heapam_xlog.h:375
#define InvalidCommandId
Definition: c.h:604
#define Max(x, y)
Definition: c.h:980
#define elog(elevel,...)
Definition: elog.h:232
CommandId cmin
Definition: heapam_xlog.h:368
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TransactionId top_xid
Definition: heapam_xlog.h:367
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)

◆ SnapBuildProcessRunningXacts()

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

Definition at line 1104 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().

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 InvalidXLogRecPtr
Definition: xlogdefs.h:28
uint32 TransactionId
Definition: c.h:587
#define DEBUG3
Definition: elog.h:23
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1632
XLogRecPtr current_restart_decoding_lsn
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1206
SnapBuildState state
Definition: snapbuild.c:150
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1493
ReorderBuffer * reorder
Definition: snapbuild.c:200
#define InvalidTransactionId
Definition: transam.h:31
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:195
TransactionId xmax
Definition: snapbuild.c:159
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1564
TransactionId xmin
Definition: snapbuild.c:156
#define elog(elevel,...)
Definition: elog.h:232
XLogRecPtr restart_decoding_lsn
TransactionId oldestRunningXid
Definition: standbydefs.h:53
static void SnapBuildPurgeCommittedTxn(SnapBuild *builder)
Definition: snapbuild.c:896

◆ SnapBuildPurgeCommittedTxn()

static void SnapBuildPurgeCommittedTxn ( SnapBuild builder)
static

Definition at line 896 of file snapbuild.c.

References SnapBuild::committed, SnapBuild::context, DEBUG3, elog, MemoryContextAlloc(), NormalTransactionIdPrecedes, pfree(), TransactionIdIsNormal, SnapBuild::xcnt, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildProcessRunningXacts().

897 {
898  int off;
899  TransactionId *workspace;
900  int surviving_xids = 0;
901 
902  /* not ready yet */
903  if (!TransactionIdIsNormal(builder->xmin))
904  return;
905 
906  /* TODO: Neater algorithm than just copying and iterating? */
907  workspace =
908  MemoryContextAlloc(builder->context,
909  builder->committed.xcnt * sizeof(TransactionId));
910 
911  /* copy xids that still are interesting to workspace */
912  for (off = 0; off < builder->committed.xcnt; off++)
913  {
914  if (NormalTransactionIdPrecedes(builder->committed.xip[off],
915  builder->xmin))
916  ; /* remove */
917  else
918  workspace[surviving_xids++] = builder->committed.xip[off];
919  }
920 
921  /* copy workspace back to persistent state */
922  memcpy(builder->committed.xip, workspace,
923  surviving_xids * sizeof(TransactionId));
924 
925  elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
926  (uint32) builder->committed.xcnt, (uint32) surviving_xids,
927  builder->xmin, builder->xmax);
928  builder->committed.xcnt = surviving_xids;
929 
930  pfree(workspace);
931 }
uint32 TransactionId
Definition: c.h:587
#define DEBUG3
Definition: elog.h:23
size_t xcnt
Definition: snapbuild.c:216
void pfree(void *pointer)
Definition: mcxt.c:1169
TransactionId * xip
Definition: snapbuild.c:242
unsigned int uint32
Definition: c.h:441
TransactionId xmax
Definition: snapbuild.c:159
MemoryContext context
Definition: snapbuild.c:153
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
struct SnapBuild::@16 committed
TransactionId xmin
Definition: snapbuild.c:156
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:863
#define elog(elevel,...)
Definition: elog.h:232
#define TransactionIdIsNormal(xid)
Definition: transam.h:42

◆ SnapBuildResetExportedSnapshotState()

void SnapBuildResetExportedSnapshotState ( void  )

Definition at line 710 of file snapbuild.c.

References ExportInProgress.

Referenced by AbortTransaction().

711 {
713  ExportInProgress = false;
714 }
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:250
static bool ExportInProgress
Definition: snapbuild.c:251

◆ SnapBuildRestore()

static bool SnapBuildRestore ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1705 of file snapbuild.c.

References Assert, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, SnapBuild::context, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errdetail(), errmsg(), ERROR, fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, SnapBuild::initial_xmin_horizon, InvalidTransactionId, LOG, LSN_FORMAT_ARGS, SnapBuildOnDisk::magic, MAXPGPATH, MemoryContextAllocZero(), OpenTransientFile(), pfree(), PG_BINARY, pgstat_report_wait_end(), pgstat_report_wait_start(), read, SnapBuild::reorder, ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildBuildSnapshot(), SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, sprintf, SnapBuild::state, TransactionIdPrecedes(), SnapBuildOnDisk::version, WAIT_EVENT_SNAPBUILD_READ, SnapBuild::xcnt, SnapBuild::xcnt_space, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildFindSnapshot(), and SnapBuildSerializationPoint().

1706 {
1707  SnapBuildOnDisk ondisk;
1708  int fd;
1709  char path[MAXPGPATH];
1710  Size sz;
1711  int readBytes;
1712  pg_crc32c checksum;
1713 
1714  /* no point in loading a snapshot if we're already there */
1715  if (builder->state == SNAPBUILD_CONSISTENT)
1716  return false;
1717 
1718  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1719  LSN_FORMAT_ARGS(lsn));
1720 
1721  fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1722 
1723  if (fd < 0 && errno == ENOENT)
1724  return false;
1725  else if (fd < 0)
1726  ereport(ERROR,
1728  errmsg("could not open file \"%s\": %m", path)));
1729 
1730  /* ----
1731  * Make sure the snapshot had been stored safely to disk, that's normally
1732  * cheap.
1733  * Note that we do not need PANIC here, nobody will be able to use the
1734  * slot without fsyncing, and saving it won't succeed without an fsync()
1735  * either...
1736  * ----
1737  */
1738  fsync_fname(path, false);
1739  fsync_fname("pg_logical/snapshots", true);
1740 
1741 
1742  /* read statically sized portion of snapshot */
1744  readBytes = read(fd, &ondisk, SnapBuildOnDiskConstantSize);
1746  if (readBytes != SnapBuildOnDiskConstantSize)
1747  {
1748  int save_errno = errno;
1749 
1750  CloseTransientFile(fd);
1751 
1752  if (readBytes < 0)
1753  {
1754  errno = save_errno;
1755  ereport(ERROR,
1757  errmsg("could not read file \"%s\": %m", path)));
1758  }
1759  else
1760  ereport(ERROR,
1762  errmsg("could not read file \"%s\": read %d of %zu",
1763  path, readBytes,
1765  }
1766 
1767  if (ondisk.magic != SNAPBUILD_MAGIC)
1768  ereport(ERROR,
1770  errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1771  path, ondisk.magic, SNAPBUILD_MAGIC)));
1772 
1773  if (ondisk.version != SNAPBUILD_VERSION)
1774  ereport(ERROR,
1776  errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1777  path, ondisk.version, SNAPBUILD_VERSION)));
1778 
1779  INIT_CRC32C(checksum);
1780  COMP_CRC32C(checksum,
1781  ((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
1783 
1784  /* read SnapBuild */
1786  readBytes = read(fd, &ondisk.builder, sizeof(SnapBuild));
1788  if (readBytes != sizeof(SnapBuild))
1789  {
1790  int save_errno = errno;
1791 
1792  CloseTransientFile(fd);
1793 
1794  if (readBytes < 0)
1795  {
1796  errno = save_errno;
1797  ereport(ERROR,
1799  errmsg("could not read file \"%s\": %m", path)));
1800  }
1801  else
1802  ereport(ERROR,
1804  errmsg("could not read file \"%s\": read %d of %zu",
1805  path, readBytes, sizeof(SnapBuild))));
1806  }
1807  COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
1808 
1809  /* restore committed xacts information */
1810  sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
1811  ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
1813  readBytes = read(fd, ondisk.builder.committed.xip, sz);
1815  if (readBytes != sz)
1816  {
1817  int save_errno = errno;
1818 
1819  CloseTransientFile(fd);
1820 
1821  if (readBytes < 0)
1822  {
1823  errno = save_errno;
1824  ereport(ERROR,
1826  errmsg("could not read file \"%s\": %m", path)));
1827  }
1828  else
1829  ereport(ERROR,
1831  errmsg("could not read file \"%s\": read %d of %zu",
1832  path, readBytes, sz)));
1833  }
1834  COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
1835 
1836  if (CloseTransientFile(fd) != 0)
1837  ereport(ERROR,
1839  errmsg("could not close file \"%s\": %m", path)));
1840 
1841  FIN_CRC32C(checksum);
1842 
1843  /* verify checksum of what we've read */
1844  if (!EQ_CRC32C(checksum, ondisk.checksum))
1845  ereport(ERROR,
1847  errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1848  path, checksum, ondisk.checksum)));
1849 
1850  /*
1851  * ok, we now have a sensible snapshot here, figure out if it has more
1852  * information than we have.
1853  */
1854 
1855  /*
1856  * We are only interested in consistent snapshots for now, comparing
1857  * whether one incomplete snapshot is more "advanced" seems to be
1858  * unnecessarily complex.
1859  */
1860  if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
1861  goto snapshot_not_interesting;
1862 
1863  /*
1864  * Don't use a snapshot that requires an xmin that we cannot guarantee to
1865  * be available.
1866  */
1868  goto snapshot_not_interesting;
1869 
1870  /* consistent snapshots have no next phase */
1872 
1873  /* ok, we think the snapshot is sensible, copy over everything important */
1874  builder->xmin = ondisk.builder.xmin;
1875  builder->xmax = ondisk.builder.xmax;
1876  builder->state = ondisk.builder.state;
1877 
1878  builder->committed.xcnt = ondisk.builder.committed.xcnt;
1879  /* We only allocated/stored xcnt, not xcnt_space xids ! */
1880  /* don't overwrite preallocated xip, if we don't have anything here */
1881  if (builder->committed.xcnt > 0)
1882  {
1883  pfree(builder->committed.xip);
1884  builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1885  builder->committed.xip = ondisk.builder.committed.xip;
1886  }
1887  ondisk.builder.committed.xip = NULL;
1888 
1889  /* our snapshot is not interesting anymore, build a new one */
1890  if (builder->snapshot != NULL)
1891  {
1893  }
1894  builder->snapshot = SnapBuildBuildSnapshot(builder);
1896 
1897  ReorderBufferSetRestartPoint(builder->reorder, lsn);
1898 
1899  Assert(builder->state == SNAPBUILD_CONSISTENT);
1900 
1901  ereport(LOG,
1902  (errmsg("logical decoding found consistent point at %X/%X",
1903  LSN_FORMAT_ARGS(lsn)),
1904  errdetail("Logical decoding will begin using saved snapshot.")));
1905  return true;
1906 
1907 snapshot_not_interesting:
1908  if (ondisk.builder.committed.xip != NULL)
1909  pfree(ondisk.builder.committed.xip);
1910  return false;
1911 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1471
static void pgstat_report_wait_end(void)
Definition: wait_event.h:274
uint32 TransactionId
Definition: c.h:587
TransactionId next_phase_at
Definition: snapbuild.c:207
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1470
pg_crc32c checksum
Definition: snapbuild.c:1450
Snapshot snapshot
Definition: snapbuild.c:190
uint32 pg_crc32c
Definition: pg_crc32c.h:38
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
int errcode(int sqlerrcode)
Definition: elog.c:698
size_t xcnt_space
Definition: snapbuild.c:219
size_t xcnt
Definition: snapbuild.c:216
#define LOG
Definition: elog.h:26
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1271
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define sprintf
Definition: port.h:219
SnapBuildState state
Definition: snapbuild.c:150
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
void pfree(void *pointer)
Definition: mcxt.c:1169
#define ERROR
Definition: elog.h:46
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2509
#define MAXPGPATH
SnapBuild builder
Definition: snapbuild.c:1460
ReorderBuffer * reorder
Definition: snapbuild.c:200
TransactionId initial_xmin_horizon
Definition: snapbuild.c:182
TransactionId * xip
Definition: snapbuild.c:242
int errdetail(const char *fmt,...)
Definition: elog.c:1042
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:450
int errcode_for_file_access(void)
Definition: elog.c:721
#define InvalidTransactionId
Definition: transam.h:31
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:258
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:47
int CloseTransientFile(int fd)
Definition: fd.c:2686
TransactionId xmax
Definition: snapbuild.c:159
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:906
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1465
#define ereport(elevel,...)
Definition: elog.h:157
#define Assert(condition)
Definition: c.h:804
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1467
size_t Size
Definition: c.h:540
MemoryContext context
Definition: snapbuild.c:153
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:418
struct SnapBuild::@16 committed
int errmsg(const char *fmt,...)
Definition: elog.c:909
TransactionId xmin
Definition: snapbuild.c:156
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:406
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define read(a, b, c)
Definition: win32.h:13

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1480 of file snapbuild.c.

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

Referenced by DecodeXLogOp().

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

◆ SnapBuildSerialize()

static void SnapBuildSerialize ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1493 of file snapbuild.c.

References Assert, SnapBuildOnDisk::checksum, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, SnapBuild::context, DEBUG1, elog, ereport, errcode_for_file_access(), errmsg(), ERROR, fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, InvalidTransactionId, InvalidXLogRecPtr, SnapBuild::last_serialized_snapshot, SnapBuildOnDisk::length, LSN_FORMAT_ARGS, SnapBuildOnDisk::magic, MAXPGPATH, MemoryContextAllocZero(), MyProcPid, SnapBuild::next_phase_at, OpenTransientFile(), pfree(), PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), SnapBuild::reorder, ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, sprintf, stat, SnapBuild::state, SnapBuildOnDisk::version, WAIT_EVENT_SNAPBUILD_SYNC, WAIT_EVENT_SNAPBUILD_WRITE, write, SnapBuild::xcnt, and SnapBuild::xip.

Referenced by SnapBuildProcessRunningXacts(), and SnapBuildSerializationPoint().

1494 {
1495  Size needed_length;
1496  SnapBuildOnDisk *ondisk = NULL;
1497  char *ondisk_c;
1498  int fd;
1499  char tmppath[MAXPGPATH];
1500  char path[MAXPGPATH];
1501  int ret;
1502  struct stat stat_buf;
1503  Size sz;
1504 
1505  Assert(lsn != InvalidXLogRecPtr);
1507  builder->last_serialized_snapshot <= lsn);
1508 
1509  /*
1510  * no point in serializing if we cannot continue to work immediately after
1511  * restoring the snapshot
1512  */
1513  if (builder->state < SNAPBUILD_CONSISTENT)
1514  return;
1515 
1516  /* consistent snapshots have no next phase */
1518 
1519  /*
1520  * We identify snapshots by the LSN they are valid for. We don't need to
1521  * include timelines in the name as each LSN maps to exactly one timeline
1522  * unless the user used pg_resetwal or similar. If a user did so, there's
1523  * no hope continuing to decode anyway.
1524  */
1525  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1526  LSN_FORMAT_ARGS(lsn));
1527 
1528  /*
1529  * first check whether some other backend already has written the snapshot
1530  * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1531  * as a valid state. Everything else is an unexpected error.
1532  */
1533  ret = stat(path, &stat_buf);
1534 
1535  if (ret != 0 && errno != ENOENT)
1536  ereport(ERROR,
1538  errmsg("could not stat file \"%s\": %m", path)));
1539 
1540  else if (ret == 0)
1541  {
1542  /*
1543  * somebody else has already serialized to this point, don't overwrite
1544  * but remember location, so we don't need to read old data again.
1545  *
1546  * To be sure it has been synced to disk after the rename() from the
1547  * tempfile filename to the real filename, we just repeat the fsync.
1548  * That ought to be cheap because in most scenarios it should already
1549  * be safely on disk.
1550  */
1551  fsync_fname(path, false);
1552  fsync_fname("pg_logical/snapshots", true);
1553 
1554  builder->last_serialized_snapshot = lsn;
1555  goto out;
1556  }
1557 
1558  /*
1559  * there is an obvious race condition here between the time we stat(2) the
1560  * file and us writing the file. But we rename the file into place
1561  * atomically and all files created need to contain the same data anyway,
1562  * so this is perfectly fine, although a bit of a resource waste. Locking
1563  * seems like pointless complication.
1564  */
1565  elog(DEBUG1, "serializing snapshot to %s", path);
1566 
1567  /* to make sure only we will write to this tempfile, include pid */
1568  sprintf(tmppath, "pg_logical/snapshots/%X-%X.snap.%d.tmp",
1569  LSN_FORMAT_ARGS(lsn), MyProcPid);
1570 
1571  /*
1572  * Unlink temporary file if it already exists, needs to have been before a
1573  * crash/error since we won't enter this function twice from within a
1574  * single decoding slot/backend and the temporary file contains the pid of
1575  * the current process.
1576  */
1577  if (unlink(tmppath) != 0 && errno != ENOENT)
1578  ereport(ERROR,
1580  errmsg("could not remove file \"%s\": %m", tmppath)));
1581 
1582  needed_length = sizeof(SnapBuildOnDisk) +
1583  sizeof(TransactionId) * builder->committed.xcnt;
1584 
1585  ondisk_c = MemoryContextAllocZero(builder->context, needed_length);
1586  ondisk = (SnapBuildOnDisk *) ondisk_c;
1587  ondisk->magic = SNAPBUILD_MAGIC;
1588  ondisk->version = SNAPBUILD_VERSION;
1589  ondisk->length = needed_length;
1590  INIT_CRC32C(ondisk->checksum);
1591  COMP_CRC32C(ondisk->checksum,
1592  ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1594  ondisk_c += sizeof(SnapBuildOnDisk);
1595 
1596  memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1597  /* NULL-ify memory-only data */
1598  ondisk->builder.context = NULL;
1599  ondisk->builder.snapshot = NULL;
1600  ondisk->builder.reorder = NULL;
1601  ondisk->builder.committed.xip = NULL;
1602 
1603  COMP_CRC32C(ondisk->checksum,
1604  &ondisk->builder,
1605  sizeof(SnapBuild));
1606 
1607  /* copy committed xacts */
1608  sz = sizeof(TransactionId) * builder->committed.xcnt;
1609  memcpy(ondisk_c, builder->committed.xip, sz);
1610  COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1611  ondisk_c += sz;
1612 
1613  FIN_CRC32C(ondisk->checksum);
1614 
1615  /* we have valid data now, open tempfile and write it there */
1616  fd = OpenTransientFile(tmppath,
1617  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1618  if (fd < 0)
1619  ereport(ERROR,
1621  errmsg("could not open file \"%s\": %m", tmppath)));
1622 
1623  errno = 0;
1625  if ((write(fd, ondisk, needed_length)) != needed_length)
1626  {
1627  int save_errno = errno;
1628 
1629  CloseTransientFile(fd);
1630 
1631  /* if write didn't set errno, assume problem is no disk space */
1632  errno = save_errno ? save_errno : ENOSPC;
1633  ereport(ERROR,
1635  errmsg("could not write to file \"%s\": %m", tmppath)));
1636  }
1638 
1639  /*
1640  * fsync the file before renaming so that even if we crash after this we
1641  * have either a fully valid file or nothing.
1642  *
1643  * It's safe to just ERROR on fsync() here because we'll retry the whole
1644  * operation including the writes.
1645  *
1646  * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1647  * some noticeable overhead since it's performed synchronously during
1648  * decoding?
1649  */
1651  if (pg_fsync(fd) != 0)
1652  {
1653  int save_errno = errno;
1654 
1655  CloseTransientFile(fd);
1656  errno = save_errno;
1657  ereport(ERROR,
1659  errmsg("could not fsync file \"%s\": %m", tmppath)));
1660  }
1662 
1663  if (CloseTransientFile(fd) != 0)
1664  ereport(ERROR,
1666  errmsg("could not close file \"%s\": %m", tmppath)));
1667 
1668  fsync_fname("pg_logical/snapshots", true);
1669 
1670  /*
1671  * We may overwrite the work from some other backend, but that's ok, our
1672  * snapshot is valid as well, we'll just have done some superfluous work.
1673  */
1674  if (rename(tmppath, path) != 0)
1675  {
1676  ereport(ERROR,
1678  errmsg("could not rename file \"%s\" to \"%s\": %m",
1679  tmppath, path)));
1680  }
1681 
1682  /* make sure we persist */
1683  fsync_fname(path, false);
1684  fsync_fname("pg_logical/snapshots", true);
1685 
1686  /*
1687  * Now there's no way we can loose the dumped state anymore, remember this
1688  * as a serialization point.
1689  */
1690  builder->last_serialized_snapshot = lsn;
1691 
1692 out:
1694  builder->last_serialized_snapshot);
1695  /* be tidy */
1696  if (ondisk)
1697  pfree(ondisk);
1698 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:43
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1471
static void pgstat_report_wait_end(void)
Definition: wait_event.h:274
uint32 TransactionId
Definition: c.h:587
TransactionId next_phase_at
Definition: snapbuild.c:207
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1470
#define write(a, b, c)
Definition: win32.h:14
pg_crc32c checksum
Definition: snapbuild.c:1450
Snapshot snapshot
Definition: snapbuild.c:190
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:671
size_t xcnt
Definition: snapbuild.c:216
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1271
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define sprintf
Definition: port.h:219
SnapBuildState state
Definition: snapbuild.c:150
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
void pfree(void *pointer)
Definition: mcxt.c:1169
#define ERROR
Definition: elog.h:46
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2509
#define MAXPGPATH
SnapBuild builder
Definition: snapbuild.c:1460
ReorderBuffer * reorder
Definition: snapbuild.c:200
TransactionId * xip
Definition: snapbuild.c:242
int errcode_for_file_access(void)
Definition: elog.c:721
#define InvalidTransactionId
Definition: transam.h:31
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:195
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:258
int CloseTransientFile(int fd)
Definition: fd.c:2686
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:906
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1465
#define ereport(elevel,...)
Definition: elog.h:157
#define Assert(condition)
Definition: c.h:804
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1467
size_t Size
Definition: c.h:540
MemoryContext context
Definition: snapbuild.c:153
struct SnapBuild::@16 committed
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define elog(elevel,...)
Definition: elog.h:232
int pg_fsync(int fd)
Definition: fd.c:357
struct SnapBuildOnDisk SnapBuildOnDisk
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define stat
Definition: win32_port.h:283

◆ SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 385 of file snapbuild.c.

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

386 {
387  builder->two_phase_at = ptr;
388 }
XLogRecPtr two_phase_at
Definition: snapbuild.c:176

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 418 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().

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 }
bool copied
Definition: snapshot.h:185
bool suboverflowed
Definition: snapshot.h:182
uint32 regd_count
Definition: snapshot.h:205
#define FirstCommandId
Definition: c.h:603
#define ERROR
Definition: elog.h:46
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:342
SnapshotType snapshot_type
Definition: snapshot.h:144
CommandId curcid
Definition: snapshot.h:187
#define Assert(condition)
Definition: c.h:804
bool takenDuringRecovery
Definition: snapshot.h:184
#define elog(elevel,...)
Definition: elog.h:232
uint32 active_count
Definition: snapshot.h:204

◆ SnapBuildSnapIncRefcount()

static void SnapBuildSnapIncRefcount ( Snapshot  snap)
static

◆ SnapBuildWaitSnapshot()

static void SnapBuildWaitSnapshot ( xl_running_xacts running,
TransactionId  cutoff 
)
static

Definition at line 1398 of file snapbuild.c.

References elog, ERROR, LogStandbySnapshot(), RecoveryInProgress(), TransactionIdFollows(), TransactionIdIsCurrentTransactionId(), XactLockTableWait(), xl_running_xacts::xcnt, xl_running_xacts::xids, and XLTW_None.

Referenced by SnapBuildFindSnapshot().

1399 {
1400  int off;
1401 
1402  for (off = 0; off < running->xcnt; off++)
1403  {
1404  TransactionId xid = running->xids[off];
1405 
1406  /*
1407  * Upper layers should prevent that we ever need to wait on ourselves.
1408  * Check anyway, since failing to do so would either result in an
1409  * endless wait or an Assert() failure.
1410  */
1412  elog(ERROR, "waiting for ourselves");
1413 
1414  if (TransactionIdFollows(xid, cutoff))
1415  continue;
1416 
1417  XactLockTableWait(xid, NULL, NULL, XLTW_None);
1418  }
1419 
1420  /*
1421  * All transactions we needed to finish finished - try to ensure there is
1422  * another xl_running_xacts record in a timely manner, without having to
1423  * wait for bgwriter or checkpointer to log one. During recovery we can't
1424  * enforce that, so we'll have to wait.
1425  */
1426  if (!RecoveryInProgress())
1427  {
1429  }
1430 }
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
Definition: transam.c:334
uint32 TransactionId
Definition: c.h:587
bool TransactionIdIsCurrentTransactionId(TransactionId xid)
Definition: xact.c:870
bool RecoveryInProgress(void)
Definition: xlog.c:8328
TransactionId xids[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:56
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1220
#define ERROR
Definition: elog.h:46
Definition: lmgr.h:26
void XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid, XLTW_Oper oper)
Definition: lmgr.c:640
#define elog(elevel,...)
Definition: elog.h:232

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 394 of file snapbuild.c.

References SnapBuild::start_decoding_at.

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

395 {
396  return ptr < builder->start_decoding_at;
397 }
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165

Variable Documentation

◆ ExportInProgress

bool ExportInProgress = false
static

◆ SavedResourceOwnerDuringExport

ResourceOwner SavedResourceOwnerDuringExport = NULL
static

Definition at line 250 of file snapbuild.c.

Referenced by SnapBuildClearExportedSnapshot().