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 "common/file_utils.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   5
 

Typedefs

typedef struct SnapBuildOnDisk SnapBuildOnDisk
 

Functions

static void SnapBuildPurgeOlderTxn (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 SnapBuildXidHasCatalogChanges (SnapBuild *builder, TransactionId xid, uint32 xinfo)
 
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 void SnapBuildRestoreContents (int fd, char *dest, Size size, const char *path)
 
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)
 
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, uint32 xinfo)
 
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 1579 of file snapbuild.c.

◆ SNAPBUILD_VERSION

#define SNAPBUILD_VERSION   5

Definition at line 1580 of file snapbuild.c.

◆ SnapBuildOnDiskConstantSize

#define SnapBuildOnDiskConstantSize    offsetof(SnapBuildOnDisk, builder)

Definition at line 1574 of file snapbuild.c.

◆ SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize    offsetof(SnapBuildOnDisk, version)

Definition at line 1576 of file snapbuild.c.

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 312 of file snapbuild.c.

317 {
318  MemoryContext context;
319  MemoryContext oldcontext;
320  SnapBuild *builder;
321 
322  /* allocate memory in own context, to have better accountability */
324  "snapshot builder context",
326  oldcontext = MemoryContextSwitchTo(context);
327 
328  builder = palloc0(sizeof(SnapBuild));
329 
330  builder->state = SNAPBUILD_START;
331  builder->context = context;
332  builder->reorder = reorder;
333  /* Other struct members initialized by zeroing via palloc0 above */
334 
335  builder->committed.xcnt = 0;
336  builder->committed.xcnt_space = 128; /* arbitrary number */
337  builder->committed.xip =
338  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
339  builder->committed.includes_all_transactions = true;
340 
341  builder->catchange.xcnt = 0;
342  builder->catchange.xip = NULL;
343 
344  builder->initial_xmin_horizon = xmin_horizon;
345  builder->start_decoding_at = start_lsn;
346  builder->building_full_snapshot = need_full_snapshot;
347  builder->two_phase_at = two_phase_at;
348 
349  MemoryContextSwitchTo(oldcontext);
350 
351  return builder;
352 }
uint32 TransactionId
Definition: c.h:588
void * palloc0(Size size)
Definition: mcxt.c:1230
MemoryContext CurrentMemoryContext
Definition: mcxt.c:124
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:135
@ SNAPBUILD_START
Definition: snapbuild.h:23
XLogRecPtr start_decoding_at
Definition: snapbuild.c:166
SnapBuildState state
Definition: snapbuild.c:151
TransactionId initial_xmin_horizon
Definition: snapbuild.c:183
struct SnapBuild::@16 committed
TransactionId * xip
Definition: snapbuild.c:243
XLogRecPtr two_phase_at
Definition: snapbuild.c:177
bool building_full_snapshot
Definition: snapbuild.c:186
struct SnapBuild::@17 catchange
size_t xcnt
Definition: snapbuild.c:217
size_t xcnt_space
Definition: snapbuild.c:220
bool includes_all_transactions
Definition: snapbuild.c:227
MemoryContext context
Definition: snapbuild.c:154
ReorderBuffer * reorder
Definition: snapbuild.c:201

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

Referenced by StartupDecodingContext().

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 2047 of file snapbuild.c.

2048 {
2049  XLogRecPtr cutoff;
2050  XLogRecPtr redo;
2051  DIR *snap_dir;
2052  struct dirent *snap_de;
2053  char path[MAXPGPATH + 21];
2054 
2055  /*
2056  * We start off with a minimum of the last redo pointer. No new
2057  * replication slot will start before that, so that's a safe upper bound
2058  * for removal.
2059  */
2060  redo = GetRedoRecPtr();
2061 
2062  /* now check for the restart ptrs from existing slots */
2064 
2065  /* don't start earlier than the restart lsn */
2066  if (redo < cutoff)
2067  cutoff = redo;
2068 
2069  snap_dir = AllocateDir("pg_logical/snapshots");
2070  while ((snap_de = ReadDir(snap_dir, "pg_logical/snapshots")) != NULL)
2071  {
2072  uint32 hi;
2073  uint32 lo;
2074  XLogRecPtr lsn;
2075  PGFileType de_type;
2076 
2077  if (strcmp(snap_de->d_name, ".") == 0 ||
2078  strcmp(snap_de->d_name, "..") == 0)
2079  continue;
2080 
2081  snprintf(path, sizeof(path), "pg_logical/snapshots/%s", snap_de->d_name);
2082  de_type = get_dirent_type(path, snap_de, false, DEBUG1);
2083 
2084  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
2085  {
2086  elog(DEBUG1, "only regular files expected: %s", path);
2087  continue;
2088  }
2089 
2090  /*
2091  * temporary filenames from SnapBuildSerialize() include the LSN and
2092  * everything but are postfixed by .$pid.tmp. We can just remove them
2093  * the same as other files because there can be none that are
2094  * currently being written that are older than cutoff.
2095  *
2096  * We just log a message if a file doesn't fit the pattern, it's
2097  * probably some editors lock/state file or similar...
2098  */
2099  if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
2100  {
2101  ereport(LOG,
2102  (errmsg("could not parse file name \"%s\"", path)));
2103  continue;
2104  }
2105 
2106  lsn = ((uint64) hi) << 32 | lo;
2107 
2108  /* check whether we still need it */
2109  if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
2110  {
2111  elog(DEBUG1, "removing snapbuild snapshot %s", path);
2112 
2113  /*
2114  * It's not particularly harmful, though strange, if we can't
2115  * remove the file here. Don't prevent the checkpoint from
2116  * completing, that'd be a cure worse than the disease.
2117  */
2118  if (unlink(path) < 0)
2119  {
2120  ereport(LOG,
2122  errmsg("could not remove file \"%s\": %m",
2123  path)));
2124  continue;
2125  }
2126  }
2127  }
2128  FreeDir(snap_dir);
2129 }
unsigned int uint32
Definition: c.h:442
int errcode_for_file_access(void)
Definition: elog.c:718
int errmsg(const char *fmt,...)
Definition: elog.c:906
#define LOG
Definition: elog.h:27
#define DEBUG1
Definition: elog.h:26
#define ereport(elevel,...)
Definition: elog.h:145
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2709
int FreeDir(DIR *dir)
Definition: fd.c:2761
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2643
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:406
PGFileType
Definition: file_utils.h:19
@ PGFILETYPE_REG
Definition: file_utils.h:22
@ PGFILETYPE_ERROR
Definition: file_utils.h:20
#define MAXPGPATH
#define snprintf
Definition: port.h:238
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:935
Definition: dirent.c:26
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6015
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References AllocateDir(), dirent::d_name, DEBUG1, elog(), ereport, errcode_for_file_access(), errmsg(), FreeDir(), get_dirent_type(), GetRedoRecPtr(), InvalidXLogRecPtr, LOG, MAXPGPATH, PGFILETYPE_ERROR, PGFILETYPE_REG, ReadDir(), ReplicationSlotsComputeLogicalRestartLSN(), and snprintf.

Referenced by CheckPointGuts().

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild builder)

Definition at line 358 of file snapbuild.c.

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

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

Referenced by FreeDecodingContext().

◆ SnapBuildAddCommittedTxn()

static void SnapBuildAddCommittedTxn ( SnapBuild builder,
TransactionId  xid 
)
static

Definition at line 909 of file snapbuild.c.

910 {
912 
913  if (builder->committed.xcnt == builder->committed.xcnt_space)
914  {
915  builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
916 
917  elog(DEBUG1, "increasing space for committed transactions to %u",
918  (uint32) builder->committed.xcnt_space);
919 
920  builder->committed.xip = repalloc(builder->committed.xip,
921  builder->committed.xcnt_space * sizeof(TransactionId));
922  }
923 
924  /*
925  * TODO: It might make sense to keep the array sorted here instead of
926  * doing it every time we build a new snapshot. On the other hand this
927  * gets called repeatedly when a transaction with subtransactions commits.
928  */
929  builder->committed.xip[builder->committed.xcnt++] = xid;
930 }
Assert(fmt[strlen(fmt) - 1] !='\n')
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1321
#define TransactionIdIsValid(xid)
Definition: transam.h:41

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

Referenced by SnapBuildCommitTxn().

◆ SnapBuildBuildSnapshot()

static Snapshot SnapBuildBuildSnapshot ( SnapBuild builder)
static

Definition at line 485 of file snapbuild.c.

486 {
487  Snapshot snapshot;
488  Size ssize;
489 
490  Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
491 
492  ssize = sizeof(SnapshotData)
493  + sizeof(TransactionId) * builder->committed.xcnt
494  + sizeof(TransactionId) * 1 /* toplevel xid */ ;
495 
496  snapshot = MemoryContextAllocZero(builder->context, ssize);
497 
499 
500  /*
501  * We misuse the original meaning of SnapshotData's xip and subxip fields
502  * to make the more fitting for our needs.
503  *
504  * In the 'xip' array we store transactions that have to be treated as
505  * committed. Since we will only ever look at tuples from transactions
506  * that have modified the catalog it's more efficient to store those few
507  * that exist between xmin and xmax (frequently there are none).
508  *
509  * Snapshots that are used in transactions that have modified the catalog
510  * also use the 'subxip' array to store their toplevel xid and all the
511  * subtransaction xids so we can recognize when we need to treat rows as
512  * visible that are not in xip but still need to be visible. Subxip only
513  * gets filled when the transaction is copied into the context of a
514  * catalog modifying transaction since we otherwise share a snapshot
515  * between transactions. As long as a txn hasn't modified the catalog it
516  * doesn't need to treat any uncommitted rows as visible, so there is no
517  * need for those xids.
518  *
519  * Both arrays are qsort'ed so that we can use bsearch() on them.
520  */
521  Assert(TransactionIdIsNormal(builder->xmin));
522  Assert(TransactionIdIsNormal(builder->xmax));
523 
524  snapshot->xmin = builder->xmin;
525  snapshot->xmax = builder->xmax;
526 
527  /* store all transactions to be treated as committed by this snapshot */
528  snapshot->xip =
529  (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
530  snapshot->xcnt = builder->committed.xcnt;
531  memcpy(snapshot->xip,
532  builder->committed.xip,
533  builder->committed.xcnt * sizeof(TransactionId));
534 
535  /* sort so we can bsearch() */
536  qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
537 
538  /*
539  * Initially, subxip is empty, i.e. it's a snapshot to be used by
540  * transactions that don't modify the catalog. Will be filled by
541  * ReorderBufferCopySnap() if necessary.
542  */
543  snapshot->subxcnt = 0;
544  snapshot->subxip = NULL;
545 
546  snapshot->suboverflowed = false;
547  snapshot->takenDuringRecovery = false;
548  snapshot->copied = false;
549  snapshot->curcid = FirstCommandId;
550  snapshot->active_count = 0;
551  snapshot->regd_count = 0;
552  snapshot->snapXactCompletionCount = 0;
553 
554  return snapshot;
555 }
#define FirstCommandId
Definition: c.h:604
size_t Size
Definition: c.h:541
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:1037
#define qsort(a, b, c, d)
Definition: port.h:445
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:39
struct SnapshotData SnapshotData
@ SNAPSHOT_HISTORIC_MVCC
Definition: snapshot.h:109
TransactionId xmin
Definition: snapbuild.c:157
TransactionId xmax
Definition: snapbuild.c:160
TransactionId xmin
Definition: snapshot.h:157
int32 subxcnt
Definition: snapshot.h:181
bool copied
Definition: snapshot.h:185
uint32 regd_count
Definition: snapshot.h:205
uint32 active_count
Definition: snapshot.h:204
CommandId curcid
Definition: snapshot.h:187
uint32 xcnt
Definition: snapshot.h:169
TransactionId * subxip
Definition: snapshot.h:180
uint64 snapXactCompletionCount
Definition: snapshot.h:216
TransactionId xmax
Definition: snapshot.h:158
SnapshotType snapshot_type
Definition: snapshot.h:144
TransactionId * xip
Definition: snapshot.h:168
bool suboverflowed
Definition: snapshot.h:182
bool takenDuringRecovery
Definition: snapshot.h:184
#define TransactionIdIsNormal(xid)
Definition: transam.h:42
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:136

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

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

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 725 of file snapbuild.c.

726 {
727  ResourceOwner tmpResOwner;
728 
729  /* nothing exported, that is the usual case */
730  if (!ExportInProgress)
731  return;
732 
733  if (!IsTransactionState())
734  elog(ERROR, "clearing exported snapshot in wrong transaction state");
735 
736  /*
737  * AbortCurrentTransaction() takes care of resetting the snapshot state,
738  * so remember SavedResourceOwnerDuringExport.
739  */
740  tmpResOwner = SavedResourceOwnerDuringExport;
741 
742  /* make sure nothing could have ever happened */
744 
745  CurrentResourceOwner = tmpResOwner;
746 }
#define ERROR
Definition: elog.h:35
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:278
static bool ExportInProgress
Definition: snapbuild.c:279
bool IsTransactionState(void)
Definition: xact.c:377
void AbortCurrentTransaction(void)
Definition: xact.c:3293

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

Referenced by exec_replication_command().

◆ SnapBuildCommitTxn()

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

Definition at line 1020 of file snapbuild.c.

1022 {
1023  int nxact;
1024 
1025  bool needs_snapshot = false;
1026  bool needs_timetravel = false;
1027  bool sub_needs_timetravel = false;
1028 
1029  TransactionId xmax = xid;
1030 
1031  /*
1032  * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
1033  * will they be part of a snapshot. So we don't need to record anything.
1034  */
1035  if (builder->state == SNAPBUILD_START ||
1036  (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
1037  TransactionIdPrecedes(xid, builder->next_phase_at)))
1038  {
1039  /* ensure that only commits after this are getting replayed */
1040  if (builder->start_decoding_at <= lsn)
1041  builder->start_decoding_at = lsn + 1;
1042  return;
1043  }
1044 
1045  if (builder->state < SNAPBUILD_CONSISTENT)
1046  {
1047  /* ensure that only commits after this are getting replayed */
1048  if (builder->start_decoding_at <= lsn)
1049  builder->start_decoding_at = lsn + 1;
1050 
1051  /*
1052  * If building an exportable snapshot, force xid to be tracked, even
1053  * if the transaction didn't modify the catalog.
1054  */
1055  if (builder->building_full_snapshot)
1056  {
1057  needs_timetravel = true;
1058  }
1059  }
1060 
1061  for (nxact = 0; nxact < nsubxacts; nxact++)
1062  {
1063  TransactionId subxid = subxacts[nxact];
1064 
1065  /*
1066  * Add subtransaction to base snapshot if catalog modifying, we don't
1067  * distinguish to toplevel transactions there.
1068  */
1069  if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
1070  {
1071  sub_needs_timetravel = true;
1072  needs_snapshot = true;
1073 
1074  elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
1075  xid, subxid);
1076 
1077  SnapBuildAddCommittedTxn(builder, subxid);
1078 
1079  if (NormalTransactionIdFollows(subxid, xmax))
1080  xmax = subxid;
1081  }
1082 
1083  /*
1084  * If we're forcing timetravel we also need visibility information
1085  * about subtransaction, so keep track of subtransaction's state, even
1086  * if not catalog modifying. Don't need to distribute a snapshot in
1087  * that case.
1088  */
1089  else if (needs_timetravel)
1090  {
1091  SnapBuildAddCommittedTxn(builder, subxid);
1092  if (NormalTransactionIdFollows(subxid, xmax))
1093  xmax = subxid;
1094  }
1095  }
1096 
1097  /* if top-level modified catalog, it'll need a snapshot */
1098  if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
1099  {
1100  elog(DEBUG2, "found top level transaction %u, with catalog changes",
1101  xid);
1102  needs_snapshot = true;
1103  needs_timetravel = true;
1104  SnapBuildAddCommittedTxn(builder, xid);
1105  }
1106  else if (sub_needs_timetravel)
1107  {
1108  /* track toplevel txn as well, subxact alone isn't meaningful */
1109  elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
1110  xid);
1111  needs_timetravel = true;
1112  SnapBuildAddCommittedTxn(builder, xid);
1113  }
1114  else if (needs_timetravel)
1115  {
1116  elog(DEBUG2, "forced transaction %u to do timetravel", xid);
1117 
1118  SnapBuildAddCommittedTxn(builder, xid);
1119  }
1120 
1121  if (!needs_timetravel)
1122  {
1123  /* record that we cannot export a general snapshot anymore */
1124  builder->committed.includes_all_transactions = false;
1125  }
1126 
1127  Assert(!needs_snapshot || needs_timetravel);
1128 
1129  /*
1130  * Adjust xmax of the snapshot builder, we only do that for committed,
1131  * catalog modifying, transactions, everything else isn't interesting for
1132  * us since we'll never look at the respective rows.
1133  */
1134  if (needs_timetravel &&
1135  (!TransactionIdIsValid(builder->xmax) ||
1136  TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1137  {
1138  builder->xmax = xmax;
1139  TransactionIdAdvance(builder->xmax);
1140  }
1141 
1142  /* if there's any reason to build a historic snapshot, do so now */
1143  if (needs_snapshot)
1144  {
1145  /*
1146  * If we haven't built a complete snapshot yet there's no need to hand
1147  * it out, it wouldn't (and couldn't) be used anyway.
1148  */
1149  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
1150  return;
1151 
1152  /*
1153  * Decrease the snapshot builder's refcount of the old snapshot, note
1154  * that it still will be used if it has been handed out to the
1155  * reorderbuffer earlier.
1156  */
1157  if (builder->snapshot)
1159 
1160  builder->snapshot = SnapBuildBuildSnapshot(builder);
1161 
1162  /* we might need to execute invalidations, add snapshot */
1163  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1164  {
1166  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1167  builder->snapshot);
1168  }
1169 
1170  /* refcount of the snapshot builder for the new snapshot */
1172 
1173  /* add a new catalog snapshot to all currently running transactions */
1175  }
1176 }
#define DEBUG2
Definition: elog.h:25
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:441
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:909
static bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid, uint32 xinfo)
Definition: snapbuild.c:1183
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:485
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:856
@ SNAPBUILD_BUILDING_SNAPSHOT
Definition: snapbuild.h:29
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46
TransactionId next_phase_at
Definition: snapbuild.c:208
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:273
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:322
#define NormalTransactionIdFollows(id1, id2)
Definition: transam.h:152
#define TransactionIdAdvance(dest)
Definition: transam.h:91

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

Referenced by DecodeCommit().

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild builder)

Definition at line 402 of file snapbuild.c.

403 {
404  return builder->state;
405 }

References SnapBuild::state.

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

◆ SnapBuildDistributeNewCatalogSnapshot()

static void SnapBuildDistributeNewCatalogSnapshot ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 856 of file snapbuild.c.

857 {
858  dlist_iter txn_i;
859  ReorderBufferTXN *txn;
860 
861  /*
862  * Iterate through all toplevel transactions. This can include
863  * subtransactions which we just don't yet know to be that, but that's
864  * fine, they will just get an unnecessary snapshot queued.
865  */
866  dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
867  {
868  txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
869 
871 
872  /*
873  * If we don't have a base snapshot yet, there are no changes in this
874  * transaction which in turn implies we don't yet need a snapshot at
875  * all. We'll add a snapshot when the first change gets queued.
876  *
877  * NB: This works correctly even for subtransactions because
878  * ReorderBufferAssignChild() takes care to transfer the base snapshot
879  * to the top-level transaction, and while iterating the changequeue
880  * we'll get the change from the subtxn.
881  */
882  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
883  continue;
884 
885  /*
886  * We don't need to add snapshot to prepared transactions as they
887  * should not see the new catalog contents.
888  */
889  if (rbtxn_prepared(txn) || rbtxn_skip_prepared(txn))
890  continue;
891 
892  elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
893  txn->xid, LSN_FORMAT_ARGS(lsn));
894 
895  /*
896  * increase the snapshot's refcount for the transaction we are handing
897  * it out to
898  */
900  ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
901  builder->snapshot);
902  }
903 }
#define dlist_foreach(iter, lhead)
Definition: ilist.h:573
#define dlist_container(type, membername, ptr)
Definition: ilist.h:543
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
#define rbtxn_prepared(txn)
#define rbtxn_skip_prepared(txn)
TransactionId xid
dlist_head toplevel_by_lsn
dlist_node * cur
Definition: ilist.h:179
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43

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

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild builder)

Definition at line 664 of file snapbuild.c.

665 {
666  Snapshot snap;
667  char *snapname;
668 
670  elog(ERROR, "cannot export a snapshot from within a transaction");
671 
673  elog(ERROR, "can only export one snapshot at a time");
674 
676  ExportInProgress = true;
677 
679 
680  /* There doesn't seem to a nice API to set these */
682  XactReadOnly = true;
683 
684  snap = SnapBuildInitialSnapshot(builder);
685 
686  /*
687  * now that we've built a plain snapshot, make it active and use the
688  * normal mechanisms for exporting it
689  */
690  snapname = ExportSnapshot(snap);
691 
692  ereport(LOG,
693  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
694  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
695  snap->xcnt,
696  snapname, snap->xcnt)));
697  return snapname;
698 }
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:1016
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:565
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1125
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4814
bool XactReadOnly
Definition: xact.c:81
void StartTransactionCommand(void)
Definition: xact.c:2925
int XactIsoLevel
Definition: xact.c:78
#define XACT_REPEATABLE_READ
Definition: xact.h:38

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

Referenced by CreateReplicationSlot().

◆ SnapBuildFindSnapshot()

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

Definition at line 1315 of file snapbuild.c.

1316 {
1317  /* ---
1318  * Build catalog decoding snapshot incrementally using information about
1319  * the currently running transactions. There are several ways to do that:
1320  *
1321  * a) There were no running transactions when the xl_running_xacts record
1322  * was inserted, jump to CONSISTENT immediately. We might find such a
1323  * state while waiting on c)'s sub-states.
1324  *
1325  * b) This (in a previous run) or another decoding slot serialized a
1326  * snapshot to disk that we can use. Can't use this method for the
1327  * initial snapshot when slot is being created and needs full snapshot
1328  * for export or direct use, as that snapshot will only contain catalog
1329  * modifying transactions.
1330  *
1331  * c) First incrementally build a snapshot for catalog tuples
1332  * (BUILDING_SNAPSHOT), that requires all, already in-progress,
1333  * transactions to finish. Every transaction starting after that
1334  * (FULL_SNAPSHOT state), has enough information to be decoded. But
1335  * for older running transactions no viable snapshot exists yet, so
1336  * CONSISTENT will only be reached once all of those have finished.
1337  * ---
1338  */
1339 
1340  /*
1341  * xl_running_xacts record is older than what we can use, we might not have
1342  * all necessary catalog rows anymore.
1343  */
1346  builder->initial_xmin_horizon))
1347  {
1348  ereport(DEBUG1,
1349  (errmsg_internal("skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1350  LSN_FORMAT_ARGS(lsn)),
1351  errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
1352  builder->initial_xmin_horizon, running->oldestRunningXid)));
1353 
1354 
1355  SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
1356 
1357  return true;
1358  }
1359 
1360  /*
1361  * a) No transaction were running, we can jump to consistent.
1362  *
1363  * This is not affected by races around xl_running_xacts, because we can
1364  * miss transaction commits, but currently not transactions starting.
1365  *
1366  * NB: We might have already started to incrementally assemble a snapshot,
1367  * so we need to be careful to deal with that.
1368  */
1369  if (running->oldestRunningXid == running->nextXid)
1370  {
1371  if (builder->start_decoding_at == InvalidXLogRecPtr ||
1372  builder->start_decoding_at <= lsn)
1373  /* can decode everything after this */
1374  builder->start_decoding_at = lsn + 1;
1375 
1376  /* As no transactions were running xmin/xmax can be trivially set. */
1377  builder->xmin = running->nextXid; /* < are finished */
1378  builder->xmax = running->nextXid; /* >= are running */
1379 
1380  /* so we can safely use the faster comparisons */
1381  Assert(TransactionIdIsNormal(builder->xmin));
1382  Assert(TransactionIdIsNormal(builder->xmax));
1383 
1384  builder->state = SNAPBUILD_CONSISTENT;
1386 
1387  ereport(LOG,
1388  (errmsg("logical decoding found consistent point at %X/%X",
1389  LSN_FORMAT_ARGS(lsn)),
1390  errdetail("There are no running transactions.")));
1391 
1392  return false;
1393  }
1394  /* b) valid on disk state and not building full snapshot */
1395  else if (!builder->building_full_snapshot &&
1396  SnapBuildRestore(builder, lsn))
1397  {
1398  /* there won't be any state to cleanup */
1399  return false;
1400  }
1401 
1402  /*
1403  * c) transition from START to BUILDING_SNAPSHOT.
1404  *
1405  * In START state, and a xl_running_xacts record with running xacts is
1406  * encountered. In that case, switch to BUILDING_SNAPSHOT state, and
1407  * record xl_running_xacts->nextXid. Once all running xacts have finished
1408  * (i.e. they're all >= nextXid), we have a complete catalog snapshot. It
1409  * might look that we could use xl_running_xacts's ->xids information to
1410  * get there quicker, but that is problematic because transactions marked
1411  * as running, might already have inserted their commit record - it's
1412  * infeasible to change that with locking.
1413  */
1414  else if (builder->state == SNAPBUILD_START)
1415  {
1417  builder->next_phase_at = running->nextXid;
1418 
1419  /*
1420  * Start with an xmin/xmax that's correct for future, when all the
1421  * currently running transactions have finished. We'll update both
1422  * while waiting for the pending transactions to finish.
1423  */
1424  builder->xmin = running->nextXid; /* < are finished */
1425  builder->xmax = running->nextXid; /* >= are running */
1426 
1427  /* so we can safely use the faster comparisons */
1428  Assert(TransactionIdIsNormal(builder->xmin));
1429  Assert(TransactionIdIsNormal(builder->xmax));
1430 
1431  ereport(LOG,
1432  (errmsg("logical decoding found initial starting point at %X/%X",
1433  LSN_FORMAT_ARGS(lsn)),
1434  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1435  running->xcnt, running->nextXid)));
1436 
1437  SnapBuildWaitSnapshot(running, running->nextXid);
1438  }
1439 
1440  /*
1441  * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
1442  *
1443  * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
1444  * is >= than nextXid from when we switched to BUILDING_SNAPSHOT. This
1445  * means all transactions starting afterwards have enough information to
1446  * be decoded. Switch to FULL_SNAPSHOT.
1447  */
1448  else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
1450  running->oldestRunningXid))
1451  {
1452  builder->state = SNAPBUILD_FULL_SNAPSHOT;
1453  builder->next_phase_at = running->nextXid;
1454 
1455  ereport(LOG,
1456  (errmsg("logical decoding found initial consistent point at %X/%X",
1457  LSN_FORMAT_ARGS(lsn)),
1458  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1459  running->xcnt, running->nextXid)));
1460 
1461  SnapBuildWaitSnapshot(running, running->nextXid);
1462  }
1463 
1464  /*
1465  * c) transition from FULL_SNAPSHOT to CONSISTENT.
1466  *
1467  * In FULL_SNAPSHOT state (see d) ), and this xl_running_xacts'
1468  * oldestRunningXid is >= than nextXid from when we switched to
1469  * FULL_SNAPSHOT. This means all transactions that are currently in
1470  * progress have a catalog snapshot, and all their changes have been
1471  * collected. Switch to CONSISTENT.
1472  */
1473  else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
1475  running->oldestRunningXid))
1476  {
1477  builder->state = SNAPBUILD_CONSISTENT;
1479 
1480  ereport(LOG,
1481  (errmsg("logical decoding found consistent point at %X/%X",
1482  LSN_FORMAT_ARGS(lsn)),
1483  errdetail("There are no old transactions anymore.")));
1484  }
1485 
1486  /*
1487  * We already started to track running xacts and need to wait for all
1488  * in-progress ones to finish. We fall through to the normal processing of
1489  * records so incremental cleanup can be performed.
1490  */
1491  return true;
1492 }
int errmsg_internal(const char *fmt,...)
Definition: elog.c:993
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1066
int errdetail(const char *fmt,...)
Definition: elog.c:1039
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
Definition: snapbuild.c:1506
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1842
TransactionId oldestRunningXid
Definition: standbydefs.h:53
TransactionId nextXid
Definition: standbydefs.h:52
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:292
#define InvalidTransactionId
Definition: transam.h:31
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147

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

◆ SnapBuildFreeSnapshot()

static void SnapBuildFreeSnapshot ( Snapshot  snap)
static

Definition at line 377 of file snapbuild.c.

378 {
379  /* make sure we don't get passed an external snapshot */
381 
382  /* make sure nobody modified our snapshot */
383  Assert(snap->curcid == FirstCommandId);
384  Assert(!snap->suboverflowed);
385  Assert(!snap->takenDuringRecovery);
386  Assert(snap->regd_count == 0);
387 
388  /* slightly more likely, so it's checked even without c-asserts */
389  if (snap->copied)
390  elog(ERROR, "cannot free a copied snapshot");
391 
392  if (snap->active_count)
393  elog(ERROR, "cannot free an active snapshot");
394 
395  pfree(snap);
396 }
void pfree(void *pointer)
Definition: mcxt.c:1306

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

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder)

Definition at line 704 of file snapbuild.c.

705 {
706  Assert(builder->state == SNAPBUILD_CONSISTENT);
707 
708  /* only build a new snapshot if we don't have a prebuilt one */
709  if (builder->snapshot == NULL)
710  {
711  builder->snapshot = SnapBuildBuildSnapshot(builder);
712  /* increase refcount for the snapshot builder */
714  }
715 
716  return builder->snapshot;
717 }

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

Referenced by logicalmsg_decode().

◆ SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt ( SnapBuild builder)

Definition at line 411 of file snapbuild.c.

412 {
413  return builder->two_phase_at;
414 }

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 565 of file snapbuild.c.

566 {
567  Snapshot snap;
568  TransactionId xid;
569  TransactionId safeXid;
570  TransactionId *newxip;
571  int newxcnt = 0;
572 
574  Assert(builder->building_full_snapshot);
575 
576  /* don't allow older snapshots */
577  InvalidateCatalogSnapshot(); /* about to overwrite MyProc->xmin */
579  elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
581 
582  if (builder->state != SNAPBUILD_CONSISTENT)
583  elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
584 
585  if (!builder->committed.includes_all_transactions)
586  elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
587 
588  /* so we don't overwrite the existing value */
590  elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
591 
592  snap = SnapBuildBuildSnapshot(builder);
593 
594  /*
595  * We know that snap->xmin is alive, enforced by the logical xmin
596  * mechanism. Due to that we can do this without locks, we're only
597  * changing our own value.
598  *
599  * Building an initial snapshot is expensive and an unenforced xmin
600  * horizon would have bad consequences, therefore always double-check that
601  * the horizon is enforced.
602  */
603  LWLockAcquire(ProcArrayLock, LW_SHARED);
604  safeXid = GetOldestSafeDecodingTransactionId(false);
605  LWLockRelease(ProcArrayLock);
606 
607  if (TransactionIdFollows(safeXid, snap->xmin))
608  elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
609  safeXid, snap->xmin);
610 
611  MyProc->xmin = snap->xmin;
612 
613  /* allocate in transaction context */
614  newxip = (TransactionId *)
616 
617  /*
618  * snapbuild.c builds transactions in an "inverted" manner, which means it
619  * stores committed transactions in ->xip, not ones in progress. Build a
620  * classical snapshot by marking all non-committed transactions as
621  * in-progress. This can be expensive.
622  */
623  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
624  {
625  void *test;
626 
627  /*
628  * Check whether transaction committed using the decoding snapshot
629  * meaning of ->xip.
630  */
631  test = bsearch(&xid, snap->xip, snap->xcnt,
632  sizeof(TransactionId), xidComparator);
633 
634  if (test == NULL)
635  {
636  if (newxcnt >= GetMaxSnapshotXidCount())
637  ereport(ERROR,
639  errmsg("initial slot snapshot too large")));
640 
641  newxip[newxcnt++] = xid;
642  }
643 
645  }
646 
647  /* adjust remaining snapshot fields as needed */
649  snap->xcnt = newxcnt;
650  snap->xip = newxip;
651 
652  return snap;
653 }
int errcode(int sqlerrcode)
Definition: elog.c:695
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1194
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1802
@ LW_SHARED
Definition: lwlock.h:113
void * palloc(Size size)
Definition: mcxt.c:1199
#define ERRCODE_T_R_SERIALIZATION_FAILURE
Definition: pgbench.c:75
static void test(void)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2958
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:2066
bool HistoricSnapshotActive(void)
Definition: snapmgr.c:2103
bool HaveRegisteredOrActiveSnapshot(void)
Definition: snapmgr.c:1641
void InvalidateCatalogSnapshot(void)
Definition: snapmgr.c:457
@ SNAPSHOT_MVCC
Definition: snapshot.h:50
PGPROC * MyProc
Definition: proc.c:68
TransactionId xmin
Definition: proc.h:178
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
Definition: transam.c:307

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

Referenced by CreateReplicationSlot(), and SnapBuildExportSnapshot().

◆ SnapBuildProcessChange()

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

Definition at line 764 of file snapbuild.c.

765 {
766  /*
767  * We can't handle data in transactions if we haven't built a snapshot
768  * yet, so don't store them.
769  */
770  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
771  return false;
772 
773  /*
774  * No point in keeping track of changes in transactions that we don't have
775  * enough information about to decode. This means that they started before
776  * we got into the SNAPBUILD_FULL_SNAPSHOT state.
777  */
778  if (builder->state < SNAPBUILD_CONSISTENT &&
779  TransactionIdPrecedes(xid, builder->next_phase_at))
780  return false;
781 
782  /*
783  * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
784  * be needed to decode the change we're currently processing.
785  */
786  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
787  {
788  /* only build a new snapshot if we don't have a prebuilt one */
789  if (builder->snapshot == NULL)
790  {
791  builder->snapshot = SnapBuildBuildSnapshot(builder);
792  /* increase refcount for the snapshot builder */
794  }
795 
796  /*
797  * Increase refcount for the transaction we're handing the snapshot
798  * out to.
799  */
801  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
802  builder->snapshot);
803  }
804 
805  return true;
806 }

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

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

◆ SnapBuildProcessNewCid()

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

Definition at line 814 of file snapbuild.c.

816 {
817  CommandId cid;
818 
819  /*
820  * we only log new_cid's if a catalog tuple was modified, so mark the
821  * transaction as containing catalog modifications
822  */
823  ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
824 
825  ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
826  xlrec->target_locator, xlrec->target_tid,
827  xlrec->cmin, xlrec->cmax,
828  xlrec->combocid);
829 
830  /* figure out new command id */
831  if (xlrec->cmin != InvalidCommandId &&
832  xlrec->cmax != InvalidCommandId)
833  cid = Max(xlrec->cmin, xlrec->cmax);
834  else if (xlrec->cmax != InvalidCommandId)
835  cid = xlrec->cmax;
836  else if (xlrec->cmin != InvalidCommandId)
837  cid = xlrec->cmin;
838  else
839  {
840  cid = InvalidCommandId; /* silence compiler */
841  elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
842  }
843 
844  ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
845 }
#define InvalidCommandId
Definition: c.h:605
#define Max(x, y)
Definition: c.h:931
uint32 CommandId
Definition: c.h:602
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
CommandId cmin
Definition: heapam_xlog.h:375
CommandId combocid
Definition: heapam_xlog.h:377
ItemPointerData target_tid
Definition: heapam_xlog.h:383
TransactionId top_xid
Definition: heapam_xlog.h:374
CommandId cmax
Definition: heapam_xlog.h:376
RelFileLocator target_locator
Definition: heapam_xlog.h:382

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_locator, xl_heap_new_cid::target_tid, and xl_heap_new_cid::top_xid.

Referenced by heap2_decode().

◆ SnapBuildProcessRunningXacts()

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

Definition at line 1213 of file snapbuild.c.

1214 {
1215  ReorderBufferTXN *txn;
1216  TransactionId xmin;
1217 
1218  /*
1219  * If we're not consistent yet, inspect the record to see whether it
1220  * allows to get closer to being consistent. If we are consistent, dump
1221  * our snapshot so others or we, after a restart, can use it.
1222  */
1223  if (builder->state < SNAPBUILD_CONSISTENT)
1224  {
1225  /* returns false if there's no point in performing cleanup just yet */
1226  if (!SnapBuildFindSnapshot(builder, lsn, running))
1227  return;
1228  }
1229  else
1230  SnapBuildSerialize(builder, lsn);
1231 
1232  /*
1233  * Update range of interesting xids based on the running xacts
1234  * information. We don't increase ->xmax using it, because once we are in
1235  * a consistent state we can do that ourselves and much more efficiently
1236  * so, because we only need to do it for catalog transactions since we
1237  * only ever look at those.
1238  *
1239  * NB: We only increase xmax when a catalog modifying transaction commits
1240  * (see SnapBuildCommitTxn). Because of this, xmax can be lower than
1241  * xmin, which looks odd but is correct and actually more efficient, since
1242  * we hit fast paths in heapam_visibility.c.
1243  */
1244  builder->xmin = running->oldestRunningXid;
1245 
1246  /* Remove transactions we don't need to keep track off anymore */
1247  SnapBuildPurgeOlderTxn(builder);
1248 
1249  /*
1250  * Advance the xmin limit for the current replication slot, to allow
1251  * vacuum to clean up the tuples this slot has been protecting.
1252  *
1253  * The reorderbuffer might have an xmin among the currently running
1254  * snapshots; use it if so. If not, we need only consider the snapshots
1255  * we'll produce later, which can't be less than the oldest running xid in
1256  * the record we're reading now.
1257  */
1258  xmin = ReorderBufferGetOldestXmin(builder->reorder);
1259  if (xmin == InvalidTransactionId)
1260  xmin = running->oldestRunningXid;
1261  elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1262  builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
1263  LogicalIncreaseXminForSlot(lsn, xmin);
1264 
1265  /*
1266  * Also tell the slot where we can restart decoding from. We don't want to
1267  * do that after every commit because changing that implies an fsync of
1268  * the logical slot's state file, so we only do it every time we see a
1269  * running xacts record.
1270  *
1271  * Do so by looking for the oldest in progress transaction (determined by
1272  * the first LSN of any of its relevant records). Every transaction
1273  * remembers the last location we stored the snapshot to disk before its
1274  * beginning. That point is where we can restart from.
1275  */
1276 
1277  /*
1278  * Can't know about a serialized snapshot's location if we're not
1279  * consistent.
1280  */
1281  if (builder->state < SNAPBUILD_CONSISTENT)
1282  return;
1283 
1284  txn = ReorderBufferGetOldestTXN(builder->reorder);
1285 
1286  /*
1287  * oldest ongoing txn might have started when we didn't yet serialize
1288  * anything because we hadn't reached a consistent state yet.
1289  */
1290  if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
1292 
1293  /*
1294  * No in-progress transaction, can reuse the last serialized snapshot if
1295  * we have one.
1296  */
1297  else if (txn == NULL &&
1301  builder->last_serialized_snapshot);
1302 }
#define DEBUG3
Definition: elog.h:24
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1662
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1594
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1602
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1315
static void SnapBuildPurgeOlderTxn(SnapBuild *builder)
Definition: snapbuild.c:943
XLogRecPtr restart_decoding_lsn
XLogRecPtr current_restart_decoding_lsn
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:196

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(), SnapBuildPurgeOlderTxn(), SnapBuildSerialize(), SnapBuild::state, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by standby_decode().

◆ SnapBuildPurgeOlderTxn()

static void SnapBuildPurgeOlderTxn ( SnapBuild builder)
static

Definition at line 943 of file snapbuild.c.

944 {
945  int off;
946  TransactionId *workspace;
947  int surviving_xids = 0;
948 
949  /* not ready yet */
950  if (!TransactionIdIsNormal(builder->xmin))
951  return;
952 
953  /* TODO: Neater algorithm than just copying and iterating? */
954  workspace =
955  MemoryContextAlloc(builder->context,
956  builder->committed.xcnt * sizeof(TransactionId));
957 
958  /* copy xids that still are interesting to workspace */
959  for (off = 0; off < builder->committed.xcnt; off++)
960  {
961  if (NormalTransactionIdPrecedes(builder->committed.xip[off],
962  builder->xmin))
963  ; /* remove */
964  else
965  workspace[surviving_xids++] = builder->committed.xip[off];
966  }
967 
968  /* copy workspace back to persistent state */
969  memcpy(builder->committed.xip, workspace,
970  surviving_xids * sizeof(TransactionId));
971 
972  elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
973  (uint32) builder->committed.xcnt, (uint32) surviving_xids,
974  builder->xmin, builder->xmax);
975  builder->committed.xcnt = surviving_xids;
976 
977  pfree(workspace);
978 
979  /*
980  * Purge xids in ->catchange as well. The purged array must also be sorted
981  * in xidComparator order.
982  */
983  if (builder->catchange.xcnt > 0)
984  {
985  /*
986  * Since catchange.xip is sorted, we find the lower bound of xids that
987  * are still interesting.
988  */
989  for (off = 0; off < builder->catchange.xcnt; off++)
990  {
991  if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],
992  builder->xmin))
993  break;
994  }
995 
996  surviving_xids = builder->catchange.xcnt - off;
997 
998  if (surviving_xids > 0)
999  {
1000  memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
1001  surviving_xids * sizeof(TransactionId));
1002  }
1003  else
1004  {
1005  pfree(builder->catchange.xip);
1006  builder->catchange.xip = NULL;
1007  }
1008 
1009  elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
1010  (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
1011  builder->xmin, builder->xmax);
1012  builder->catchange.xcnt = surviving_xids;
1013  }
1014 }
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:994

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

Referenced by SnapBuildProcessRunningXacts().

◆ SnapBuildResetExportedSnapshotState()

void SnapBuildResetExportedSnapshotState ( void  )

Definition at line 752 of file snapbuild.c.

753 {
755  ExportInProgress = false;
756 }

References ExportInProgress, and SavedResourceOwnerDuringExport.

Referenced by AbortTransaction().

◆ SnapBuildRestore()

static bool SnapBuildRestore ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1842 of file snapbuild.c.

1843 {
1844  SnapBuildOnDisk ondisk;
1845  int fd;
1846  char path[MAXPGPATH];
1847  Size sz;
1848  pg_crc32c checksum;
1849 
1850  /* no point in loading a snapshot if we're already there */
1851  if (builder->state == SNAPBUILD_CONSISTENT)
1852  return false;
1853 
1854  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1855  LSN_FORMAT_ARGS(lsn));
1856 
1857  fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1858 
1859  if (fd < 0 && errno == ENOENT)
1860  return false;
1861  else if (fd < 0)
1862  ereport(ERROR,
1864  errmsg("could not open file \"%s\": %m", path)));
1865 
1866  /* ----
1867  * Make sure the snapshot had been stored safely to disk, that's normally
1868  * cheap.
1869  * Note that we do not need PANIC here, nobody will be able to use the
1870  * slot without fsyncing, and saving it won't succeed without an fsync()
1871  * either...
1872  * ----
1873  */
1874  fsync_fname(path, false);
1875  fsync_fname("pg_logical/snapshots", true);
1876 
1877 
1878  /* read statically sized portion of snapshot */
1879  SnapBuildRestoreContents(fd, (char *) &ondisk, SnapBuildOnDiskConstantSize, path);
1880 
1881  if (ondisk.magic != SNAPBUILD_MAGIC)
1882  ereport(ERROR,
1884  errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1885  path, ondisk.magic, SNAPBUILD_MAGIC)));
1886 
1887  if (ondisk.version != SNAPBUILD_VERSION)
1888  ereport(ERROR,
1890  errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1891  path, ondisk.version, SNAPBUILD_VERSION)));
1892 
1893  INIT_CRC32C(checksum);
1894  COMP_CRC32C(checksum,
1895  ((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
1897 
1898  /* read SnapBuild */
1899  SnapBuildRestoreContents(fd, (char *) &ondisk.builder, sizeof(SnapBuild), path);
1900  COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
1901 
1902  /* restore committed xacts information */
1903  if (ondisk.builder.committed.xcnt > 0)
1904  {
1905  sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
1906  ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
1907  SnapBuildRestoreContents(fd, (char *) ondisk.builder.committed.xip, sz, path);
1908  COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
1909  }
1910 
1911  /* restore catalog modifying xacts information */
1912  if (ondisk.builder.catchange.xcnt > 0)
1913  {
1914  sz = sizeof(TransactionId) * ondisk.builder.catchange.xcnt;
1915  ondisk.builder.catchange.xip = MemoryContextAllocZero(builder->context, sz);
1916  SnapBuildRestoreContents(fd, (char *) ondisk.builder.catchange.xip, sz, path);
1917  COMP_CRC32C(checksum, ondisk.builder.catchange.xip, sz);
1918  }
1919 
1920  if (CloseTransientFile(fd) != 0)
1921  ereport(ERROR,
1923  errmsg("could not close file \"%s\": %m", path)));
1924 
1925  FIN_CRC32C(checksum);
1926 
1927  /* verify checksum of what we've read */
1928  if (!EQ_CRC32C(checksum, ondisk.checksum))
1929  ereport(ERROR,
1931  errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1932  path, checksum, ondisk.checksum)));
1933 
1934  /*
1935  * ok, we now have a sensible snapshot here, figure out if it has more
1936  * information than we have.
1937  */
1938 
1939  /*
1940  * We are only interested in consistent snapshots for now, comparing
1941  * whether one incomplete snapshot is more "advanced" seems to be
1942  * unnecessarily complex.
1943  */
1944  if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
1945  goto snapshot_not_interesting;
1946 
1947  /*
1948  * Don't use a snapshot that requires an xmin that we cannot guarantee to
1949  * be available.
1950  */
1952  goto snapshot_not_interesting;
1953 
1954  /* consistent snapshots have no next phase */
1956 
1957  /* ok, we think the snapshot is sensible, copy over everything important */
1958  builder->xmin = ondisk.builder.xmin;
1959  builder->xmax = ondisk.builder.xmax;
1960  builder->state = ondisk.builder.state;
1961 
1962  builder->committed.xcnt = ondisk.builder.committed.xcnt;
1963  /* We only allocated/stored xcnt, not xcnt_space xids ! */
1964  /* don't overwrite preallocated xip, if we don't have anything here */
1965  if (builder->committed.xcnt > 0)
1966  {
1967  pfree(builder->committed.xip);
1968  builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1969  builder->committed.xip = ondisk.builder.committed.xip;
1970  }
1971  ondisk.builder.committed.xip = NULL;
1972 
1973  /* set catalog modifying transactions */
1974  if (builder->catchange.xip)
1975  pfree(builder->catchange.xip);
1976  builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
1977  builder->catchange.xip = ondisk.builder.catchange.xip;
1978  ondisk.builder.catchange.xip = NULL;
1979 
1980  /* our snapshot is not interesting anymore, build a new one */
1981  if (builder->snapshot != NULL)
1982  {
1984  }
1985  builder->snapshot = SnapBuildBuildSnapshot(builder);
1987 
1988  ReorderBufferSetRestartPoint(builder->reorder, lsn);
1989 
1990  Assert(builder->state == SNAPBUILD_CONSISTENT);
1991 
1992  ereport(LOG,
1993  (errmsg("logical decoding found consistent point at %X/%X",
1994  LSN_FORMAT_ARGS(lsn)),
1995  errdetail("Logical decoding will begin using saved snapshot.")));
1996  return true;
1997 
1998 snapshot_not_interesting:
1999  if (ondisk.builder.committed.xip != NULL)
2000  pfree(ondisk.builder.committed.xip);
2001  if (ondisk.builder.catchange.xip != NULL)
2002  pfree(ondisk.builder.catchange.xip);
2003  return false;
2004 }
#define PG_BINARY
Definition: c.h:1209
int CloseTransientFile(int fd)
Definition: fd.c:2609
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:662
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2433
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:41
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define sprintf
Definition: port.h:240
static int fd(const char *x, int i)
Definition: preproc-init.c:105
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1580
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1576
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1579
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1574
static void SnapBuildRestoreContents(int fd, char *dest, Size size, const char *path)
Definition: snapbuild.c:2010
SnapBuild builder
Definition: snapbuild.c:1569
pg_crc32c checksum
Definition: snapbuild.c:1559

References Assert(), SnapBuildOnDisk::builder, SnapBuild::catchange, SnapBuildOnDisk::checksum, 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(), SnapBuild::next_phase_at, OpenTransientFile(), pfree(), PG_BINARY, SnapBuild::reorder, ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildBuildSnapshot(), SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuildRestoreContents(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, sprintf, SnapBuild::state, TransactionIdPrecedes(), SnapBuildOnDisk::version, SnapBuild::xcnt, SnapBuild::xcnt_space, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildFindSnapshot(), and SnapBuildSerializationPoint().

◆ SnapBuildRestoreContents()

static void SnapBuildRestoreContents ( int  fd,
char *  dest,
Size  size,
const char *  path 
)
static

Definition at line 2010 of file snapbuild.c.

2011 {
2012  int readBytes;
2013 
2015  readBytes = read(fd, dest, size);
2017  if (readBytes != size)
2018  {
2019  int save_errno = errno;
2020 
2022 
2023  if (readBytes < 0)
2024  {
2025  errno = save_errno;
2026  ereport(ERROR,
2028  errmsg("could not read file \"%s\": %m", path)));
2029  }
2030  else
2031  ereport(ERROR,
2033  errmsg("could not read file \"%s\": read %d of %zu",
2034  path, readBytes, sizeof(SnapBuild))));
2035  }
2036 }
#define read(a, b, c)
Definition: win32.h:13
@ WAIT_EVENT_SNAPBUILD_READ
Definition: wait_event.h:211
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:268
static void pgstat_report_wait_end(void)
Definition: wait_event.h:284

References CloseTransientFile(), generate_unaccent_rules::dest, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg(), ERROR, fd(), pgstat_report_wait_end(), pgstat_report_wait_start(), read, and WAIT_EVENT_SNAPBUILD_READ.

Referenced by SnapBuildRestore().

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1589 of file snapbuild.c.

1590 {
1591  if (builder->state < SNAPBUILD_CONSISTENT)
1592  SnapBuildRestore(builder, lsn);
1593  else
1594  SnapBuildSerialize(builder, lsn);
1595 }

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

Referenced by xlog_decode().

◆ SnapBuildSerialize()

static void SnapBuildSerialize ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1602 of file snapbuild.c.

1603 {
1604  Size needed_length;
1605  SnapBuildOnDisk *ondisk = NULL;
1606  TransactionId *catchange_xip = NULL;
1607  MemoryContext old_ctx;
1608  size_t catchange_xcnt;
1609  char *ondisk_c;
1610  int fd;
1611  char tmppath[MAXPGPATH];
1612  char path[MAXPGPATH];
1613  int ret;
1614  struct stat stat_buf;
1615  Size sz;
1616 
1617  Assert(lsn != InvalidXLogRecPtr);
1619  builder->last_serialized_snapshot <= lsn);
1620 
1621  /*
1622  * no point in serializing if we cannot continue to work immediately after
1623  * restoring the snapshot
1624  */
1625  if (builder->state < SNAPBUILD_CONSISTENT)
1626  return;
1627 
1628  /* consistent snapshots have no next phase */
1630 
1631  /*
1632  * We identify snapshots by the LSN they are valid for. We don't need to
1633  * include timelines in the name as each LSN maps to exactly one timeline
1634  * unless the user used pg_resetwal or similar. If a user did so, there's
1635  * no hope continuing to decode anyway.
1636  */
1637  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1638  LSN_FORMAT_ARGS(lsn));
1639 
1640  /*
1641  * first check whether some other backend already has written the snapshot
1642  * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1643  * as a valid state. Everything else is an unexpected error.
1644  */
1645  ret = stat(path, &stat_buf);
1646 
1647  if (ret != 0 && errno != ENOENT)
1648  ereport(ERROR,
1650  errmsg("could not stat file \"%s\": %m", path)));
1651 
1652  else if (ret == 0)
1653  {
1654  /*
1655  * somebody else has already serialized to this point, don't overwrite
1656  * but remember location, so we don't need to read old data again.
1657  *
1658  * To be sure it has been synced to disk after the rename() from the
1659  * tempfile filename to the real filename, we just repeat the fsync.
1660  * That ought to be cheap because in most scenarios it should already
1661  * be safely on disk.
1662  */
1663  fsync_fname(path, false);
1664  fsync_fname("pg_logical/snapshots", true);
1665 
1666  builder->last_serialized_snapshot = lsn;
1667  goto out;
1668  }
1669 
1670  /*
1671  * there is an obvious race condition here between the time we stat(2) the
1672  * file and us writing the file. But we rename the file into place
1673  * atomically and all files created need to contain the same data anyway,
1674  * so this is perfectly fine, although a bit of a resource waste. Locking
1675  * seems like pointless complication.
1676  */
1677  elog(DEBUG1, "serializing snapshot to %s", path);
1678 
1679  /* to make sure only we will write to this tempfile, include pid */
1680  sprintf(tmppath, "pg_logical/snapshots/%X-%X.snap.%d.tmp",
1681  LSN_FORMAT_ARGS(lsn), MyProcPid);
1682 
1683  /*
1684  * Unlink temporary file if it already exists, needs to have been before a
1685  * crash/error since we won't enter this function twice from within a
1686  * single decoding slot/backend and the temporary file contains the pid of
1687  * the current process.
1688  */
1689  if (unlink(tmppath) != 0 && errno != ENOENT)
1690  ereport(ERROR,
1692  errmsg("could not remove file \"%s\": %m", tmppath)));
1693 
1694  old_ctx = MemoryContextSwitchTo(builder->context);
1695 
1696  /* Get the catalog modifying transactions that are yet not committed */
1697  catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
1698  catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
1699 
1700  needed_length = sizeof(SnapBuildOnDisk) +
1701  sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
1702 
1703  ondisk_c = palloc0(needed_length);
1704  ondisk = (SnapBuildOnDisk *) ondisk_c;
1705  ondisk->magic = SNAPBUILD_MAGIC;
1706  ondisk->version = SNAPBUILD_VERSION;
1707  ondisk->length = needed_length;
1708  INIT_CRC32C(ondisk->checksum);
1709  COMP_CRC32C(ondisk->checksum,
1710  ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1712  ondisk_c += sizeof(SnapBuildOnDisk);
1713 
1714  memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1715  /* NULL-ify memory-only data */
1716  ondisk->builder.context = NULL;
1717  ondisk->builder.snapshot = NULL;
1718  ondisk->builder.reorder = NULL;
1719  ondisk->builder.committed.xip = NULL;
1720  ondisk->builder.catchange.xip = NULL;
1721  /* update catchange only on disk data */
1722  ondisk->builder.catchange.xcnt = catchange_xcnt;
1723 
1724  COMP_CRC32C(ondisk->checksum,
1725  &ondisk->builder,
1726  sizeof(SnapBuild));
1727 
1728  /* copy committed xacts */
1729  if (builder->committed.xcnt > 0)
1730  {
1731  sz = sizeof(TransactionId) * builder->committed.xcnt;
1732  memcpy(ondisk_c, builder->committed.xip, sz);
1733  COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1734  ondisk_c += sz;
1735  }
1736 
1737  /* copy catalog modifying xacts */
1738  if (catchange_xcnt > 0)
1739  {
1740  sz = sizeof(TransactionId) * catchange_xcnt;
1741  memcpy(ondisk_c, catchange_xip, sz);
1742  COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1743  ondisk_c += sz;
1744  }
1745 
1746  FIN_CRC32C(ondisk->checksum);
1747 
1748  /* we have valid data now, open tempfile and write it there */
1749  fd = OpenTransientFile(tmppath,
1750  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1751  if (fd < 0)
1752  ereport(ERROR,
1754  errmsg("could not open file \"%s\": %m", tmppath)));
1755 
1756  errno = 0;
1758  if ((write(fd, ondisk, needed_length)) != needed_length)
1759  {
1760  int save_errno = errno;
1761 
1763 
1764  /* if write didn't set errno, assume problem is no disk space */
1765  errno = save_errno ? save_errno : ENOSPC;
1766  ereport(ERROR,
1768  errmsg("could not write to file \"%s\": %m", tmppath)));
1769  }
1771 
1772  /*
1773  * fsync the file before renaming so that even if we crash after this we
1774  * have either a fully valid file or nothing.
1775  *
1776  * It's safe to just ERROR on fsync() here because we'll retry the whole
1777  * operation including the writes.
1778  *
1779  * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1780  * some noticeable overhead since it's performed synchronously during
1781  * decoding?
1782  */
1784  if (pg_fsync(fd) != 0)
1785  {
1786  int save_errno = errno;
1787 
1789  errno = save_errno;
1790  ereport(ERROR,
1792  errmsg("could not fsync file \"%s\": %m", tmppath)));
1793  }
1795 
1796  if (CloseTransientFile(fd) != 0)
1797  ereport(ERROR,
1799  errmsg("could not close file \"%s\": %m", tmppath)));
1800 
1801  fsync_fname("pg_logical/snapshots", true);
1802 
1803  /*
1804  * We may overwrite the work from some other backend, but that's ok, our
1805  * snapshot is valid as well, we'll just have done some superfluous work.
1806  */
1807  if (rename(tmppath, path) != 0)
1808  {
1809  ereport(ERROR,
1811  errmsg("could not rename file \"%s\" to \"%s\": %m",
1812  tmppath, path)));
1813  }
1814 
1815  /* make sure we persist */
1816  fsync_fname(path, false);
1817  fsync_fname("pg_logical/snapshots", true);
1818 
1819  /*
1820  * Now there's no way we can loose the dumped state anymore, remember this
1821  * as a serialization point.
1822  */
1823  builder->last_serialized_snapshot = lsn;
1824 
1825  MemoryContextSwitchTo(old_ctx);
1826 
1827 out:
1829  builder->last_serialized_snapshot);
1830  /* be tidy */
1831  if (ondisk)
1832  pfree(ondisk);
1833  if (catchange_xip)
1834  pfree(catchange_xip);
1835 }
int pg_fsync(int fd)
Definition: fd.c:356
int MyProcPid
Definition: globals.c:44
static uint32 dclist_count(dclist_head *head)
Definition: ilist.h:869
#define write(a, b, c)
Definition: win32.h:14
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
struct SnapBuildOnDisk SnapBuildOnDisk
dclist_head catchange_txns
@ WAIT_EVENT_SNAPBUILD_SYNC
Definition: wait_event.h:212
@ WAIT_EVENT_SNAPBUILD_WRITE
Definition: wait_event.h:213
#define stat
Definition: win32_port.h:286

References Assert(), SnapBuildOnDisk::builder, SnapBuild::catchange, ReorderBuffer::catchange_txns, SnapBuildOnDisk::checksum, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, SnapBuild::context, dclist_count(), 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, MemoryContextSwitchTo(), MyProcPid, SnapBuild::next_phase_at, OpenTransientFile(), palloc0(), pfree(), PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), SnapBuild::reorder, ReorderBufferGetCatalogChangesXacts(), ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuild::snapshot, 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().

◆ SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 420 of file snapbuild.c.

421 {
422  builder->two_phase_at = ptr;
423 }

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 453 of file snapbuild.c.

454 {
455  /* make sure we don't get passed an external snapshot */
457 
458  /* make sure nobody modified our snapshot */
459  Assert(snap->curcid == FirstCommandId);
460  Assert(!snap->suboverflowed);
461  Assert(!snap->takenDuringRecovery);
462 
463  Assert(snap->regd_count == 0);
464 
465  Assert(snap->active_count > 0);
466 
467  /* slightly more likely, so it's checked even without casserts */
468  if (snap->copied)
469  elog(ERROR, "cannot free a copied snapshot");
470 
471  snap->active_count--;
472  if (snap->active_count == 0)
473  SnapBuildFreeSnapshot(snap);
474 }
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:377

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

◆ SnapBuildSnapIncRefcount()

static void SnapBuildSnapIncRefcount ( Snapshot  snap)
static

◆ SnapBuildWaitSnapshot()

static void SnapBuildWaitSnapshot ( xl_running_xacts running,
TransactionId  cutoff 
)
static

Definition at line 1506 of file snapbuild.c.

1507 {
1508  int off;
1509 
1510  for (off = 0; off < running->xcnt; off++)
1511  {
1512  TransactionId xid = running->xids[off];
1513 
1514  /*
1515  * Upper layers should prevent that we ever need to wait on ourselves.
1516  * Check anyway, since failing to do so would either result in an
1517  * endless wait or an Assert() failure.
1518  */
1520  elog(ERROR, "waiting for ourselves");
1521 
1522  if (TransactionIdFollows(xid, cutoff))
1523  continue;
1524 
1525  XactLockTableWait(xid, NULL, NULL, XLTW_None);
1526  }
1527 
1528  /*
1529  * All transactions we needed to finish finished - try to ensure there is
1530  * another xl_running_xacts record in a timely manner, without having to
1531  * wait for bgwriter or checkpointer to log one. During recovery we can't
1532  * enforce that, so we'll have to wait.
1533  */
1534  if (!RecoveryInProgress())
1535  {
1537  }
1538 }
void XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid, XLTW_Oper oper)
Definition: lmgr.c:668
@ XLTW_None
Definition: lmgr.h:26
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1272
TransactionId xids[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:56
bool TransactionIdIsCurrentTransactionId(TransactionId xid)
Definition: xact.c:925
bool RecoveryInProgress(void)
Definition: xlog.c:5912

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

Referenced by SnapBuildFindSnapshot().

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 429 of file snapbuild.c.

430 {
431  return ptr < builder->start_decoding_at;
432 }

References SnapBuild::start_decoding_at.

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

◆ SnapBuildXidHasCatalogChanges()

static bool SnapBuildXidHasCatalogChanges ( SnapBuild builder,
TransactionId  xid,
uint32  xinfo 
)
inlinestatic

Definition at line 1183 of file snapbuild.c.

1185 {
1186  if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
1187  return true;
1188 
1189  /*
1190  * The transactions that have changed catalogs must have invalidation
1191  * info.
1192  */
1193  if (!(xinfo & XACT_XINFO_HAS_INVALS))
1194  return false;
1195 
1196  /* Check the catchange XID array */
1197  return ((builder->catchange.xcnt > 0) &&
1198  (bsearch(&xid, builder->catchange.xip, builder->catchange.xcnt,
1199  sizeof(TransactionId), xidComparator) != NULL));
1200 }
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
#define XACT_XINFO_HAS_INVALS
Definition: xact.h:184

References SnapBuild::catchange, SnapBuild::reorder, ReorderBufferXidHasCatalogChanges(), XACT_XINFO_HAS_INVALS, SnapBuild::xcnt, xidComparator(), and SnapBuild::xip.

Referenced by SnapBuildCommitTxn().

Variable Documentation

◆ ExportInProgress

bool ExportInProgress = false
static

◆ SavedResourceOwnerDuringExport

ResourceOwner SavedResourceOwnerDuringExport = NULL
static