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 "replication/snapbuild_internal.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.

Macros

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

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, void *dest, Size size, const char *path)
 
SnapBuildAllocateSnapshotBuilder (ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, bool in_slot_creation, 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)
 
bool SnapBuildRestoreSnapshot (SnapBuildOnDisk *ondisk, XLogRecPtr lsn, MemoryContext context, bool missing_ok)
 
void CheckPointSnapBuild (void)
 
bool SnapBuildSnapshotExists (XLogRecPtr lsn)
 

Variables

static ResourceOwner SavedResourceOwnerDuringExport = NULL
 
static bool ExportInProgress = false
 

Macro Definition Documentation

◆ SNAPBUILD_MAGIC

#define SNAPBUILD_MAGIC   0x51A1E001

Definition at line 1426 of file snapbuild.c.

◆ SNAPBUILD_VERSION

#define SNAPBUILD_VERSION   6

Definition at line 1427 of file snapbuild.c.

◆ SnapBuildOnDiskConstantSize

#define SnapBuildOnDiskConstantSize    offsetof(SnapBuildOnDisk, builder)

Definition at line 1421 of file snapbuild.c.

◆ SnapBuildOnDiskNotChecksummedSize

#define SnapBuildOnDiskNotChecksummedSize    offsetof(SnapBuildOnDisk, version)

Definition at line 1423 of file snapbuild.c.

Function Documentation

◆ AllocateSnapshotBuilder()

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

Definition at line 185 of file snapbuild.c.

191{
192 MemoryContext context;
193 MemoryContext oldcontext;
194 SnapBuild *builder;
195
196 /* allocate memory in own context, to have better accountability */
198 "snapshot builder context",
200 oldcontext = MemoryContextSwitchTo(context);
201
202 builder = palloc0(sizeof(SnapBuild));
203
204 builder->state = SNAPBUILD_START;
205 builder->context = context;
206 builder->reorder = reorder;
207 /* Other struct members initialized by zeroing via palloc0 above */
208
209 builder->committed.xcnt = 0;
210 builder->committed.xcnt_space = 128; /* arbitrary number */
211 builder->committed.xip =
212 palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
213 builder->committed.includes_all_transactions = true;
214
215 builder->catchange.xcnt = 0;
216 builder->catchange.xip = NULL;
217
218 builder->initial_xmin_horizon = xmin_horizon;
219 builder->start_decoding_at = start_lsn;
220 builder->in_slot_creation = in_slot_creation;
221 builder->building_full_snapshot = need_full_snapshot;
222 builder->two_phase_at = two_phase_at;
223
224 MemoryContextSwitchTo(oldcontext);
225
226 return builder;
227}
uint32 TransactionId
Definition: c.h:623
void * palloc0(Size size)
Definition: mcxt.c:1347
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
@ SNAPBUILD_START
Definition: snapbuild.h:27
XLogRecPtr start_decoding_at
SnapBuildState state
TransactionId initial_xmin_horizon
struct SnapBuild::@117 committed
TransactionId * xip
XLogRecPtr two_phase_at
bool building_full_snapshot
bool includes_all_transactions
MemoryContext context
ReorderBuffer * reorder
struct SnapBuild::@118 catchange

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, SnapBuild::building_full_snapshot, SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, CurrentMemoryContext, SnapBuild::in_slot_creation, 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 1922 of file snapbuild.c.

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

Referenced by CheckPointGuts().

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild builder)

Definition at line 233 of file snapbuild.c.

234{
235 MemoryContext context = builder->context;
236
237 /* free snapshot explicitly, that contains some error checking */
238 if (builder->snapshot != NULL)
239 {
241 builder->snapshot = NULL;
242 }
243
244 /* other resources are deallocated via memory context reset */
245 MemoryContextDelete(context);
246}
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:454
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:328
Snapshot snapshot

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

Referenced by FreeDecodingContext().

◆ SnapBuildAddCommittedTxn()

static void SnapBuildAddCommittedTxn ( SnapBuild builder,
TransactionId  xid 
)
static

Definition at line 784 of file snapbuild.c.

785{
787
788 if (builder->committed.xcnt == builder->committed.xcnt_space)
789 {
790 builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
791
792 elog(DEBUG1, "increasing space for committed transactions to %u",
793 (uint32) builder->committed.xcnt_space);
794
795 builder->committed.xip = repalloc(builder->committed.xip,
796 builder->committed.xcnt_space * sizeof(TransactionId));
797 }
798
799 /*
800 * TODO: It might make sense to keep the array sorted here instead of
801 * doing it every time we build a new snapshot. On the other hand this
802 * gets called repeatedly when a transaction with subtransactions commits.
803 */
804 builder->committed.xip[builder->committed.xcnt++] = xid;
805}
Assert(PointerIsAligned(start, uint64))
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1544
#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 360 of file snapbuild.c.

361{
362 Snapshot snapshot;
363 Size ssize;
364
366
367 ssize = sizeof(SnapshotData)
368 + sizeof(TransactionId) * builder->committed.xcnt
369 + sizeof(TransactionId) * 1 /* toplevel xid */ ;
370
371 snapshot = MemoryContextAllocZero(builder->context, ssize);
372
374
375 /*
376 * We misuse the original meaning of SnapshotData's xip and subxip fields
377 * to make the more fitting for our needs.
378 *
379 * In the 'xip' array we store transactions that have to be treated as
380 * committed. Since we will only ever look at tuples from transactions
381 * that have modified the catalog it's more efficient to store those few
382 * that exist between xmin and xmax (frequently there are none).
383 *
384 * Snapshots that are used in transactions that have modified the catalog
385 * also use the 'subxip' array to store their toplevel xid and all the
386 * subtransaction xids so we can recognize when we need to treat rows as
387 * visible that are not in xip but still need to be visible. Subxip only
388 * gets filled when the transaction is copied into the context of a
389 * catalog modifying transaction since we otherwise share a snapshot
390 * between transactions. As long as a txn hasn't modified the catalog it
391 * doesn't need to treat any uncommitted rows as visible, so there is no
392 * need for those xids.
393 *
394 * Both arrays are qsort'ed so that we can use bsearch() on them.
395 */
398
399 snapshot->xmin = builder->xmin;
400 snapshot->xmax = builder->xmax;
401
402 /* store all transactions to be treated as committed by this snapshot */
403 snapshot->xip =
404 (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
405 snapshot->xcnt = builder->committed.xcnt;
406 memcpy(snapshot->xip,
407 builder->committed.xip,
408 builder->committed.xcnt * sizeof(TransactionId));
409
410 /* sort so we can bsearch() */
411 qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
412
413 /*
414 * Initially, subxip is empty, i.e. it's a snapshot to be used by
415 * transactions that don't modify the catalog. Will be filled by
416 * ReorderBufferCopySnap() if necessary.
417 */
418 snapshot->subxcnt = 0;
419 snapshot->subxip = NULL;
420
421 snapshot->suboverflowed = false;
422 snapshot->takenDuringRecovery = false;
423 snapshot->copied = false;
424 snapshot->curcid = FirstCommandId;
425 snapshot->active_count = 0;
426 snapshot->regd_count = 0;
427 snapshot->snapXactCompletionCount = 0;
428
429 return snapshot;
430}
#define FirstCommandId
Definition: c.h:639
size_t Size
Definition: c.h:576
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:1215
#define qsort(a, b, c, d)
Definition: port.h:475
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:43
struct SnapshotData SnapshotData
@ SNAPSHOT_HISTORIC_MVCC
Definition: snapshot.h:105
TransactionId xmin
TransactionId xmax
TransactionId xmin
Definition: snapshot.h:153
int32 subxcnt
Definition: snapshot.h:177
bool copied
Definition: snapshot.h:181
uint32 regd_count
Definition: snapshot.h:201
uint32 active_count
Definition: snapshot.h:200
CommandId curcid
Definition: snapshot.h:183
uint32 xcnt
Definition: snapshot.h:165
TransactionId * subxip
Definition: snapshot.h:176
uint64 snapXactCompletionCount
Definition: snapshot.h:209
TransactionId xmax
Definition: snapshot.h:154
SnapshotType snapshot_type
Definition: snapshot.h:140
TransactionId * xip
Definition: snapshot.h:164
bool suboverflowed
Definition: snapshot.h:178
bool takenDuringRecovery
Definition: snapshot.h:180
#define TransactionIdIsNormal(xid)
Definition: transam.h:42
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:152

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

601{
602 ResourceOwner tmpResOwner;
603
604 /* nothing exported, that is the usual case */
605 if (!ExportInProgress)
606 return;
607
608 if (!IsTransactionState())
609 elog(ERROR, "clearing exported snapshot in wrong transaction state");
610
611 /*
612 * AbortCurrentTransaction() takes care of resetting the snapshot state,
613 * so remember SavedResourceOwnerDuringExport.
614 */
615 tmpResOwner = SavedResourceOwnerDuringExport;
616
617 /* make sure nothing could have ever happened */
619
620 CurrentResourceOwner = tmpResOwner;
621}
#define ERROR
Definition: elog.h:39
ResourceOwner CurrentResourceOwner
Definition: resowner.c:173
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:151
static bool ExportInProgress
Definition: snapbuild.c:152
bool IsTransactionState(void)
Definition: xact.c:387
void AbortCurrentTransaction(void)
Definition: xact.c:3451

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

897{
898 int nxact;
899
900 bool needs_snapshot = false;
901 bool needs_timetravel = false;
902 bool sub_needs_timetravel = false;
903
904 TransactionId xmax = xid;
905
906 /*
907 * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
908 * will they be part of a snapshot. So we don't need to record anything.
909 */
910 if (builder->state == SNAPBUILD_START ||
911 (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
912 TransactionIdPrecedes(xid, builder->next_phase_at)))
913 {
914 /* ensure that only commits after this are getting replayed */
915 if (builder->start_decoding_at <= lsn)
916 builder->start_decoding_at = lsn + 1;
917 return;
918 }
919
920 if (builder->state < SNAPBUILD_CONSISTENT)
921 {
922 /* ensure that only commits after this are getting replayed */
923 if (builder->start_decoding_at <= lsn)
924 builder->start_decoding_at = lsn + 1;
925
926 /*
927 * If building an exportable snapshot, force xid to be tracked, even
928 * if the transaction didn't modify the catalog.
929 */
930 if (builder->building_full_snapshot)
931 {
932 needs_timetravel = true;
933 }
934 }
935
936 for (nxact = 0; nxact < nsubxacts; nxact++)
937 {
938 TransactionId subxid = subxacts[nxact];
939
940 /*
941 * Add subtransaction to base snapshot if catalog modifying, we don't
942 * distinguish to toplevel transactions there.
943 */
944 if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
945 {
946 sub_needs_timetravel = true;
947 needs_snapshot = true;
948
949 elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
950 xid, subxid);
951
952 SnapBuildAddCommittedTxn(builder, subxid);
953
954 if (NormalTransactionIdFollows(subxid, xmax))
955 xmax = subxid;
956 }
957
958 /*
959 * If we're forcing timetravel we also need visibility information
960 * about subtransaction, so keep track of subtransaction's state, even
961 * if not catalog modifying. Don't need to distribute a snapshot in
962 * that case.
963 */
964 else if (needs_timetravel)
965 {
966 SnapBuildAddCommittedTxn(builder, subxid);
967 if (NormalTransactionIdFollows(subxid, xmax))
968 xmax = subxid;
969 }
970 }
971
972 /* if top-level modified catalog, it'll need a snapshot */
973 if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
974 {
975 elog(DEBUG2, "found top level transaction %u, with catalog changes",
976 xid);
977 needs_snapshot = true;
978 needs_timetravel = true;
979 SnapBuildAddCommittedTxn(builder, xid);
980 }
981 else if (sub_needs_timetravel)
982 {
983 /* track toplevel txn as well, subxact alone isn't meaningful */
984 elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
985 xid);
986 needs_timetravel = true;
987 SnapBuildAddCommittedTxn(builder, xid);
988 }
989 else if (needs_timetravel)
990 {
991 elog(DEBUG2, "forced transaction %u to do timetravel", xid);
992
993 SnapBuildAddCommittedTxn(builder, xid);
994 }
995
996 if (!needs_timetravel)
997 {
998 /* record that we cannot export a general snapshot anymore */
999 builder->committed.includes_all_transactions = false;
1000 }
1001
1002 Assert(!needs_snapshot || needs_timetravel);
1003
1004 /*
1005 * Adjust xmax of the snapshot builder, we only do that for committed,
1006 * catalog modifying, transactions, everything else isn't interesting for
1007 * us since we'll never look at the respective rows.
1008 */
1009 if (needs_timetravel &&
1010 (!TransactionIdIsValid(builder->xmax) ||
1011 TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1012 {
1013 builder->xmax = xmax;
1014 TransactionIdAdvance(builder->xmax);
1015 }
1016
1017 /* if there's any reason to build a historic snapshot, do so now */
1018 if (needs_snapshot)
1019 {
1020 /*
1021 * If we haven't built a complete snapshot yet there's no need to hand
1022 * it out, it wouldn't (and couldn't) be used anyway.
1023 */
1024 if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
1025 return;
1026
1027 /*
1028 * Decrease the snapshot builder's refcount of the old snapshot, note
1029 * that it still will be used if it has been handed out to the
1030 * reorderbuffer earlier.
1031 */
1032 if (builder->snapshot)
1034
1035 builder->snapshot = SnapBuildBuildSnapshot(builder);
1036
1037 /* we might need to execute invalidations, add snapshot */
1038 if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1039 {
1041 ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1042 builder->snapshot);
1043 }
1044
1045 /* refcount of the snapshot builder for the new snapshot */
1047
1048 /* add a new catalog snapshot to all currently running transactions */
1050 }
1051}
#define DEBUG2
Definition: elog.h:29
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:316
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:784
static bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid, uint32 xinfo)
Definition: snapbuild.c:1058
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:360
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:731
@ SNAPBUILD_BUILDING_SNAPSHOT
Definition: snapbuild.h:33
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:50
TransactionId next_phase_at
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:329
#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)

◆ SnapBuildDistributeNewCatalogSnapshot()

static void SnapBuildDistributeNewCatalogSnapshot ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 731 of file snapbuild.c.

732{
733 dlist_iter txn_i;
734 ReorderBufferTXN *txn;
735
736 /*
737 * Iterate through all toplevel transactions. This can include
738 * subtransactions which we just don't yet know to be that, but that's
739 * fine, they will just get an unnecessary snapshot queued.
740 */
741 dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
742 {
743 txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
744
746
747 /*
748 * If we don't have a base snapshot yet, there are no changes in this
749 * transaction which in turn implies we don't yet need a snapshot at
750 * all. We'll add a snapshot when the first change gets queued.
751 *
752 * NB: This works correctly even for subtransactions because
753 * ReorderBufferAssignChild() takes care to transfer the base snapshot
754 * to the top-level transaction, and while iterating the changequeue
755 * we'll get the change from the subtxn.
756 */
757 if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
758 continue;
759
760 /*
761 * We don't need to add snapshot to prepared transactions as they
762 * should not see the new catalog contents.
763 */
764 if (rbtxn_is_prepared(txn))
765 continue;
766
767 elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
768 txn->xid, LSN_FORMAT_ARGS(lsn));
769
770 /*
771 * increase the snapshot's refcount for the transaction we are handing
772 * it out to
773 */
775 ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
776 builder->snapshot);
777 }
778}
#define dlist_foreach(iter, lhead)
Definition: ilist.h:623
#define dlist_container(type, membername, ptr)
Definition: ilist.h:593
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
#define rbtxn_is_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_is_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 539 of file snapbuild.c.

540{
541 Snapshot snap;
542 char *snapname;
543
545 elog(ERROR, "cannot export a snapshot from within a transaction");
546
548 elog(ERROR, "can only export one snapshot at a time");
549
551 ExportInProgress = true;
552
554
555 /* There doesn't seem to a nice API to set these */
557 XactReadOnly = true;
558
559 snap = SnapBuildInitialSnapshot(builder);
560
561 /*
562 * now that we've built a plain snapshot, make it active and use the
563 * normal mechanisms for exporting it
564 */
565 snapname = ExportSnapshot(snap);
566
567 ereport(LOG,
568 (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
569 "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
570 snap->xcnt,
571 snapname, snap->xcnt)));
572 return snapname;
573}
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:1180
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:440
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1102
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4989
bool XactReadOnly
Definition: xact.c:82
void StartTransactionCommand(void)
Definition: xact.c:3059
int XactIsoLevel
Definition: xact.c:79
#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 1190 of file snapbuild.c.

1191{
1192 /* ---
1193 * Build catalog decoding snapshot incrementally using information about
1194 * the currently running transactions. There are several ways to do that:
1195 *
1196 * a) There were no running transactions when the xl_running_xacts record
1197 * was inserted, jump to CONSISTENT immediately. We might find such a
1198 * state while waiting on c)'s sub-states.
1199 *
1200 * b) This (in a previous run) or another decoding slot serialized a
1201 * snapshot to disk that we can use. Can't use this method while finding
1202 * the start point for decoding changes as the restart LSN would be an
1203 * arbitrary LSN but we need to find the start point to extract changes
1204 * where we won't see the data for partial transactions. Also, we cannot
1205 * use this method when a slot needs a full snapshot for export or direct
1206 * use, as that snapshot will only contain catalog modifying transactions.
1207 *
1208 * c) First incrementally build a snapshot for catalog tuples
1209 * (BUILDING_SNAPSHOT), that requires all, already in-progress,
1210 * transactions to finish. Every transaction starting after that
1211 * (FULL_SNAPSHOT state), has enough information to be decoded. But
1212 * for older running transactions no viable snapshot exists yet, so
1213 * CONSISTENT will only be reached once all of those have finished.
1214 * ---
1215 */
1216
1217 /*
1218 * xl_running_xacts record is older than what we can use, we might not
1219 * have all necessary catalog rows anymore.
1220 */
1223 builder->initial_xmin_horizon))
1224 {
1226 (errmsg_internal("skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1227 LSN_FORMAT_ARGS(lsn)),
1228 errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
1229 builder->initial_xmin_horizon, running->oldestRunningXid)));
1230
1231
1233
1234 return true;
1235 }
1236
1237 /*
1238 * a) No transaction were running, we can jump to consistent.
1239 *
1240 * This is not affected by races around xl_running_xacts, because we can
1241 * miss transaction commits, but currently not transactions starting.
1242 *
1243 * NB: We might have already started to incrementally assemble a snapshot,
1244 * so we need to be careful to deal with that.
1245 */
1246 if (running->oldestRunningXid == running->nextXid)
1247 {
1248 if (builder->start_decoding_at == InvalidXLogRecPtr ||
1249 builder->start_decoding_at <= lsn)
1250 /* can decode everything after this */
1251 builder->start_decoding_at = lsn + 1;
1252
1253 /* As no transactions were running xmin/xmax can be trivially set. */
1254 builder->xmin = running->nextXid; /* < are finished */
1255 builder->xmax = running->nextXid; /* >= are running */
1256
1257 /* so we can safely use the faster comparisons */
1260
1261 builder->state = SNAPBUILD_CONSISTENT;
1263
1264 ereport(LOG,
1265 (errmsg("logical decoding found consistent point at %X/%X",
1266 LSN_FORMAT_ARGS(lsn)),
1267 errdetail("There are no running transactions.")));
1268
1269 return false;
1270 }
1271
1272 /*
1273 * b) valid on disk state and while neither building full snapshot nor
1274 * creating a slot.
1275 */
1276 else if (!builder->building_full_snapshot &&
1277 !builder->in_slot_creation &&
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_xacts'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 builder->next_phase_at = 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 */
1312
1313 ereport(LOG,
1314 (errmsg("logical decoding found initial starting point at %X/%X",
1315 LSN_FORMAT_ARGS(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 builder->next_phase_at = running->nextXid;
1336
1337 ereport(LOG,
1338 (errmsg("logical decoding found initial consistent point at %X/%X",
1339 LSN_FORMAT_ARGS(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, and this xl_running_xacts' oldestRunningXid is
1350 * >= than nextXid from when we switched to FULL_SNAPSHOT. This means all
1351 * transactions that are currently in progress have a catalog snapshot,
1352 * and all their changes have been collected. Switch to CONSISTENT.
1353 */
1354 else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
1356 running->oldestRunningXid))
1357 {
1358 builder->state = SNAPBUILD_CONSISTENT;
1360
1361 ereport(LOG,
1362 (errmsg("logical decoding found consistent point at %X/%X",
1363 LSN_FORMAT_ARGS(lsn)),
1364 errdetail("There are no old transactions anymore.")));
1365 }
1366
1367 /*
1368 * We already started to track running xacts and need to wait for all
1369 * in-progress ones to finish. We fall through to the normal processing of
1370 * records so incremental cleanup can be performed.
1371 */
1372 return true;
1373}
int errmsg_internal(const char *fmt,...)
Definition: elog.c:1157
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1230
int errdetail(const char *fmt,...)
Definition: elog.c:1203
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
Definition: snapbuild.c:1387
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1793
TransactionId oldestRunningXid
Definition: standbydefs.h:53
TransactionId nextXid
Definition: standbydefs.h:52
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:299
#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::in_slot_creation, 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 252 of file snapbuild.c.

253{
254 /* make sure we don't get passed an external snapshot */
256
257 /* make sure nobody modified our snapshot */
258 Assert(snap->curcid == FirstCommandId);
259 Assert(!snap->suboverflowed);
261 Assert(snap->regd_count == 0);
262
263 /* slightly more likely, so it's checked even without c-asserts */
264 if (snap->copied)
265 elog(ERROR, "cannot free a copied snapshot");
266
267 if (snap->active_count)
268 elog(ERROR, "cannot free an active snapshot");
269
270 pfree(snap);
271}
void pfree(void *pointer)
Definition: mcxt.c:1524

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

580{
581 Assert(builder->state == SNAPBUILD_CONSISTENT);
582
583 /* only build a new snapshot if we don't have a prebuilt one */
584 if (builder->snapshot == NULL)
585 {
586 builder->snapshot = SnapBuildBuildSnapshot(builder);
587 /* increase refcount for the snapshot builder */
589 }
590
591 return builder->snapshot;
592}

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

Referenced by logicalmsg_decode().

◆ SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt ( SnapBuild builder)

Definition at line 286 of file snapbuild.c.

287{
288 return builder->two_phase_at;
289}

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 440 of file snapbuild.c.

441{
442 Snapshot snap;
443 TransactionId xid;
444 TransactionId safeXid;
445 TransactionId *newxip;
446 int newxcnt = 0;
447
450
451 /* don't allow older snapshots */
452 InvalidateCatalogSnapshot(); /* about to overwrite MyProc->xmin */
454 elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
456
457 if (builder->state != SNAPBUILD_CONSISTENT)
458 elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
459
461 elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
462
463 /* so we don't overwrite the existing value */
465 elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
466
467 snap = SnapBuildBuildSnapshot(builder);
468
469 /*
470 * We know that snap->xmin is alive, enforced by the logical xmin
471 * mechanism. Due to that we can do this without locks, we're only
472 * changing our own value.
473 *
474 * Building an initial snapshot is expensive and an unenforced xmin
475 * horizon would have bad consequences, therefore always double-check that
476 * the horizon is enforced.
477 */
478 LWLockAcquire(ProcArrayLock, LW_SHARED);
479 safeXid = GetOldestSafeDecodingTransactionId(false);
480 LWLockRelease(ProcArrayLock);
481
482 if (TransactionIdFollows(safeXid, snap->xmin))
483 elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
484 safeXid, snap->xmin);
485
486 MyProc->xmin = snap->xmin;
487
488 /* allocate in transaction context */
489 newxip = (TransactionId *)
491
492 /*
493 * snapbuild.c builds transactions in an "inverted" manner, which means it
494 * stores committed transactions in ->xip, not ones in progress. Build a
495 * classical snapshot by marking all non-committed transactions as
496 * in-progress. This can be expensive.
497 */
498 for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
499 {
500 void *test;
501
502 /*
503 * Check whether transaction committed using the decoding snapshot
504 * meaning of ->xip.
505 */
506 test = bsearch(&xid, snap->xip, snap->xcnt,
508
509 if (test == NULL)
510 {
511 if (newxcnt >= GetMaxSnapshotXidCount())
514 errmsg("initial slot snapshot too large")));
515
516 newxip[newxcnt++] = xid;
517 }
518
520 }
521
522 /* adjust remaining snapshot fields as needed */
524 snap->xcnt = newxcnt;
525 snap->xip = newxip;
526
527 return snap;
528}
int errcode(int sqlerrcode)
Definition: elog.c:853
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1179
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1899
@ LW_SHARED
Definition: lwlock.h:115
void * palloc(Size size)
Definition: mcxt.c:1317
#define ERRCODE_T_R_SERIALIZATION_FAILURE
Definition: pgbench.c:77
static void test(void)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2945
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:2069
bool HistoricSnapshotActive(void)
Definition: snapmgr.c:1679
bool HaveRegisteredOrActiveSnapshot(void)
Definition: snapmgr.c:1631
void InvalidateCatalogSnapshot(void)
Definition: snapmgr.c:443
@ SNAPSHOT_MVCC
Definition: snapshot.h:46
PGPROC * MyProc
Definition: proc.c:66
TransactionId xmin
Definition: proc.h:178
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
Definition: transam.c:314

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

640{
641 /*
642 * We can't handle data in transactions if we haven't built a snapshot
643 * yet, so don't store them.
644 */
645 if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
646 return false;
647
648 /*
649 * No point in keeping track of changes in transactions that we don't have
650 * enough information about to decode. This means that they started before
651 * we got into the SNAPBUILD_FULL_SNAPSHOT state.
652 */
653 if (builder->state < SNAPBUILD_CONSISTENT &&
655 return false;
656
657 /*
658 * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
659 * be needed to decode the change we're currently processing.
660 */
661 if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
662 {
663 /* only build a new snapshot if we don't have a prebuilt one */
664 if (builder->snapshot == NULL)
665 {
666 builder->snapshot = SnapBuildBuildSnapshot(builder);
667 /* increase refcount for the snapshot builder */
669 }
670
671 /*
672 * Increase refcount for the transaction we're handing the snapshot
673 * out to.
674 */
676 ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
677 builder->snapshot);
678 }
679
680 return true;
681}

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

691{
692 CommandId cid;
693
694 /*
695 * we only log new_cid's if a catalog tuple was modified, so mark the
696 * transaction as containing catalog modifications
697 */
698 ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
699
700 ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
701 xlrec->target_locator, xlrec->target_tid,
702 xlrec->cmin, xlrec->cmax,
703 xlrec->combocid);
704
705 /* figure out new command id */
706 if (xlrec->cmin != InvalidCommandId &&
707 xlrec->cmax != InvalidCommandId)
708 cid = Max(xlrec->cmin, xlrec->cmax);
709 else if (xlrec->cmax != InvalidCommandId)
710 cid = xlrec->cmax;
711 else if (xlrec->cmin != InvalidCommandId)
712 cid = xlrec->cmin;
713 else
714 {
715 cid = InvalidCommandId; /* silence compiler */
716 elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
717 }
718
719 ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
720}
#define InvalidCommandId
Definition: c.h:640
#define Max(x, y)
Definition: c.h:969
uint32 CommandId
Definition: c.h:637
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:459
CommandId combocid
Definition: heapam_xlog.h:461
ItemPointerData target_tid
Definition: heapam_xlog.h:467
TransactionId top_xid
Definition: heapam_xlog.h:458
CommandId cmax
Definition: heapam_xlog.h:460
RelFileLocator target_locator
Definition: heapam_xlog.h:466

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

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

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

819{
820 int off;
821 TransactionId *workspace;
822 int surviving_xids = 0;
823
824 /* not ready yet */
825 if (!TransactionIdIsNormal(builder->xmin))
826 return;
827
828 /* TODO: Neater algorithm than just copying and iterating? */
829 workspace =
831 builder->committed.xcnt * sizeof(TransactionId));
832
833 /* copy xids that still are interesting to workspace */
834 for (off = 0; off < builder->committed.xcnt; off++)
835 {
836 if (NormalTransactionIdPrecedes(builder->committed.xip[off],
837 builder->xmin))
838 ; /* remove */
839 else
840 workspace[surviving_xids++] = builder->committed.xip[off];
841 }
842
843 /* copy workspace back to persistent state */
844 memcpy(builder->committed.xip, workspace,
845 surviving_xids * sizeof(TransactionId));
846
847 elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
848 (uint32) builder->committed.xcnt, (uint32) surviving_xids,
849 builder->xmin, builder->xmax);
850 builder->committed.xcnt = surviving_xids;
851
852 pfree(workspace);
853
854 /*
855 * Purge xids in ->catchange as well. The purged array must also be sorted
856 * in xidComparator order.
857 */
858 if (builder->catchange.xcnt > 0)
859 {
860 /*
861 * Since catchange.xip is sorted, we find the lower bound of xids that
862 * are still interesting.
863 */
864 for (off = 0; off < builder->catchange.xcnt; off++)
865 {
867 builder->xmin))
868 break;
869 }
870
871 surviving_xids = builder->catchange.xcnt - off;
872
873 if (surviving_xids > 0)
874 {
875 memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
876 surviving_xids * sizeof(TransactionId));
877 }
878 else
879 {
880 pfree(builder->catchange.xip);
881 builder->catchange.xip = NULL;
882 }
883
884 elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
885 (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
886 builder->xmin, builder->xmax);
887 builder->catchange.xcnt = surviving_xids;
888 }
889}
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:1181

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

628{
630 ExportInProgress = false;
631}

References ExportInProgress, and SavedResourceOwnerDuringExport.

Referenced by AbortTransaction().

◆ SnapBuildRestore()

static bool SnapBuildRestore ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1793 of file snapbuild.c.

1794{
1795 SnapBuildOnDisk ondisk;
1796
1797 /* no point in loading a snapshot if we're already there */
1798 if (builder->state == SNAPBUILD_CONSISTENT)
1799 return false;
1800
1801 /* validate and restore the snapshot to 'ondisk' */
1802 if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true))
1803 return false;
1804
1805 /*
1806 * ok, we now have a sensible snapshot here, figure out if it has more
1807 * information than we have.
1808 */
1809
1810 /*
1811 * We are only interested in consistent snapshots for now, comparing
1812 * whether one incomplete snapshot is more "advanced" seems to be
1813 * unnecessarily complex.
1814 */
1815 if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
1816 goto snapshot_not_interesting;
1817
1818 /*
1819 * Don't use a snapshot that requires an xmin that we cannot guarantee to
1820 * be available.
1821 */
1823 goto snapshot_not_interesting;
1824
1825 /*
1826 * Consistent snapshots have no next phase. Reset next_phase_at as it is
1827 * possible that an old value may remain.
1828 */
1831
1832 /* ok, we think the snapshot is sensible, copy over everything important */
1833 builder->xmin = ondisk.builder.xmin;
1834 builder->xmax = ondisk.builder.xmax;
1835 builder->state = ondisk.builder.state;
1836
1837 builder->committed.xcnt = ondisk.builder.committed.xcnt;
1838 /* We only allocated/stored xcnt, not xcnt_space xids ! */
1839 /* don't overwrite preallocated xip, if we don't have anything here */
1840 if (builder->committed.xcnt > 0)
1841 {
1842 pfree(builder->committed.xip);
1843 builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1844 builder->committed.xip = ondisk.builder.committed.xip;
1845 }
1846 ondisk.builder.committed.xip = NULL;
1847
1848 /* set catalog modifying transactions */
1849 if (builder->catchange.xip)
1850 pfree(builder->catchange.xip);
1851 builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
1852 builder->catchange.xip = ondisk.builder.catchange.xip;
1853 ondisk.builder.catchange.xip = NULL;
1854
1855 /* our snapshot is not interesting anymore, build a new one */
1856 if (builder->snapshot != NULL)
1857 {
1859 }
1860 builder->snapshot = SnapBuildBuildSnapshot(builder);
1862
1864
1865 Assert(builder->state == SNAPBUILD_CONSISTENT);
1866
1867 ereport(LOG,
1868 (errmsg("logical decoding found consistent point at %X/%X",
1869 LSN_FORMAT_ARGS(lsn)),
1870 errdetail("Logical decoding will begin using saved snapshot.")));
1871 return true;
1872
1873snapshot_not_interesting:
1874 if (ondisk.builder.committed.xip != NULL)
1875 pfree(ondisk.builder.committed.xip);
1876 if (ondisk.builder.catchange.xip != NULL)
1877 pfree(ondisk.builder.catchange.xip);
1878 return false;
1879}
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
bool SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, XLogRecPtr lsn, MemoryContext context, bool missing_ok)
Definition: snapbuild.c:1694

References Assert(), SnapBuildOnDisk::builder, SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, ereport, errdetail(), errmsg(), SnapBuild::initial_xmin_horizon, InvalidTransactionId, LOG, LSN_FORMAT_ARGS, SnapBuild::next_phase_at, pfree(), SnapBuild::reorder, ReorderBufferSetRestartPoint(), SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SnapBuildRestoreSnapshot(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::state, TransactionIdPrecedes(), SnapBuild::xcnt, SnapBuild::xcnt_space, SnapBuild::xip, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by SnapBuildFindSnapshot(), and SnapBuildSerializationPoint().

◆ SnapBuildRestoreContents()

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

Definition at line 1885 of file snapbuild.c.

1886{
1887 int readBytes;
1888
1889 pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
1890 readBytes = read(fd, dest, size);
1892 if (readBytes != size)
1893 {
1894 int save_errno = errno;
1895
1897
1898 if (readBytes < 0)
1899 {
1900 errno = save_errno;
1901 ereport(ERROR,
1903 errmsg("could not read file \"%s\": %m", path)));
1904 }
1905 else
1906 ereport(ERROR,
1908 errmsg("could not read file \"%s\": read %d of %zu",
1909 path, readBytes, size)));
1910 }
1911}
int CloseTransientFile(int fd)
Definition: fd.c:2831
#define read(a, b, c)
Definition: win32.h:13
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:41
static int fd(const char *x, int i)
Definition: preproc-init.c:105
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:85
static void pgstat_report_wait_end(void)
Definition: wait_event.h:101

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(), and read.

Referenced by SnapBuildRestoreSnapshot().

◆ SnapBuildRestoreSnapshot()

bool SnapBuildRestoreSnapshot ( SnapBuildOnDisk ondisk,
XLogRecPtr  lsn,
MemoryContext  context,
bool  missing_ok 
)

Definition at line 1694 of file snapbuild.c.

1696{
1697 int fd;
1698 pg_crc32c checksum;
1699 Size sz;
1700 char path[MAXPGPATH];
1701
1702 sprintf(path, "%s/%X-%X.snap",
1704 LSN_FORMAT_ARGS(lsn));
1705
1706 fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1707
1708 if (fd < 0)
1709 {
1710 if (missing_ok && errno == ENOENT)
1711 return false;
1712
1713 ereport(ERROR,
1715 errmsg("could not open file \"%s\": %m", path)));
1716 }
1717
1718 /* ----
1719 * Make sure the snapshot had been stored safely to disk, that's normally
1720 * cheap.
1721 * Note that we do not need PANIC here, nobody will be able to use the
1722 * slot without fsyncing, and saving it won't succeed without an fsync()
1723 * either...
1724 * ----
1725 */
1726 fsync_fname(path, false);
1728
1729 /* read statically sized portion of snapshot */
1731
1732 if (ondisk->magic != SNAPBUILD_MAGIC)
1733 ereport(ERROR,
1735 errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1736 path, ondisk->magic, SNAPBUILD_MAGIC)));
1737
1738 if (ondisk->version != SNAPBUILD_VERSION)
1739 ereport(ERROR,
1741 errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1742 path, ondisk->version, SNAPBUILD_VERSION)));
1743
1744 INIT_CRC32C(checksum);
1745 COMP_CRC32C(checksum,
1746 ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1748
1749 /* read SnapBuild */
1750 SnapBuildRestoreContents(fd, &ondisk->builder, sizeof(SnapBuild), path);
1751 COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));
1752
1753 /* restore committed xacts information */
1754 if (ondisk->builder.committed.xcnt > 0)
1755 {
1756 sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
1757 ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
1758 SnapBuildRestoreContents(fd, ondisk->builder.committed.xip, sz, path);
1759 COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
1760 }
1761
1762 /* restore catalog modifying xacts information */
1763 if (ondisk->builder.catchange.xcnt > 0)
1764 {
1765 sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
1766 ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
1767 SnapBuildRestoreContents(fd, ondisk->builder.catchange.xip, sz, path);
1768 COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
1769 }
1770
1771 if (CloseTransientFile(fd) != 0)
1772 ereport(ERROR,
1774 errmsg("could not close file \"%s\": %m", path)));
1775
1776 FIN_CRC32C(checksum);
1777
1778 /* verify checksum of what we've read */
1779 if (!EQ_CRC32C(checksum, ondisk->checksum))
1780 ereport(ERROR,
1782 errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1783 path, checksum, ondisk->checksum)));
1784
1785 return true;
1786}
#define PG_BINARY
Definition: c.h:1244
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:755
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2655
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:98
#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:103
#define sprintf
Definition: port.h:241
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1427
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1423
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1426
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1421
static void SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path)
Definition: snapbuild.c:1885

References SnapBuildOnDisk::builder, SnapBuild::catchange, SnapBuildOnDisk::checksum, CloseTransientFile(), SnapBuild::committed, COMP_CRC32C, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg(), ERROR, fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, LSN_FORMAT_ARGS, SnapBuildOnDisk::magic, MAXPGPATH, MemoryContextAllocZero(), OpenTransientFile(), PG_BINARY, PG_LOGICAL_SNAPSHOTS_DIR, SNAPBUILD_MAGIC, SNAPBUILD_VERSION, SnapBuildOnDiskConstantSize, SnapBuildOnDiskNotChecksummedSize, SnapBuildRestoreContents(), sprintf, SnapBuildOnDisk::version, SnapBuild::xcnt, and SnapBuild::xip.

Referenced by pg_get_logical_snapshot_info(), pg_get_logical_snapshot_meta(), and SnapBuildRestore().

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1436 of file snapbuild.c.

1437{
1438 if (builder->state < SNAPBUILD_CONSISTENT)
1439 SnapBuildRestore(builder, lsn);
1440 else
1441 SnapBuildSerialize(builder, lsn);
1442}

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

Referenced by xlog_decode().

◆ SnapBuildSerialize()

static void SnapBuildSerialize ( SnapBuild builder,
XLogRecPtr  lsn 
)
static

Definition at line 1449 of file snapbuild.c.

1450{
1451 Size needed_length;
1452 SnapBuildOnDisk *ondisk = NULL;
1453 TransactionId *catchange_xip = NULL;
1454 MemoryContext old_ctx;
1455 size_t catchange_xcnt;
1456 char *ondisk_c;
1457 int fd;
1458 char tmppath[MAXPGPATH];
1459 char path[MAXPGPATH];
1460 int ret;
1461 struct stat stat_buf;
1462 Size sz;
1463
1464 Assert(lsn != InvalidXLogRecPtr);
1466 builder->last_serialized_snapshot <= lsn);
1467
1468 /*
1469 * no point in serializing if we cannot continue to work immediately after
1470 * restoring the snapshot
1471 */
1472 if (builder->state < SNAPBUILD_CONSISTENT)
1473 return;
1474
1475 /* consistent snapshots have no next phase */
1477
1478 /*
1479 * We identify snapshots by the LSN they are valid for. We don't need to
1480 * include timelines in the name as each LSN maps to exactly one timeline
1481 * unless the user used pg_resetwal or similar. If a user did so, there's
1482 * no hope continuing to decode anyway.
1483 */
1484 sprintf(path, "%s/%X-%X.snap",
1486 LSN_FORMAT_ARGS(lsn));
1487
1488 /*
1489 * first check whether some other backend already has written the snapshot
1490 * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1491 * as a valid state. Everything else is an unexpected error.
1492 */
1493 ret = stat(path, &stat_buf);
1494
1495 if (ret != 0 && errno != ENOENT)
1496 ereport(ERROR,
1498 errmsg("could not stat file \"%s\": %m", path)));
1499
1500 else if (ret == 0)
1501 {
1502 /*
1503 * somebody else has already serialized to this point, don't overwrite
1504 * but remember location, so we don't need to read old data again.
1505 *
1506 * To be sure it has been synced to disk after the rename() from the
1507 * tempfile filename to the real filename, we just repeat the fsync.
1508 * That ought to be cheap because in most scenarios it should already
1509 * be safely on disk.
1510 */
1511 fsync_fname(path, false);
1513
1514 builder->last_serialized_snapshot = lsn;
1515 goto out;
1516 }
1517
1518 /*
1519 * there is an obvious race condition here between the time we stat(2) the
1520 * file and us writing the file. But we rename the file into place
1521 * atomically and all files created need to contain the same data anyway,
1522 * so this is perfectly fine, although a bit of a resource waste. Locking
1523 * seems like pointless complication.
1524 */
1525 elog(DEBUG1, "serializing snapshot to %s", path);
1526
1527 /* to make sure only we will write to this tempfile, include pid */
1528 sprintf(tmppath, "%s/%X-%X.snap.%d.tmp",
1531
1532 /*
1533 * Unlink temporary file if it already exists, needs to have been before a
1534 * crash/error since we won't enter this function twice from within a
1535 * single decoding slot/backend and the temporary file contains the pid of
1536 * the current process.
1537 */
1538 if (unlink(tmppath) != 0 && errno != ENOENT)
1539 ereport(ERROR,
1541 errmsg("could not remove file \"%s\": %m", tmppath)));
1542
1543 old_ctx = MemoryContextSwitchTo(builder->context);
1544
1545 /* Get the catalog modifying transactions that are yet not committed */
1546 catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
1547 catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
1548
1549 needed_length = sizeof(SnapBuildOnDisk) +
1550 sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
1551
1552 ondisk_c = palloc0(needed_length);
1553 ondisk = (SnapBuildOnDisk *) ondisk_c;
1554 ondisk->magic = SNAPBUILD_MAGIC;
1555 ondisk->version = SNAPBUILD_VERSION;
1556 ondisk->length = needed_length;
1557 INIT_CRC32C(ondisk->checksum);
1558 COMP_CRC32C(ondisk->checksum,
1559 ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1561 ondisk_c += sizeof(SnapBuildOnDisk);
1562
1563 memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1564 /* NULL-ify memory-only data */
1565 ondisk->builder.context = NULL;
1566 ondisk->builder.snapshot = NULL;
1567 ondisk->builder.reorder = NULL;
1568 ondisk->builder.committed.xip = NULL;
1569 ondisk->builder.catchange.xip = NULL;
1570 /* update catchange only on disk data */
1571 ondisk->builder.catchange.xcnt = catchange_xcnt;
1572
1573 COMP_CRC32C(ondisk->checksum,
1574 &ondisk->builder,
1575 sizeof(SnapBuild));
1576
1577 /* copy committed xacts */
1578 if (builder->committed.xcnt > 0)
1579 {
1580 sz = sizeof(TransactionId) * builder->committed.xcnt;
1581 memcpy(ondisk_c, builder->committed.xip, sz);
1582 COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1583 ondisk_c += sz;
1584 }
1585
1586 /* copy catalog modifying xacts */
1587 if (catchange_xcnt > 0)
1588 {
1589 sz = sizeof(TransactionId) * catchange_xcnt;
1590 memcpy(ondisk_c, catchange_xip, sz);
1591 COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1592 ondisk_c += sz;
1593 }
1594
1595 FIN_CRC32C(ondisk->checksum);
1596
1597 /* we have valid data now, open tempfile and write it there */
1598 fd = OpenTransientFile(tmppath,
1599 O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1600 if (fd < 0)
1601 ereport(ERROR,
1603 errmsg("could not open file \"%s\": %m", tmppath)));
1604
1605 errno = 0;
1606 pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
1607 if ((write(fd, ondisk, needed_length)) != needed_length)
1608 {
1609 int save_errno = errno;
1610
1612
1613 /* if write didn't set errno, assume problem is no disk space */
1614 errno = save_errno ? save_errno : ENOSPC;
1615 ereport(ERROR,
1617 errmsg("could not write to file \"%s\": %m", tmppath)));
1618 }
1620
1621 /*
1622 * fsync the file before renaming so that even if we crash after this we
1623 * have either a fully valid file or nothing.
1624 *
1625 * It's safe to just ERROR on fsync() here because we'll retry the whole
1626 * operation including the writes.
1627 *
1628 * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1629 * some noticeable overhead since it's performed synchronously during
1630 * decoding?
1631 */
1632 pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
1633 if (pg_fsync(fd) != 0)
1634 {
1635 int save_errno = errno;
1636
1638 errno = save_errno;
1639 ereport(ERROR,
1641 errmsg("could not fsync file \"%s\": %m", tmppath)));
1642 }
1644
1645 if (CloseTransientFile(fd) != 0)
1646 ereport(ERROR,
1648 errmsg("could not close file \"%s\": %m", tmppath)));
1649
1651
1652 /*
1653 * We may overwrite the work from some other backend, but that's ok, our
1654 * snapshot is valid as well, we'll just have done some superfluous work.
1655 */
1656 if (rename(tmppath, path) != 0)
1657 {
1658 ereport(ERROR,
1660 errmsg("could not rename file \"%s\" to \"%s\": %m",
1661 tmppath, path)));
1662 }
1663
1664 /* make sure we persist */
1665 fsync_fname(path, false);
1667
1668 /*
1669 * Now there's no way we can lose the dumped state anymore, remember this
1670 * as a serialization point.
1671 */
1672 builder->last_serialized_snapshot = lsn;
1673
1674 MemoryContextSwitchTo(old_ctx);
1675
1676out:
1678 builder->last_serialized_snapshot);
1679 /* be tidy */
1680 if (ondisk)
1681 pfree(ondisk);
1682 if (catchange_xip)
1683 pfree(catchange_xip);
1684}
int pg_fsync(int fd)
Definition: fd.c:385
int MyProcPid
Definition: globals.c:46
static uint32 dclist_count(const dclist_head *head)
Definition: ilist.h:932
#define write(a, b, c)
Definition: win32.h:14
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
struct SnapBuildOnDisk SnapBuildOnDisk
dclist_head catchange_txns
#define stat
Definition: win32_port.h:274

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(), PG_LOGICAL_SNAPSHOTS_DIR, 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, write, SnapBuild::xcnt, and SnapBuild::xip.

Referenced by SnapBuildProcessRunningXacts(), and SnapBuildSerializationPoint().

◆ SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 295 of file snapbuild.c.

296{
297 builder->two_phase_at = ptr;
298}

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 328 of file snapbuild.c.

329{
330 /* make sure we don't get passed an external snapshot */
332
333 /* make sure nobody modified our snapshot */
334 Assert(snap->curcid == FirstCommandId);
335 Assert(!snap->suboverflowed);
337
338 Assert(snap->regd_count == 0);
339
340 Assert(snap->active_count > 0);
341
342 /* slightly more likely, so it's checked even without casserts */
343 if (snap->copied)
344 elog(ERROR, "cannot free a copied snapshot");
345
346 snap->active_count--;
347 if (snap->active_count == 0)
349}
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:252

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

◆ SnapBuildSnapshotExists()

bool SnapBuildSnapshotExists ( XLogRecPtr  lsn)

Definition at line 2010 of file snapbuild.c.

2011{
2012 char path[MAXPGPATH];
2013 int ret;
2014 struct stat stat_buf;
2015
2016 sprintf(path, "%s/%X-%X.snap",
2018 LSN_FORMAT_ARGS(lsn));
2019
2020 ret = stat(path, &stat_buf);
2021
2022 if (ret != 0 && errno != ENOENT)
2023 ereport(ERROR,
2025 errmsg("could not stat file \"%s\": %m", path)));
2026
2027 return ret == 0;
2028}

References ereport, errcode_for_file_access(), errmsg(), ERROR, LSN_FORMAT_ARGS, MAXPGPATH, PG_LOGICAL_SNAPSHOTS_DIR, sprintf, and stat.

Referenced by update_local_synced_slot().

◆ SnapBuildWaitSnapshot()

static void SnapBuildWaitSnapshot ( xl_running_xacts running,
TransactionId  cutoff 
)
static

Definition at line 1387 of file snapbuild.c.

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

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

305{
306 return ptr < builder->start_decoding_at;
307}

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

1060{
1061 if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
1062 return true;
1063
1064 /*
1065 * The transactions that have changed catalogs must have invalidation
1066 * info.
1067 */
1068 if (!(xinfo & XACT_XINFO_HAS_INVALS))
1069 return false;
1070
1071 /* Check the catchange XID array */
1072 return ((builder->catchange.xcnt > 0) &&
1073 (bsearch(&xid, builder->catchange.xip, builder->catchange.xcnt,
1074 sizeof(TransactionId), xidComparator) != NULL));
1075}
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
#define XACT_XINFO_HAS_INVALS
Definition: xact.h:191

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