PostgreSQL Source Code  git master
snapbuild.c File Reference
#include "postgres.h"
#include <sys/stat.h>
#include <unistd.h>
#include "miscadmin.h"
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/snapshot.h"
#include "utils/snapmgr.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 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   2
 

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)
 
static TransactionId SnapBuildNextPhaseAt (SnapBuild *builder)
 
static void SnapBuildStartNextPhaseAt (SnapBuild *builder, TransactionId at)
 
SnapBuildAllocateSnapshotBuilder (ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot)
 
void FreeSnapshotBuilder (SnapBuild *builder)
 
SnapBuildState SnapBuildCurrentState (SnapBuild *builder)
 
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)
 
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 1462 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SNAPBUILD_VERSION

#define SNAPBUILD_VERSION   2

Definition at line 1463 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SnapBuildOnDiskConstantSize

#define SnapBuildOnDiskConstantSize   offsetof(SnapBuildOnDisk, builder)

Definition at line 1457 of file snapbuild.c.

Referenced by SnapBuildRestore(), and SnapBuildSerialize().

◆ SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize   offsetof(SnapBuildOnDisk, version)

Definition at line 1459 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 
)

Definition at line 315 of file snapbuild.c.

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

Referenced by StartupDecodingContext().

319 {
320  MemoryContext context;
321  MemoryContext oldcontext;
322  SnapBuild *builder;
323 
324  /* allocate memory in own context, to have better accountability */
326  "snapshot builder context",
328  oldcontext = MemoryContextSwitchTo(context);
329 
330  builder = palloc0(sizeof(SnapBuild));
331 
332  builder->state = SNAPBUILD_START;
333  builder->context = context;
334  builder->reorder = reorder;
335  /* Other struct members initialized by zeroing via palloc0 above */
336 
337  builder->committed.xcnt = 0;
338  builder->committed.xcnt_space = 128; /* arbitrary number */
339  builder->committed.xip =
340  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
341  builder->committed.includes_all_transactions = true;
342 
343  builder->initial_xmin_horizon = xmin_horizon;
344  builder->start_decoding_at = start_lsn;
345  builder->building_full_snapshot = need_full_snapshot;
346 
347  MemoryContextSwitchTo(oldcontext);
348 
349  return builder;
350 }
#define AllocSetContextCreate
Definition: memutils.h:170
struct SnapBuild::@24 committed
uint32 TransactionId
Definition: c.h:507
bool building_full_snapshot
Definition: snapbuild.c:179
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
size_t xcnt_space
Definition: snapbuild.c:226
size_t xcnt
Definition: snapbuild.c:223
SnapBuildState state
Definition: snapbuild.c:155
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:192
ReorderBuffer * reorder
Definition: snapbuild.c:194
TransactionId initial_xmin_horizon
Definition: snapbuild.c:176
TransactionId * xip
Definition: snapbuild.c:249
bool includes_all_transactions
Definition: snapbuild.c:233
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
void * palloc0(Size size)
Definition: mcxt.c:980
MemoryContext context
Definition: snapbuild.c:158
XLogRecPtr start_decoding_at
Definition: snapbuild.c:170

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 1937 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.

Referenced by CheckPointGuts().

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

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild builder)

Definition at line 356 of file snapbuild.c.

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

Referenced by FreeDecodingContext().

357 {
358  MemoryContext context = builder->context;
359 
360  /* free snapshot explicitly, that contains some error checking */
361  if (builder->snapshot != NULL)
362  {
364  builder->snapshot = NULL;
365  }
366 
367  /* other resources are deallocated via memory context reset */
368  MemoryContextDelete(context);
369 }
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:211
Snapshot snapshot
Definition: snapbuild.c:184
MemoryContext context
Definition: snapbuild.c:158
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:433

◆ SnapBuildAddCommittedTxn()

static void SnapBuildAddCommittedTxn ( SnapBuild builder,
TransactionId  xid 
)
static

Definition at line 858 of file snapbuild.c.

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

Referenced by SnapBuildCommitTxn().

859 {
861 
862  if (builder->committed.xcnt == builder->committed.xcnt_space)
863  {
864  builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
865 
866  elog(DEBUG1, "increasing space for committed transactions to %u",
867  (uint32) builder->committed.xcnt_space);
868 
869  builder->committed.xip = repalloc(builder->committed.xip,
870  builder->committed.xcnt_space * sizeof(TransactionId));
871  }
872 
873  /*
874  * TODO: It might make sense to keep the array sorted here instead of
875  * doing it every time we build a new snapshot. On the other hand this
876  * gets called repeatedly when a transaction with subtransactions commits.
877  */
878  builder->committed.xip[builder->committed.xcnt++] = xid;
879 }
struct SnapBuild::@24 committed
#define DEBUG1
Definition: elog.h:25
uint32 TransactionId
Definition: c.h:507
size_t xcnt_space
Definition: snapbuild.c:226
size_t xcnt
Definition: snapbuild.c:223
TransactionId * xip
Definition: snapbuild.c:249
unsigned int uint32
Definition: c.h:358
#define Assert(condition)
Definition: c.h:732
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1069
#define elog(elevel,...)
Definition: elog.h:226
#define TransactionIdIsValid(xid)
Definition: transam.h:41

◆ SnapBuildBuildSnapshot()

static Snapshot SnapBuildBuildSnapshot ( SnapBuild builder)
static

Definition at line 465 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, SnapBuild::state, SnapshotData::suboverflowed, SnapshotData::subxcnt, SnapshotData::subxip, SnapshotData::takenDuringRecovery, TransactionIdIsNormal, SnapshotData::xcnt, SnapBuild::xcnt, xidComparator(), SnapshotData::xip, SnapBuild::xip, SnapshotData::xmax, SnapBuild::xmax, SnapshotData::xmin, and SnapBuild::xmin.

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

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

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 697 of file snapbuild.c.

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

Referenced by exec_replication_command().

698 {
699  /* nothing exported, that is the usual case */
700  if (!ExportInProgress)
701  return;
702 
703  if (!IsTransactionState())
704  elog(ERROR, "clearing exported snapshot in wrong transaction state");
705 
706  /* make sure nothing could have ever happened */
708 
711  ExportInProgress = false;
712 }
void AbortCurrentTransaction(void)
Definition: xact.c:3159
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:257
#define ERROR
Definition: elog.h:43
static bool ExportInProgress
Definition: snapbuild.c:258
bool IsTransactionState(void)
Definition: xact.c:356
#define elog(elevel,...)
Definition: elog.h:226

◆ SnapBuildCommitTxn()

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

Definition at line 928 of file snapbuild.c.

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

Referenced by DecodeCommit().

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

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild builder)

Definition at line 400 of file snapbuild.c.

References SnapBuild::state.

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

401 {
402  return builder->state;
403 }
SnapBuildState state
Definition: snapbuild.c:155

◆ SnapBuildDistributeNewCatalogSnapshot()

static void SnapBuildDistributeNewCatalogSnapshot ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 812 of file snapbuild.c.

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

Referenced by SnapBuildCommitTxn().

813 {
814  dlist_iter txn_i;
815  ReorderBufferTXN *txn;
816 
817  /*
818  * Iterate through all toplevel transactions. This can include
819  * subtransactions which we just don't yet know to be that, but that's
820  * fine, they will just get an unnecessary snapshot queued.
821  */
822  dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
823  {
824  txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
825 
827 
828  /*
829  * If we don't have a base snapshot yet, there are no changes in this
830  * transaction which in turn implies we don't yet need a snapshot at
831  * all. We'll add a snapshot when the first change gets queued.
832  *
833  * NB: This works correctly even for subtransactions because
834  * ReorderBufferAssignChild() takes care to transfer the base snapshot
835  * to the top-level transaction, and while iterating the changequeue
836  * we'll get the change from the subtxn.
837  */
838  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
839  continue;
840 
841  elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
842  txn->xid, (uint32) (lsn >> 32), (uint32) lsn);
843 
844  /*
845  * increase the snapshot's refcount for the transaction we are handing
846  * it out to
847  */
849  ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
850  builder->snapshot);
851  }
852 }
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
Snapshot snapshot
Definition: snapbuild.c:184
#define dlist_foreach(iter, lhead)
Definition: ilist.h:507
#define dlist_container(type, membername, ptr)
Definition: ilist.h:477
#define DEBUG2
Definition: elog.h:24
ReorderBuffer * reorder
Definition: snapbuild.c:194
unsigned int uint32
Definition: c.h:358
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:732
#define elog(elevel,...)
Definition: elog.h:226
#define TransactionIdIsValid(xid)
Definition: transam.h:41
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:421

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild builder)

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

637 {
638  Snapshot snap;
639  char *snapname;
640 
642  elog(ERROR, "cannot export a snapshot from within a transaction");
643 
645  elog(ERROR, "can only export one snapshot at a time");
646 
648  ExportInProgress = true;
649 
651 
652  /* There doesn't seem to a nice API to set these */
654  XactReadOnly = true;
655 
656  snap = SnapBuildInitialSnapshot(builder);
657 
658  /*
659  * now that we've built a plain snapshot, make it active and use the
660  * normal mechanisms for exporting it
661  */
662  snapname = ExportSnapshot(snap);
663 
664  ereport(LOG,
665  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
666  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
667  snap->xcnt,
668  snapname, snap->xcnt)));
669  return snapname;
670 }
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:837
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
#define XACT_REPEATABLE_READ
Definition: xact.h:38
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4651
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1191
#define LOG
Definition: elog.h:26
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:257
#define ERROR
Definition: elog.h:43
static bool ExportInProgress
Definition: snapbuild.c:258
#define ereport(elevel, rest)
Definition: elog.h:141
bool XactReadOnly
Definition: xact.c:78
void StartTransactionCommand(void)
Definition: xact.c:2794
int XactIsoLevel
Definition: xact.c:75
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:544
uint32 xcnt
Definition: snapshot.h:169
#define elog(elevel,...)
Definition: elog.h:226

◆ SnapBuildFindSnapshot()

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

Definition at line 1197 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, xl_running_xacts::nextXid, NormalTransactionIdPrecedes, xl_running_xacts::oldestRunningXid, SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildNextPhaseAt(), SnapBuildRestore(), SnapBuildStartNextPhaseAt(), SnapBuildWaitSnapshot(), SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdIsNormal, TransactionIdPrecedesOrEquals(), xl_running_xacts::xcnt, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildProcessRunningXacts().

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

◆ SnapBuildFreeSnapshot()

static void SnapBuildFreeSnapshot ( Snapshot  snap)
static

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

376 {
377  /* make sure we don't get passed an external snapshot */
379 
380  /* make sure nobody modified our snapshot */
381  Assert(snap->curcid == FirstCommandId);
382  Assert(!snap->suboverflowed);
383  Assert(!snap->takenDuringRecovery);
384  Assert(snap->regd_count == 0);
385 
386  /* slightly more likely, so it's checked even without c-asserts */
387  if (snap->copied)
388  elog(ERROR, "cannot free a copied snapshot");
389 
390  if (snap->active_count)
391  elog(ERROR, "cannot free an active snapshot");
392 
393  pfree(snap);
394 }
bool copied
Definition: snapshot.h:185
bool suboverflowed
Definition: snapshot.h:182
uint32 regd_count
Definition: snapshot.h:199
#define FirstCommandId
Definition: c.h:523
void pfree(void *pointer)
Definition: mcxt.c:1056
#define ERROR
Definition: elog.h:43
SnapshotType snapshot_type
Definition: snapshot.h:144
CommandId curcid
Definition: snapshot.h:187
#define Assert(condition)
Definition: c.h:732
bool takenDuringRecovery
Definition: snapshot.h:184
#define elog(elevel,...)
Definition: elog.h:226
uint32 active_count
Definition: snapshot.h:198

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder,
TransactionId  xid 
)

Definition at line 676 of file snapbuild.c.

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

Referenced by DecodeLogicalMsgOp().

677 {
678  Assert(builder->state == SNAPBUILD_CONSISTENT);
679 
680  /* only build a new snapshot if we don't have a prebuilt one */
681  if (builder->snapshot == NULL)
682  {
683  builder->snapshot = SnapBuildBuildSnapshot(builder);
684  /* increase refcount for the snapshot builder */
686  }
687 
688  return builder->snapshot;
689 }
Snapshot snapshot
Definition: snapbuild.c:184
SnapBuildState state
Definition: snapbuild.c:155
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:465
#define Assert(condition)
Definition: c.h:732
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:421

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 544 of file snapbuild.c.

References Assert, SnapBuild::committed, elog, ereport, errcode(), errmsg(), ERROR, FirstSnapshotSet, GetMaxSnapshotXidCount(), GetOldestSafeDecodingTransactionId(), SnapBuild::includes_all_transactions, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyPgXact, 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, SnapshotData::xmin, and PGXACT::xmin.

Referenced by CreateReplicationSlot(), and SnapBuildExportSnapshot().

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

◆ SnapBuildNextPhaseAt()

static TransactionId SnapBuildNextPhaseAt ( SnapBuild builder)
inlinestatic

Definition at line 285 of file snapbuild.c.

References SnapBuild::was_running, and SnapBuild::was_xmax.

Referenced by SnapBuildCommitTxn(), SnapBuildFindSnapshot(), and SnapBuildProcessChange().

286 {
287  /*
288  * For backward compatibility reasons this has to be stored in the wrongly
289  * named field. Will be fixed in next major version.
290  */
291  return builder->was_running.was_xmax;
292 }
struct SnapBuild::@23 was_running
TransactionId was_xmax
Definition: snapbuild.c:209

◆ SnapBuildProcessChange()

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

Definition at line 720 of file snapbuild.c.

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

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

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

◆ SnapBuildProcessNewCid()

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

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

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

◆ SnapBuildProcessRunningXacts()

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

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

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

◆ SnapBuildPurgeCommittedTxn()

static void SnapBuildPurgeCommittedTxn ( SnapBuild builder)
static

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

888 {
889  int off;
890  TransactionId *workspace;
891  int surviving_xids = 0;
892 
893  /* not ready yet */
894  if (!TransactionIdIsNormal(builder->xmin))
895  return;
896 
897  /* TODO: Neater algorithm than just copying and iterating? */
898  workspace =
899  MemoryContextAlloc(builder->context,
900  builder->committed.xcnt * sizeof(TransactionId));
901 
902  /* copy xids that still are interesting to workspace */
903  for (off = 0; off < builder->committed.xcnt; off++)
904  {
905  if (NormalTransactionIdPrecedes(builder->committed.xip[off],
906  builder->xmin))
907  ; /* remove */
908  else
909  workspace[surviving_xids++] = builder->committed.xip[off];
910  }
911 
912  /* copy workspace back to persistent state */
913  memcpy(builder->committed.xip, workspace,
914  surviving_xids * sizeof(TransactionId));
915 
916  elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
917  (uint32) builder->committed.xcnt, (uint32) surviving_xids,
918  builder->xmin, builder->xmax);
919  builder->committed.xcnt = surviving_xids;
920 
921  pfree(workspace);
922 }
struct SnapBuild::@24 committed
uint32 TransactionId
Definition: c.h:507
#define DEBUG3
Definition: elog.h:23
size_t xcnt
Definition: snapbuild.c:223
void pfree(void *pointer)
Definition: mcxt.c:1056
TransactionId * xip
Definition: snapbuild.c:249
unsigned int uint32
Definition: c.h:358
TransactionId xmax
Definition: snapbuild.c:164
MemoryContext context
Definition: snapbuild.c:158
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:98
TransactionId xmin
Definition: snapbuild.c:161
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:796
#define elog(elevel,...)
Definition: elog.h:226
#define TransactionIdIsNormal(xid)
Definition: transam.h:42

◆ SnapBuildRestore()

static bool SnapBuildRestore ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1694 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, LOG, 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().

1695 {
1696  SnapBuildOnDisk ondisk;
1697  int fd;
1698  char path[MAXPGPATH];
1699  Size sz;
1700  int readBytes;
1701  pg_crc32c checksum;
1702 
1703  /* no point in loading a snapshot if we're already there */
1704  if (builder->state == SNAPBUILD_CONSISTENT)
1705  return false;
1706 
1707  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1708  (uint32) (lsn >> 32), (uint32) lsn);
1709 
1710  fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1711 
1712  if (fd < 0 && errno == ENOENT)
1713  return false;
1714  else if (fd < 0)
1715  ereport(ERROR,
1717  errmsg("could not open file \"%s\": %m", path)));
1718 
1719  /* ----
1720  * Make sure the snapshot had been stored safely to disk, that's normally
1721  * cheap.
1722  * Note that we do not need PANIC here, nobody will be able to use the
1723  * slot without fsyncing, and saving it won't succeed without an fsync()
1724  * either...
1725  * ----
1726  */
1727  fsync_fname(path, false);
1728  fsync_fname("pg_logical/snapshots", true);
1729 
1730 
1731  /* read statically sized portion of snapshot */
1733  readBytes = read(fd, &ondisk, SnapBuildOnDiskConstantSize);
1735  if (readBytes != SnapBuildOnDiskConstantSize)
1736  {
1737  int save_errno = errno;
1738 
1739  CloseTransientFile(fd);
1740 
1741  if (readBytes < 0)
1742  {
1743  errno = save_errno;
1744  ereport(ERROR,
1746  errmsg("could not read file \"%s\": %m", path)));
1747  }
1748  else
1749  ereport(ERROR,
1751  errmsg("could not read file \"%s\": read %d of %zu",
1752  path, readBytes,
1754  }
1755 
1756  if (ondisk.magic != SNAPBUILD_MAGIC)
1757  ereport(ERROR,
1759  errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1760  path, ondisk.magic, SNAPBUILD_MAGIC)));
1761 
1762  if (ondisk.version != SNAPBUILD_VERSION)
1763  ereport(ERROR,
1765  errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1766  path, ondisk.version, SNAPBUILD_VERSION)));
1767 
1768  INIT_CRC32C(checksum);
1769  COMP_CRC32C(checksum,
1770  ((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
1772 
1773  /* read SnapBuild */
1775  readBytes = read(fd, &ondisk.builder, sizeof(SnapBuild));
1777  if (readBytes != sizeof(SnapBuild))
1778  {
1779  int save_errno = errno;
1780 
1781  CloseTransientFile(fd);
1782 
1783  if (readBytes < 0)
1784  {
1785  errno = save_errno;
1786  ereport(ERROR,
1788  errmsg("could not read file \"%s\": %m", path)));
1789  }
1790  else
1791  ereport(ERROR,
1793  errmsg("could not read file \"%s\": read %d of %zu",
1794  path, readBytes, sizeof(SnapBuild))));
1795  }
1796  COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
1797 
1798  /* restore running xacts (dead, but kept for backward compat) */
1799  sz = sizeof(TransactionId) * ondisk.builder.was_running.was_xcnt_space;
1800  ondisk.builder.was_running.was_xip =
1801  MemoryContextAllocZero(builder->context, sz);
1803  readBytes = read(fd, ondisk.builder.was_running.was_xip, sz);
1805  if (readBytes != sz)
1806  {
1807  int save_errno = errno;
1808 
1809  CloseTransientFile(fd);
1810 
1811  if (readBytes < 0)
1812  {
1813  errno = save_errno;
1814  ereport(ERROR,
1816  errmsg("could not read file \"%s\": %m", path)));
1817  }
1818  else
1819  ereport(ERROR,
1821  errmsg("could not read file \"%s\": read %d of %zu",
1822  path, readBytes, sz)));
1823  }
1824  COMP_CRC32C(checksum, ondisk.builder.was_running.was_xip, sz);
1825 
1826  /* restore committed xacts information */
1827  sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
1828  ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
1830  readBytes = read(fd, ondisk.builder.committed.xip, sz);
1832  if (readBytes != sz)
1833  {
1834  int save_errno = errno;
1835 
1836  CloseTransientFile(fd);
1837 
1838  if (readBytes < 0)
1839  {
1840  errno = save_errno;
1841  ereport(ERROR,
1843  errmsg("could not read file \"%s\": %m", path)));
1844  }
1845  else
1846  ereport(ERROR,
1848  errmsg("could not read file \"%s\": read %d of %zu",
1849  path, readBytes, sz)));
1850  }
1851  COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
1852 
1853  if (CloseTransientFile(fd) != 0)
1854  ereport(ERROR,
1856  errmsg("could not close file \"%s\": %m", path)));
1857 
1858  FIN_CRC32C(checksum);
1859 
1860  /* verify checksum of what we've read */
1861  if (!EQ_CRC32C(checksum, ondisk.checksum))
1862  ereport(ERROR,
1864  errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1865  path, checksum, ondisk.checksum)));
1866 
1867  /*
1868  * ok, we now have a sensible snapshot here, figure out if it has more
1869  * information than we have.
1870  */
1871 
1872  /*
1873  * We are only interested in consistent snapshots for now, comparing
1874  * whether one incomplete snapshot is more "advanced" seems to be
1875  * unnecessarily complex.
1876  */
1877  if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
1878  goto snapshot_not_interesting;
1879 
1880  /*
1881  * Don't use a snapshot that requires an xmin that we cannot guarantee to
1882  * be available.
1883  */
1885  goto snapshot_not_interesting;
1886 
1887 
1888  /* ok, we think the snapshot is sensible, copy over everything important */
1889  builder->xmin = ondisk.builder.xmin;
1890  builder->xmax = ondisk.builder.xmax;
1891  builder->state = ondisk.builder.state;
1892 
1893  builder->committed.xcnt = ondisk.builder.committed.xcnt;
1894  /* We only allocated/stored xcnt, not xcnt_space xids ! */
1895  /* don't overwrite preallocated xip, if we don't have anything here */
1896  if (builder->committed.xcnt > 0)
1897  {
1898  pfree(builder->committed.xip);
1899  builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1900  builder->committed.xip = ondisk.builder.committed.xip;
1901  }
1902  ondisk.builder.committed.xip = NULL;
1903 
1904  /* our snapshot is not interesting anymore, build a new one */
1905  if (builder->snapshot != NULL)
1906  {
1908  }
1909  builder->snapshot = SnapBuildBuildSnapshot(builder);
1911 
1912  ReorderBufferSetRestartPoint(builder->reorder, lsn);
1913 
1914  Assert(builder->state == SNAPBUILD_CONSISTENT);
1915 
1916  ereport(LOG,
1917  (errmsg("logical decoding found consistent point at %X/%X",
1918  (uint32) (lsn >> 32), (uint32) lsn),
1919  errdetail("Logical decoding will begin using saved snapshot.")));
1920  return true;
1921 
1922 snapshot_not_interesting:
1923  if (ondisk.builder.committed.xip != NULL)
1924  pfree(ondisk.builder.committed.xip);
1925  return false;
1926 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
struct SnapBuild::@24 committed
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1463
uint32 TransactionId
Definition: c.h:507
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1462
pg_crc32c checksum
Definition: snapbuild.c:1442
Snapshot snapshot
Definition: snapbuild.c:184
uint32 pg_crc32c
Definition: pg_crc32c.h:38
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:580
int errcode(int sqlerrcode)
Definition: elog.c:570
size_t xcnt_space
Definition: snapbuild.c:226
size_t xcnt
Definition: snapbuild.c:223
#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:1191
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define sprintf
Definition: port.h:194
SnapBuildState state
Definition: snapbuild.c:155
void pfree(void *pointer)
Definition: mcxt.c:1056
#define ERROR
Definition: elog.h:43
struct SnapBuild::@23 was_running
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2255
TransactionId * was_xip
Definition: snapbuild.c:213
#define MAXPGPATH
SnapBuild builder
Definition: snapbuild.c:1452
ReorderBuffer * reorder
Definition: snapbuild.c:194
TransactionId initial_xmin_horizon
Definition: snapbuild.c:176
TransactionId * xip
Definition: snapbuild.c:249
int errdetail(const char *fmt,...)
Definition: elog.c:860
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:465
int errcode_for_file_access(void)
Definition: elog.c:593
unsigned int uint32
Definition: c.h:358
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1342
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define ereport(elevel, rest)
Definition: elog.h:141
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:45
int CloseTransientFile(int fd)
Definition: fd.c:2432
TransactionId xmax
Definition: snapbuild.c:164
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:839
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1457
#define Assert(condition)
Definition: c.h:732
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1459
size_t Size
Definition: c.h:466
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1318
MemoryContext context
Definition: snapbuild.c:158
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:433
int errmsg(const char *fmt,...)
Definition: elog.c:784
TransactionId xmin
Definition: snapbuild.c:161
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:421
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define read(a, b, c)
Definition: win32.h:13
size_t was_xcnt_space
Definition: snapbuild.c:212

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1472 of file snapbuild.c.

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

Referenced by DecodeXLogOp().

1473 {
1474  if (builder->state < SNAPBUILD_CONSISTENT)
1475  SnapBuildRestore(builder, lsn);
1476  else
1477  SnapBuildSerialize(builder, lsn);
1478 }
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1694
SnapBuildState state
Definition: snapbuild.c:155
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1485

◆ SnapBuildSerialize()

static void SnapBuildSerialize ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1485 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, InvalidXLogRecPtr, SnapBuild::last_serialized_snapshot, SnapBuildOnDisk::length, SnapBuildOnDisk::magic, MAXPGPATH, MemoryContextAllocZero(), MyProcPid, OpenTransientFile(), 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, SnapBuild::was_running, SnapBuild::was_xcnt, write, SnapBuild::xcnt, and SnapBuild::xip.

Referenced by SnapBuildProcessRunningXacts(), and SnapBuildSerializationPoint().

1486 {
1487  Size needed_length;
1488  SnapBuildOnDisk *ondisk;
1489  char *ondisk_c;
1490  int fd;
1491  char tmppath[MAXPGPATH];
1492  char path[MAXPGPATH];
1493  int ret;
1494  struct stat stat_buf;
1495  Size sz;
1496 
1497  Assert(lsn != InvalidXLogRecPtr);
1499  builder->last_serialized_snapshot <= lsn);
1500 
1501  /*
1502  * no point in serializing if we cannot continue to work immediately after
1503  * restoring the snapshot
1504  */
1505  if (builder->state < SNAPBUILD_CONSISTENT)
1506  return;
1507 
1508  /*
1509  * We identify snapshots by the LSN they are valid for. We don't need to
1510  * include timelines in the name as each LSN maps to exactly one timeline
1511  * unless the user used pg_resetwal or similar. If a user did so, there's
1512  * no hope continuing to decode anyway.
1513  */
1514  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1515  (uint32) (lsn >> 32), (uint32) lsn);
1516 
1517  /*
1518  * first check whether some other backend already has written the snapshot
1519  * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1520  * as a valid state. Everything else is an unexpected error.
1521  */
1522  ret = stat(path, &stat_buf);
1523 
1524  if (ret != 0 && errno != ENOENT)
1525  ereport(ERROR,
1527  errmsg("could not stat file \"%s\": %m", path)));
1528 
1529  else if (ret == 0)
1530  {
1531  /*
1532  * somebody else has already serialized to this point, don't overwrite
1533  * but remember location, so we don't need to read old data again.
1534  *
1535  * To be sure it has been synced to disk after the rename() from the
1536  * tempfile filename to the real filename, we just repeat the fsync.
1537  * That ought to be cheap because in most scenarios it should already
1538  * be safely on disk.
1539  */
1540  fsync_fname(path, false);
1541  fsync_fname("pg_logical/snapshots", true);
1542 
1543  builder->last_serialized_snapshot = lsn;
1544  goto out;
1545  }
1546 
1547  /*
1548  * there is an obvious race condition here between the time we stat(2) the
1549  * file and us writing the file. But we rename the file into place
1550  * atomically and all files created need to contain the same data anyway,
1551  * so this is perfectly fine, although a bit of a resource waste. Locking
1552  * seems like pointless complication.
1553  */
1554  elog(DEBUG1, "serializing snapshot to %s", path);
1555 
1556  /* to make sure only we will write to this tempfile, include pid */
1557  sprintf(tmppath, "pg_logical/snapshots/%X-%X.snap.%u.tmp",
1558  (uint32) (lsn >> 32), (uint32) lsn, MyProcPid);
1559 
1560  /*
1561  * Unlink temporary file if it already exists, needs to have been before a
1562  * crash/error since we won't enter this function twice from within a
1563  * single decoding slot/backend and the temporary file contains the pid of
1564  * the current process.
1565  */
1566  if (unlink(tmppath) != 0 && errno != ENOENT)
1567  ereport(ERROR,
1569  errmsg("could not remove file \"%s\": %m", tmppath)));
1570 
1571  needed_length = sizeof(SnapBuildOnDisk) +
1572  sizeof(TransactionId) * builder->committed.xcnt;
1573 
1574  ondisk_c = MemoryContextAllocZero(builder->context, needed_length);
1575  ondisk = (SnapBuildOnDisk *) ondisk_c;
1576  ondisk->magic = SNAPBUILD_MAGIC;
1577  ondisk->version = SNAPBUILD_VERSION;
1578  ondisk->length = needed_length;
1579  INIT_CRC32C(ondisk->checksum);
1580  COMP_CRC32C(ondisk->checksum,
1581  ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1583  ondisk_c += sizeof(SnapBuildOnDisk);
1584 
1585  memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1586  /* NULL-ify memory-only data */
1587  ondisk->builder.context = NULL;
1588  ondisk->builder.snapshot = NULL;
1589  ondisk->builder.reorder = NULL;
1590  ondisk->builder.committed.xip = NULL;
1591 
1592  COMP_CRC32C(ondisk->checksum,
1593  &ondisk->builder,
1594  sizeof(SnapBuild));
1595 
1596  /* there shouldn't be any running xacts */
1597  Assert(builder->was_running.was_xcnt == 0);
1598 
1599  /* copy committed xacts */
1600  sz = sizeof(TransactionId) * builder->committed.xcnt;
1601  memcpy(ondisk_c, builder->committed.xip, sz);
1602  COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1603  ondisk_c += sz;
1604 
1605  FIN_CRC32C(ondisk->checksum);
1606 
1607  /* we have valid data now, open tempfile and write it there */
1608  fd = OpenTransientFile(tmppath,
1609  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1610  if (fd < 0)
1611  ereport(ERROR,
1613  errmsg("could not open file \"%s\": %m", tmppath)));
1614 
1615  errno = 0;
1617  if ((write(fd, ondisk, needed_length)) != needed_length)
1618  {
1619  int save_errno = errno;
1620 
1621  CloseTransientFile(fd);
1622 
1623  /* if write didn't set errno, assume problem is no disk space */
1624  errno = save_errno ? save_errno : ENOSPC;
1625  ereport(ERROR,
1627  errmsg("could not write to file \"%s\": %m", tmppath)));
1628  }
1630 
1631  /*
1632  * fsync the file before renaming so that even if we crash after this we
1633  * have either a fully valid file or nothing.
1634  *
1635  * It's safe to just ERROR on fsync() here because we'll retry the whole
1636  * operation including the writes.
1637  *
1638  * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1639  * some noticeable overhead since it's performed synchronously during
1640  * decoding?
1641  */
1643  if (pg_fsync(fd) != 0)
1644  {
1645  int save_errno = errno;
1646 
1647  CloseTransientFile(fd);
1648  errno = save_errno;
1649  ereport(ERROR,
1651  errmsg("could not fsync file \"%s\": %m", tmppath)));
1652  }
1654 
1655  if (CloseTransientFile(fd) != 0)
1656  ereport(ERROR,
1658  errmsg("could not close file \"%s\": %m", tmppath)));
1659 
1660  fsync_fname("pg_logical/snapshots", true);
1661 
1662  /*
1663  * We may overwrite the work from some other backend, but that's ok, our
1664  * snapshot is valid as well, we'll just have done some superfluous work.
1665  */
1666  if (rename(tmppath, path) != 0)
1667  {
1668  ereport(ERROR,
1670  errmsg("could not rename file \"%s\" to \"%s\": %m",
1671  tmppath, path)));
1672  }
1673 
1674  /* make sure we persist */
1675  fsync_fname(path, false);
1676  fsync_fname("pg_logical/snapshots", true);
1677 
1678  /*
1679  * Now there's no way we can loose the dumped state anymore, remember this
1680  * as a serialization point.
1681  */
1682  builder->last_serialized_snapshot = lsn;
1683 
1684 out:
1686  builder->last_serialized_snapshot);
1687 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
struct SnapBuild::@24 committed
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:40
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1463
uint32 TransactionId
Definition: c.h:507
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1462
#define write(a, b, c)
Definition: win32.h:14
pg_crc32c checksum
Definition: snapbuild.c:1442
Snapshot snapshot
Definition: snapbuild.c:184
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:580
size_t xcnt
Definition: snapbuild.c:223
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1191
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define sprintf
Definition: port.h:194
SnapBuildState state
Definition: snapbuild.c:155
#define ERROR
Definition: elog.h:43
struct stat stat_buf
Definition: pg_standby.c:101
struct SnapBuild::@23 was_running
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2255
size_t was_xcnt
Definition: snapbuild.c:211
#define MAXPGPATH
SnapBuild builder
Definition: snapbuild.c:1452
ReorderBuffer * reorder
Definition: snapbuild.c:194
TransactionId * xip
Definition: snapbuild.c:249
int errcode_for_file_access(void)
Definition: elog.c:593
unsigned int uint32
Definition: c.h:358
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1342
#define ereport(elevel, rest)
Definition: elog.h:141
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:189
int CloseTransientFile(int fd)
Definition: fd.c:2432
#define stat(a, b)
Definition: win32_port.h:255
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:839
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1457
#define Assert(condition)
Definition: c.h:732
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1459
size_t Size
Definition: c.h:466
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1318
MemoryContext context
Definition: snapbuild.c:158
int errmsg(const char *fmt,...)
Definition: elog.c:784
#define elog(elevel,...)
Definition: elog.h:226
int pg_fsync(int fd)
Definition: fd.c:331
struct SnapBuildOnDisk SnapBuildOnDisk
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

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

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

◆ SnapBuildSnapIncRefcount()

static void SnapBuildSnapIncRefcount ( Snapshot  snap)
static

◆ SnapBuildStartNextPhaseAt()

static void SnapBuildStartNextPhaseAt ( SnapBuild builder,
TransactionId  at 
)
inlinestatic

Definition at line 299 of file snapbuild.c.

References SnapBuild::was_running, and SnapBuild::was_xmax.

Referenced by SnapBuildFindSnapshot().

300 {
301  /*
302  * For backward compatibility reasons this has to be stored in the wrongly
303  * named field. Will be fixed in next major version.
304  */
305  builder->was_running.was_xmax = at;
306 }
struct SnapBuild::@23 was_running
TransactionId was_xmax
Definition: snapbuild.c:209

◆ SnapBuildWaitSnapshot()

static void SnapBuildWaitSnapshot ( xl_running_xacts running,
TransactionId  cutoff 
)
static

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

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

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 409 of file snapbuild.c.

References SnapBuild::start_decoding_at.

Referenced by DecodeCommit(), and DecodeLogicalMsgOp().

410 {
411  return ptr < builder->start_decoding_at;
412 }
XLogRecPtr start_decoding_at
Definition: snapbuild.c:170

Variable Documentation

◆ ExportInProgress

bool ExportInProgress = false
static

Definition at line 258 of file snapbuild.c.

Referenced by SnapBuildClearExportedSnapshot(), and SnapBuildExportSnapshot().

◆ SavedResourceOwnerDuringExport

ResourceOwner SavedResourceOwnerDuringExport = NULL
static

Definition at line 257 of file snapbuild.c.

Referenced by SnapBuildClearExportedSnapshot().