PostgreSQL Source Code  git master
snapbuild.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * snapbuild.c
4  *
5  * Infrastructure for building historic catalog snapshots based on contents
6  * of the WAL, for the purpose of decoding heapam.c style values in the
7  * WAL.
8  *
9  * NOTES:
10  *
11  * We build snapshots which can *only* be used to read catalog contents and we
12  * do so by reading and interpreting the WAL stream. The aim is to build a
13  * snapshot that behaves the same as a freshly taken MVCC snapshot would have
14  * at the time the XLogRecord was generated.
15  *
16  * To build the snapshots we reuse the infrastructure built for Hot
17  * Standby. The in-memory snapshots we build look different than HS' because
18  * we have different needs. To successfully decode data from the WAL we only
19  * need to access catalog tables and (sys|rel|cat)cache, not the actual user
20  * tables since the data we decode is wholly contained in the WAL
21  * records. Also, our snapshots need to be different in comparison to normal
22  * MVCC ones because in contrast to those we cannot fully rely on the clog and
23  * pg_subtrans for information about committed transactions because they might
24  * commit in the future from the POV of the WAL entry we're currently
25  * decoding. This definition has the advantage that we only need to prevent
26  * removal of catalog rows, while normal table's rows can still be
27  * removed. This is achieved by using the replication slot mechanism.
28  *
29  * As the percentage of transactions modifying the catalog normally is fairly
30  * small in comparisons to ones only manipulating user data, we keep track of
31  * the committed catalog modifying ones inside [xmin, xmax) instead of keeping
32  * track of all running transactions like it's done in a normal snapshot. Note
33  * that we're generally only looking at transactions that have acquired an
34  * xid. That is we keep a list of transactions between snapshot->(xmin, xmax)
35  * that we consider committed, everything else is considered aborted/in
36  * progress. That also allows us not to care about subtransactions before they
37  * have committed which means this module, in contrast to HS, doesn't have to
38  * care about suboverflowed subtransactions and similar.
39  *
40  * One complexity of doing this is that to e.g. handle mixed DDL/DML
41  * transactions we need Snapshots that see intermediate versions of the
42  * catalog in a transaction. During normal operation this is achieved by using
43  * CommandIds/cmin/cmax. The problem with that however is that for space
44  * efficiency reasons only one value of that is stored
45  * (cf. combocid.c). Since ComboCids are only available in memory we log
46  * additional information which allows us to get the original (cmin, cmax)
47  * pair during visibility checks. Check the reorderbuffer.c's comment above
48  * ResolveCminCmaxDuringDecoding() for details.
49  *
50  * To facilitate all this we need our own visibility routine, as the normal
51  * ones are optimized for different usecases.
52  *
53  * To replace the normal catalog snapshots with decoding ones use the
54  * SetupHistoricSnapshot() and TeardownHistoricSnapshot() functions.
55  *
56  *
57  *
58  * The snapbuild machinery is starting up in several stages, as illustrated
59  * by the following graph describing the SnapBuild->state transitions:
60  *
61  * +-------------------------+
62  * +----| START |-------------+
63  * | +-------------------------+ |
64  * | | |
65  * | | |
66  * | running_xacts #1 |
67  * | | |
68  * | | |
69  * | v |
70  * | +-------------------------+ v
71  * | | BUILDING_SNAPSHOT |------------>|
72  * | +-------------------------+ |
73  * | | |
74  * | | |
75  * | running_xacts #2, xacts from #1 finished |
76  * | | |
77  * | | |
78  * | v |
79  * | +-------------------------+ v
80  * | | FULL_SNAPSHOT |------------>|
81  * | +-------------------------+ |
82  * | | |
83  * running_xacts | saved snapshot
84  * with zero xacts | at running_xacts's lsn
85  * | | |
86  * | running_xacts with xacts from #2 finished |
87  * | | |
88  * | v |
89  * | +-------------------------+ |
90  * +--->|SNAPBUILD_CONSISTENT |<------------+
91  * +-------------------------+
92  *
93  * Initially the machinery is in the START stage. When an xl_running_xacts
94  * record is read that is sufficiently new (above the safe xmin horizon),
95  * there's a state transition. If there were no running xacts when the
96  * running_xacts record was generated, we'll directly go into CONSISTENT
97  * state, otherwise we'll switch to the BUILDING_SNAPSHOT state. Having a full
98  * snapshot means that all transactions that start henceforth can be decoded
99  * in their entirety, but transactions that started previously can't. In
100  * FULL_SNAPSHOT we'll switch into CONSISTENT once all those previously
101  * running transactions have committed or aborted.
102  *
103  * Only transactions that commit after CONSISTENT state has been reached will
104  * be replayed, even though they might have started while still in
105  * FULL_SNAPSHOT. That ensures that we'll reach a point where no previous
106  * changes has been exported, but all the following ones will be. That point
107  * is a convenient point to initialize replication from, which is why we
108  * export a snapshot at that point, which *can* be used to read normal data.
109  *
110  * Copyright (c) 2012-2021, PostgreSQL Global Development Group
111  *
112  * IDENTIFICATION
113  * src/backend/replication/snapbuild.c
114  *
115  *-------------------------------------------------------------------------
116  */
117 
118 #include "postgres.h"
119 
120 #include <sys/stat.h>
121 #include <unistd.h>
122 
123 #include "access/heapam_xlog.h"
124 #include "access/transam.h"
125 #include "access/xact.h"
126 #include "miscadmin.h"
127 #include "pgstat.h"
128 #include "replication/logical.h"
130 #include "replication/snapbuild.h"
131 #include "storage/block.h" /* debugging output */
132 #include "storage/fd.h"
133 #include "storage/lmgr.h"
134 #include "storage/proc.h"
135 #include "storage/procarray.h"
136 #include "storage/standby.h"
137 #include "utils/builtins.h"
138 #include "utils/memutils.h"
139 #include "utils/snapmgr.h"
140 #include "utils/snapshot.h"
141 
142 /*
143  * This struct contains the current state of the snapshot building
144  * machinery. Besides a forward declaration in the header, it is not exposed
145  * to the public, so we can easily change its contents.
146  */
147 struct SnapBuild
148 {
149  /* how far are we along building our first full snapshot */
151 
152  /* private memory context used to allocate memory for this module. */
154 
155  /* all transactions < than this have committed/aborted */
157 
158  /* all transactions >= than this are uncommitted */
160 
161  /*
162  * Don't replay commits from an LSN < this LSN. This can be set externally
163  * but it will also be advanced (never retreat) from within snapbuild.c.
164  */
166 
167  /*
168  * LSN at which we found a consistent point at the time of slot creation.
169  * This is also the point where we have exported a snapshot for the
170  * initial copy.
171  *
172  * The prepared transactions that are not covered by initial snapshot
173  * needs to be sent later along with commit prepared and they must be
174  * before this point.
175  */
177 
178  /*
179  * Don't start decoding WAL until the "xl_running_xacts" information
180  * indicates there are no running xids with an xid smaller than this.
181  */
183 
184  /* Indicates if we are building full snapshot or just catalog one. */
186 
187  /*
188  * Snapshot that's valid to see the catalog state seen at this moment.
189  */
191 
192  /*
193  * LSN of the last location we are sure a snapshot has been serialized to.
194  */
196 
197  /*
198  * The reorderbuffer we need to update with usable snapshots et al.
199  */
201 
202  /*
203  * TransactionId at which the next phase of initial snapshot building will
204  * happen. InvalidTransactionId if not known (i.e. SNAPBUILD_START), or
205  * when no next phase necessary (SNAPBUILD_CONSISTENT).
206  */
208 
209  /*
210  * Array of transactions which could have catalog changes that committed
211  * between xmin and xmax.
212  */
213  struct
214  {
215  /* number of committed transactions */
216  size_t xcnt;
217 
218  /* available space for committed transactions */
219  size_t xcnt_space;
220 
221  /*
222  * Until we reach a CONSISTENT state, we record commits of all
223  * transactions, not just the catalog changing ones. Record when that
224  * changes so we know we cannot export a snapshot safely anymore.
225  */
227 
228  /*
229  * Array of committed transactions that have modified the catalog.
230  *
231  * As this array is frequently modified we do *not* keep it in
232  * xidComparator order. Instead we sort the array when building &
233  * distributing a snapshot.
234  *
235  * TODO: It's unclear whether that reasoning has much merit. Every
236  * time we add something here after becoming consistent will also
237  * require distributing a snapshot. Storing them sorted would
238  * potentially also make it easier to purge (but more complicated wrt
239  * wraparound?). Should be improved if sorting while building the
240  * snapshot shows up in profiles.
241  */
243  } committed;
244 };
245 
246 /*
247  * Starting a transaction -- which we need to do while exporting a snapshot --
248  * removes knowledge about the previously used resowner, so we save it here.
249  */
251 static bool ExportInProgress = false;
252 
253 /* ->committed manipulation */
254 static void SnapBuildPurgeCommittedTxn(SnapBuild *builder);
255 
256 /* snapshot building/manipulation/distribution functions */
257 static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder);
258 
259 static void SnapBuildFreeSnapshot(Snapshot snap);
260 
261 static void SnapBuildSnapIncRefcount(Snapshot snap);
262 
264 
265 /* xlog reading helper functions for SnapBuildProcessRunningXacts */
266 static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running);
267 static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);
268 
269 /* serialization functions */
270 static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);
271 static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
272 
273 /*
274  * Allocate a new snapshot builder.
275  *
276  * xmin_horizon is the xid >= which we can be sure no catalog rows have been
277  * removed, start_lsn is the LSN >= we want to replay commits.
278  */
279 SnapBuild *
281  TransactionId xmin_horizon,
282  XLogRecPtr start_lsn,
283  bool need_full_snapshot,
285 {
287  MemoryContext oldcontext;
288  SnapBuild *builder;
289 
290  /* allocate memory in own context, to have better accountability */
292  "snapshot builder context",
294  oldcontext = MemoryContextSwitchTo(context);
295 
296  builder = palloc0(sizeof(SnapBuild));
297 
298  builder->state = SNAPBUILD_START;
299  builder->context = context;
300  builder->reorder = reorder;
301  /* Other struct members initialized by zeroing via palloc0 above */
302 
303  builder->committed.xcnt = 0;
304  builder->committed.xcnt_space = 128; /* arbitrary number */
305  builder->committed.xip =
306  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
307  builder->committed.includes_all_transactions = true;
308 
309  builder->initial_xmin_horizon = xmin_horizon;
310  builder->start_decoding_at = start_lsn;
311  builder->building_full_snapshot = need_full_snapshot;
313 
314  MemoryContextSwitchTo(oldcontext);
315 
316  return builder;
317 }
318 
319 /*
320  * Free a snapshot builder.
321  */
322 void
324 {
325  MemoryContext context = builder->context;
326 
327  /* free snapshot explicitly, that contains some error checking */
328  if (builder->snapshot != NULL)
329  {
331  builder->snapshot = NULL;
332  }
333 
334  /* other resources are deallocated via memory context reset */
335  MemoryContextDelete(context);
336 }
337 
338 /*
339  * Free an unreferenced snapshot that has previously been built by us.
340  */
341 static void
343 {
344  /* make sure we don't get passed an external snapshot */
346 
347  /* make sure nobody modified our snapshot */
348  Assert(snap->curcid == FirstCommandId);
349  Assert(!snap->suboverflowed);
350  Assert(!snap->takenDuringRecovery);
351  Assert(snap->regd_count == 0);
352 
353  /* slightly more likely, so it's checked even without c-asserts */
354  if (snap->copied)
355  elog(ERROR, "cannot free a copied snapshot");
356 
357  if (snap->active_count)
358  elog(ERROR, "cannot free an active snapshot");
359 
360  pfree(snap);
361 }
362 
363 /*
364  * In which state of snapshot building are we?
365  */
368 {
369  return builder->state;
370 }
371 
372 /*
373  * Return the LSN at which the snapshot was exported
374  */
377 {
378  return builder->initial_consistent_point;
379 }
380 
381 /*
382  * Should the contents of transaction ending at 'ptr' be decoded?
383  */
384 bool
386 {
387  return ptr < builder->start_decoding_at;
388 }
389 
390 /*
391  * Increase refcount of a snapshot.
392  *
393  * This is used when handing out a snapshot to some external resource or when
394  * adding a Snapshot as builder->snapshot.
395  */
396 static void
398 {
399  snap->active_count++;
400 }
401 
402 /*
403  * Decrease refcount of a snapshot and free if the refcount reaches zero.
404  *
405  * Externally visible, so that external resources that have been handed an
406  * IncRef'ed Snapshot can adjust its refcount easily.
407  */
408 void
410 {
411  /* make sure we don't get passed an external snapshot */
413 
414  /* make sure nobody modified our snapshot */
415  Assert(snap->curcid == FirstCommandId);
416  Assert(!snap->suboverflowed);
417  Assert(!snap->takenDuringRecovery);
418 
419  Assert(snap->regd_count == 0);
420 
421  Assert(snap->active_count > 0);
422 
423  /* slightly more likely, so it's checked even without casserts */
424  if (snap->copied)
425  elog(ERROR, "cannot free a copied snapshot");
426 
427  snap->active_count--;
428  if (snap->active_count == 0)
429  SnapBuildFreeSnapshot(snap);
430 }
431 
432 /*
433  * Build a new snapshot, based on currently committed catalog-modifying
434  * transactions.
435  *
436  * In-progress transactions with catalog access are *not* allowed to modify
437  * these snapshots; they have to copy them and fill in appropriate ->curcid
438  * and ->subxip/subxcnt values.
439  */
440 static Snapshot
442 {
444  Size ssize;
445 
446  Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
447 
448  ssize = sizeof(SnapshotData)
449  + sizeof(TransactionId) * builder->committed.xcnt
450  + sizeof(TransactionId) * 1 /* toplevel xid */ ;
451 
452  snapshot = MemoryContextAllocZero(builder->context, ssize);
453 
455 
456  /*
457  * We misuse the original meaning of SnapshotData's xip and subxip fields
458  * to make the more fitting for our needs.
459  *
460  * In the 'xip' array we store transactions that have to be treated as
461  * committed. Since we will only ever look at tuples from transactions
462  * that have modified the catalog it's more efficient to store those few
463  * that exist between xmin and xmax (frequently there are none).
464  *
465  * Snapshots that are used in transactions that have modified the catalog
466  * also use the 'subxip' array to store their toplevel xid and all the
467  * subtransaction xids so we can recognize when we need to treat rows as
468  * visible that are not in xip but still need to be visible. Subxip only
469  * gets filled when the transaction is copied into the context of a
470  * catalog modifying transaction since we otherwise share a snapshot
471  * between transactions. As long as a txn hasn't modified the catalog it
472  * doesn't need to treat any uncommitted rows as visible, so there is no
473  * need for those xids.
474  *
475  * Both arrays are qsort'ed so that we can use bsearch() on them.
476  */
477  Assert(TransactionIdIsNormal(builder->xmin));
478  Assert(TransactionIdIsNormal(builder->xmax));
479 
480  snapshot->xmin = builder->xmin;
481  snapshot->xmax = builder->xmax;
482 
483  /* store all transactions to be treated as committed by this snapshot */
484  snapshot->xip =
485  (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
486  snapshot->xcnt = builder->committed.xcnt;
487  memcpy(snapshot->xip,
488  builder->committed.xip,
489  builder->committed.xcnt * sizeof(TransactionId));
490 
491  /* sort so we can bsearch() */
492  qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
493 
494  /*
495  * Initially, subxip is empty, i.e. it's a snapshot to be used by
496  * transactions that don't modify the catalog. Will be filled by
497  * ReorderBufferCopySnap() if necessary.
498  */
499  snapshot->subxcnt = 0;
500  snapshot->subxip = NULL;
501 
502  snapshot->suboverflowed = false;
503  snapshot->takenDuringRecovery = false;
504  snapshot->copied = false;
505  snapshot->curcid = FirstCommandId;
506  snapshot->active_count = 0;
507  snapshot->regd_count = 0;
508  snapshot->snapXactCompletionCount = 0;
509 
510  return snapshot;
511 }
512 
513 /*
514  * Build the initial slot snapshot and convert it to a normal snapshot that
515  * is understood by HeapTupleSatisfiesMVCC.
516  *
517  * The snapshot will be usable directly in current transaction or exported
518  * for loading in different transaction.
519  */
520 Snapshot
522 {
523  Snapshot snap;
524  TransactionId xid;
525  TransactionId *newxip;
526  int newxcnt = 0;
527 
530 
531  if (builder->state != SNAPBUILD_CONSISTENT)
532  elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
533 
534  if (!builder->committed.includes_all_transactions)
535  elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
536 
537  /* so we don't overwrite the existing value */
539  elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
540 
541  snap = SnapBuildBuildSnapshot(builder);
542 
543  /*
544  * We know that snap->xmin is alive, enforced by the logical xmin
545  * mechanism. Due to that we can do this without locks, we're only
546  * changing our own value.
547  */
548 #ifdef USE_ASSERT_CHECKING
549  {
550  TransactionId safeXid;
551 
552  LWLockAcquire(ProcArrayLock, LW_SHARED);
553  safeXid = GetOldestSafeDecodingTransactionId(false);
554  LWLockRelease(ProcArrayLock);
555 
556  Assert(TransactionIdPrecedesOrEquals(safeXid, snap->xmin));
557  }
558 #endif
559 
560  MyProc->xmin = snap->xmin;
561 
562  /* allocate in transaction context */
563  newxip = (TransactionId *)
565 
566  /*
567  * snapbuild.c builds transactions in an "inverted" manner, which means it
568  * stores committed transactions in ->xip, not ones in progress. Build a
569  * classical snapshot by marking all non-committed transactions as
570  * in-progress. This can be expensive.
571  */
572  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
573  {
574  void *test;
575 
576  /*
577  * Check whether transaction committed using the decoding snapshot
578  * meaning of ->xip.
579  */
580  test = bsearch(&xid, snap->xip, snap->xcnt,
581  sizeof(TransactionId), xidComparator);
582 
583  if (test == NULL)
584  {
585  if (newxcnt >= GetMaxSnapshotXidCount())
586  ereport(ERROR,
587  (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
588  errmsg("initial slot snapshot too large")));
589 
590  newxip[newxcnt++] = xid;
591  }
592 
594  }
595 
596  /* adjust remaining snapshot fields as needed */
598  snap->xcnt = newxcnt;
599  snap->xip = newxip;
600 
601  return snap;
602 }
603 
604 /*
605  * Export a snapshot so it can be set in another session with SET TRANSACTION
606  * SNAPSHOT.
607  *
608  * For that we need to start a transaction in the current backend as the
609  * importing side checks whether the source transaction is still open to make
610  * sure the xmin horizon hasn't advanced since then.
611  */
612 const char *
614 {
615  Snapshot snap;
616  char *snapname;
617 
619  elog(ERROR, "cannot export a snapshot from within a transaction");
620 
621  if (SavedResourceOwnerDuringExport)
622  elog(ERROR, "can only export one snapshot at a time");
623 
624  SavedResourceOwnerDuringExport = CurrentResourceOwner;
625  ExportInProgress = true;
626 
628 
629  /* There doesn't seem to a nice API to set these */
631  XactReadOnly = true;
632 
633  snap = SnapBuildInitialSnapshot(builder);
634 
635  /*
636  * now that we've built a plain snapshot, make it active and use the
637  * normal mechanisms for exporting it
638  */
639  snapname = ExportSnapshot(snap);
640 
641  ereport(LOG,
642  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
643  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
644  snap->xcnt,
645  snapname, snap->xcnt)));
646  return snapname;
647 }
648 
649 /*
650  * Ensure there is a snapshot and if not build one for current transaction.
651  */
652 Snapshot
654 {
655  Assert(builder->state == SNAPBUILD_CONSISTENT);
656 
657  /* only build a new snapshot if we don't have a prebuilt one */
658  if (builder->snapshot == NULL)
659  {
660  builder->snapshot = SnapBuildBuildSnapshot(builder);
661  /* increase refcount for the snapshot builder */
663  }
664 
665  return builder->snapshot;
666 }
667 
668 /*
669  * Reset a previously SnapBuildExportSnapshot()'ed snapshot if there is
670  * any. Aborts the previously started transaction and resets the resource
671  * owner back to its original value.
672  */
673 void
675 {
676  /* nothing exported, that is the usual case */
677  if (!ExportInProgress)
678  return;
679 
680  if (!IsTransactionState())
681  elog(ERROR, "clearing exported snapshot in wrong transaction state");
682 
683  /* make sure nothing could have ever happened */
685 
687  SavedResourceOwnerDuringExport = NULL;
688  ExportInProgress = false;
689 }
690 
691 /*
692  * Handle the effects of a single heap change, appropriate to the current state
693  * of the snapshot builder and returns whether changes made at (xid, lsn) can
694  * be decoded.
695  */
696 bool
698 {
699  /*
700  * We can't handle data in transactions if we haven't built a snapshot
701  * yet, so don't store them.
702  */
703  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
704  return false;
705 
706  /*
707  * No point in keeping track of changes in transactions that we don't have
708  * enough information about to decode. This means that they started before
709  * we got into the SNAPBUILD_FULL_SNAPSHOT state.
710  */
711  if (builder->state < SNAPBUILD_CONSISTENT &&
712  TransactionIdPrecedes(xid, builder->next_phase_at))
713  return false;
714 
715  /*
716  * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
717  * be needed to decode the change we're currently processing.
718  */
719  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
720  {
721  /* only build a new snapshot if we don't have a prebuilt one */
722  if (builder->snapshot == NULL)
723  {
724  builder->snapshot = SnapBuildBuildSnapshot(builder);
725  /* increase refcount for the snapshot builder */
727  }
728 
729  /*
730  * Increase refcount for the transaction we're handing the snapshot
731  * out to.
732  */
734  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
735  builder->snapshot);
736  }
737 
738  return true;
739 }
740 
741 /*
742  * Do CommandId/ComboCid handling after reading an xl_heap_new_cid record.
743  * This implies that a transaction has done some form of write to system
744  * catalogs.
745  */
746 void
748  XLogRecPtr lsn, xl_heap_new_cid *xlrec)
749 {
750  CommandId cid;
751 
752  /*
753  * we only log new_cid's if a catalog tuple was modified, so mark the
754  * transaction as containing catalog modifications
755  */
756  ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
757 
758  ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
759  xlrec->target_node, xlrec->target_tid,
760  xlrec->cmin, xlrec->cmax,
761  xlrec->combocid);
762 
763  /* figure out new command id */
764  if (xlrec->cmin != InvalidCommandId &&
765  xlrec->cmax != InvalidCommandId)
766  cid = Max(xlrec->cmin, xlrec->cmax);
767  else if (xlrec->cmax != InvalidCommandId)
768  cid = xlrec->cmax;
769  else if (xlrec->cmin != InvalidCommandId)
770  cid = xlrec->cmin;
771  else
772  {
773  cid = InvalidCommandId; /* silence compiler */
774  elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
775  }
776 
777  ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
778 }
779 
780 /*
781  * Add a new Snapshot to all transactions we're decoding that currently are
782  * in-progress so they can see new catalog contents made by the transaction
783  * that just committed. This is necessary because those in-progress
784  * transactions will use the new catalog's contents from here on (at the very
785  * least everything they do needs to be compatible with newer catalog
786  * contents).
787  */
788 static void
790 {
791  dlist_iter txn_i;
792  ReorderBufferTXN *txn;
793 
794  /*
795  * Iterate through all toplevel transactions. This can include
796  * subtransactions which we just don't yet know to be that, but that's
797  * fine, they will just get an unnecessary snapshot queued.
798  */
799  dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
800  {
801  txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
802 
804 
805  /*
806  * If we don't have a base snapshot yet, there are no changes in this
807  * transaction which in turn implies we don't yet need a snapshot at
808  * all. We'll add a snapshot when the first change gets queued.
809  *
810  * NB: This works correctly even for subtransactions because
811  * ReorderBufferAssignChild() takes care to transfer the base snapshot
812  * to the top-level transaction, and while iterating the changequeue
813  * we'll get the change from the subtxn.
814  */
815  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
816  continue;
817 
818  /*
819  * We don't need to add snapshot to prepared transactions as they
820  * should not see the new catalog contents.
821  */
822  if (rbtxn_prepared(txn) || rbtxn_skip_prepared(txn))
823  continue;
824 
825  elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
826  txn->xid, LSN_FORMAT_ARGS(lsn));
827 
828  /*
829  * increase the snapshot's refcount for the transaction we are handing
830  * it out to
831  */
833  ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
834  builder->snapshot);
835  }
836 }
837 
838 /*
839  * Keep track of a new catalog changing transaction that has committed.
840  */
841 static void
843 {
845 
846  if (builder->committed.xcnt == builder->committed.xcnt_space)
847  {
848  builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
849 
850  elog(DEBUG1, "increasing space for committed transactions to %u",
851  (uint32) builder->committed.xcnt_space);
852 
853  builder->committed.xip = repalloc(builder->committed.xip,
854  builder->committed.xcnt_space * sizeof(TransactionId));
855  }
856 
857  /*
858  * TODO: It might make sense to keep the array sorted here instead of
859  * doing it every time we build a new snapshot. On the other hand this
860  * gets called repeatedly when a transaction with subtransactions commits.
861  */
862  builder->committed.xip[builder->committed.xcnt++] = xid;
863 }
864 
865 /*
866  * Remove knowledge about transactions we treat as committed that are smaller
867  * than ->xmin. Those won't ever get checked via the ->committed array but via
868  * the clog machinery, so we don't need to waste memory on them.
869  */
870 static void
872 {
873  int off;
874  TransactionId *workspace;
875  int surviving_xids = 0;
876 
877  /* not ready yet */
878  if (!TransactionIdIsNormal(builder->xmin))
879  return;
880 
881  /* TODO: Neater algorithm than just copying and iterating? */
882  workspace =
883  MemoryContextAlloc(builder->context,
884  builder->committed.xcnt * sizeof(TransactionId));
885 
886  /* copy xids that still are interesting to workspace */
887  for (off = 0; off < builder->committed.xcnt; off++)
888  {
889  if (NormalTransactionIdPrecedes(builder->committed.xip[off],
890  builder->xmin))
891  ; /* remove */
892  else
893  workspace[surviving_xids++] = builder->committed.xip[off];
894  }
895 
896  /* copy workspace back to persistent state */
897  memcpy(builder->committed.xip, workspace,
898  surviving_xids * sizeof(TransactionId));
899 
900  elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
901  (uint32) builder->committed.xcnt, (uint32) surviving_xids,
902  builder->xmin, builder->xmax);
903  builder->committed.xcnt = surviving_xids;
904 
905  pfree(workspace);
906 }
907 
908 /*
909  * Handle everything that needs to be done when a transaction commits
910  */
911 void
913  int nsubxacts, TransactionId *subxacts)
914 {
915  int nxact;
916 
917  bool needs_snapshot = false;
918  bool needs_timetravel = false;
919  bool sub_needs_timetravel = false;
920 
921  TransactionId xmax = xid;
922 
923  /*
924  * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
925  * will they be part of a snapshot. So we don't need to record anything.
926  */
927  if (builder->state == SNAPBUILD_START ||
928  (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
929  TransactionIdPrecedes(xid, builder->next_phase_at)))
930  {
931  /* ensure that only commits after this are getting replayed */
932  if (builder->start_decoding_at <= lsn)
933  builder->start_decoding_at = lsn + 1;
934  return;
935  }
936 
937  if (builder->state < SNAPBUILD_CONSISTENT)
938  {
939  /* ensure that only commits after this are getting replayed */
940  if (builder->start_decoding_at <= lsn)
941  builder->start_decoding_at = lsn + 1;
942 
943  /*
944  * If building an exportable snapshot, force xid to be tracked, even
945  * if the transaction didn't modify the catalog.
946  */
947  if (builder->building_full_snapshot)
948  {
949  needs_timetravel = true;
950  }
951  }
952 
953  for (nxact = 0; nxact < nsubxacts; nxact++)
954  {
955  TransactionId subxid = subxacts[nxact];
956 
957  /*
958  * Add subtransaction to base snapshot if catalog modifying, we don't
959  * distinguish to toplevel transactions there.
960  */
961  if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid))
962  {
963  sub_needs_timetravel = true;
964  needs_snapshot = true;
965 
966  elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
967  xid, subxid);
968 
969  SnapBuildAddCommittedTxn(builder, subxid);
970 
971  if (NormalTransactionIdFollows(subxid, xmax))
972  xmax = subxid;
973  }
974 
975  /*
976  * If we're forcing timetravel we also need visibility information
977  * about subtransaction, so keep track of subtransaction's state, even
978  * if not catalog modifying. Don't need to distribute a snapshot in
979  * that case.
980  */
981  else if (needs_timetravel)
982  {
983  SnapBuildAddCommittedTxn(builder, subxid);
984  if (NormalTransactionIdFollows(subxid, xmax))
985  xmax = subxid;
986  }
987  }
988 
989  /* if top-level modified catalog, it'll need a snapshot */
990  if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
991  {
992  elog(DEBUG2, "found top level transaction %u, with catalog changes",
993  xid);
994  needs_snapshot = true;
995  needs_timetravel = true;
996  SnapBuildAddCommittedTxn(builder, xid);
997  }
998  else if (sub_needs_timetravel)
999  {
1000  /* track toplevel txn as well, subxact alone isn't meaningful */
1001  SnapBuildAddCommittedTxn(builder, xid);
1002  }
1003  else if (needs_timetravel)
1004  {
1005  elog(DEBUG2, "forced transaction %u to do timetravel", xid);
1006 
1007  SnapBuildAddCommittedTxn(builder, xid);
1008  }
1009 
1010  if (!needs_timetravel)
1011  {
1012  /* record that we cannot export a general snapshot anymore */
1013  builder->committed.includes_all_transactions = false;
1014  }
1015 
1016  Assert(!needs_snapshot || needs_timetravel);
1017 
1018  /*
1019  * Adjust xmax of the snapshot builder, we only do that for committed,
1020  * catalog modifying, transactions, everything else isn't interesting for
1021  * us since we'll never look at the respective rows.
1022  */
1023  if (needs_timetravel &&
1024  (!TransactionIdIsValid(builder->xmax) ||
1025  TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1026  {
1027  builder->xmax = xmax;
1028  TransactionIdAdvance(builder->xmax);
1029  }
1030 
1031  /* if there's any reason to build a historic snapshot, do so now */
1032  if (needs_snapshot)
1033  {
1034  /*
1035  * If we haven't built a complete snapshot yet there's no need to hand
1036  * it out, it wouldn't (and couldn't) be used anyway.
1037  */
1038  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
1039  return;
1040 
1041  /*
1042  * Decrease the snapshot builder's refcount of the old snapshot, note
1043  * that it still will be used if it has been handed out to the
1044  * reorderbuffer earlier.
1045  */
1046  if (builder->snapshot)
1048 
1049  builder->snapshot = SnapBuildBuildSnapshot(builder);
1050 
1051  /* we might need to execute invalidations, add snapshot */
1052  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1053  {
1055  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1056  builder->snapshot);
1057  }
1058 
1059  /* refcount of the snapshot builder for the new snapshot */
1061 
1062  /* add a new catalog snapshot to all currently running transactions */
1064  }
1065 }
1066 
1067 
1068 /* -----------------------------------
1069  * Snapshot building functions dealing with xlog records
1070  * -----------------------------------
1071  */
1072 
1073 /*
1074  * Process a running xacts record, and use its information to first build a
1075  * historic snapshot and later to release resources that aren't needed
1076  * anymore.
1077  */
1078 void
1080 {
1081  ReorderBufferTXN *txn;
1083 
1084  /*
1085  * If we're not consistent yet, inspect the record to see whether it
1086  * allows to get closer to being consistent. If we are consistent, dump
1087  * our snapshot so others or we, after a restart, can use it.
1088  */
1089  if (builder->state < SNAPBUILD_CONSISTENT)
1090  {
1091  /* returns false if there's no point in performing cleanup just yet */
1092  if (!SnapBuildFindSnapshot(builder, lsn, running))
1093  return;
1094  }
1095  else
1096  SnapBuildSerialize(builder, lsn);
1097 
1098  /*
1099  * Update range of interesting xids based on the running xacts
1100  * information. We don't increase ->xmax using it, because once we are in
1101  * a consistent state we can do that ourselves and much more efficiently
1102  * so, because we only need to do it for catalog transactions since we
1103  * only ever look at those.
1104  *
1105  * NB: We only increase xmax when a catalog modifying transaction commits
1106  * (see SnapBuildCommitTxn). Because of this, xmax can be lower than
1107  * xmin, which looks odd but is correct and actually more efficient, since
1108  * we hit fast paths in heapam_visibility.c.
1109  */
1110  builder->xmin = running->oldestRunningXid;
1111 
1112  /* Remove transactions we don't need to keep track off anymore */
1113  SnapBuildPurgeCommittedTxn(builder);
1114 
1115  /*
1116  * Advance the xmin limit for the current replication slot, to allow
1117  * vacuum to clean up the tuples this slot has been protecting.
1118  *
1119  * The reorderbuffer might have an xmin among the currently running
1120  * snapshots; use it if so. If not, we need only consider the snapshots
1121  * we'll produce later, which can't be less than the oldest running xid in
1122  * the record we're reading now.
1123  */
1124  xmin = ReorderBufferGetOldestXmin(builder->reorder);
1125  if (xmin == InvalidTransactionId)
1126  xmin = running->oldestRunningXid;
1127  elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1128  builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
1129  LogicalIncreaseXminForSlot(lsn, xmin);
1130 
1131  /*
1132  * Also tell the slot where we can restart decoding from. We don't want to
1133  * do that after every commit because changing that implies an fsync of
1134  * the logical slot's state file, so we only do it every time we see a
1135  * running xacts record.
1136  *
1137  * Do so by looking for the oldest in progress transaction (determined by
1138  * the first LSN of any of its relevant records). Every transaction
1139  * remembers the last location we stored the snapshot to disk before its
1140  * beginning. That point is where we can restart from.
1141  */
1142 
1143  /*
1144  * Can't know about a serialized snapshot's location if we're not
1145  * consistent.
1146  */
1147  if (builder->state < SNAPBUILD_CONSISTENT)
1148  return;
1149 
1150  txn = ReorderBufferGetOldestTXN(builder->reorder);
1151 
1152  /*
1153  * oldest ongoing txn might have started when we didn't yet serialize
1154  * anything because we hadn't reached a consistent state yet.
1155  */
1156  if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
1158 
1159  /*
1160  * No in-progress transaction, can reuse the last serialized snapshot if
1161  * we have one.
1162  */
1163  else if (txn == NULL &&
1167  builder->last_serialized_snapshot);
1168 }
1169 
1170 
1171 /*
1172  * Build the start of a snapshot that's capable of decoding the catalog.
1173  *
1174  * Helper function for SnapBuildProcessRunningXacts() while we're not yet
1175  * consistent.
1176  *
1177  * Returns true if there is a point in performing internal maintenance/cleanup
1178  * using the xl_running_xacts record.
1179  */
1180 static bool
1182 {
1183  /* ---
1184  * Build catalog decoding snapshot incrementally using information about
1185  * the currently running transactions. There are several ways to do that:
1186  *
1187  * a) There were no running transactions when the xl_running_xacts record
1188  * was inserted, jump to CONSISTENT immediately. We might find such a
1189  * state while waiting on c)'s sub-states.
1190  *
1191  * b) This (in a previous run) or another decoding slot serialized a
1192  * snapshot to disk that we can use. Can't use this method for the
1193  * initial snapshot when slot is being created and needs full snapshot
1194  * for export or direct use, as that snapshot will only contain catalog
1195  * modifying transactions.
1196  *
1197  * c) First incrementally build a snapshot for catalog tuples
1198  * (BUILDING_SNAPSHOT), that requires all, already in-progress,
1199  * transactions to finish. Every transaction starting after that
1200  * (FULL_SNAPSHOT state), has enough information to be decoded. But
1201  * for older running transactions no viable snapshot exists yet, so
1202  * CONSISTENT will only be reached once all of those have finished.
1203  * ---
1204  */
1205 
1206  /*
1207  * xl_running_xact record is older than what we can use, we might not have
1208  * all necessary catalog rows anymore.
1209  */
1212  builder->initial_xmin_horizon))
1213  {
1214  ereport(DEBUG1,
1215  (errmsg_internal("skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1216  LSN_FORMAT_ARGS(lsn)),
1217  errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
1218  builder->initial_xmin_horizon, running->oldestRunningXid)));
1219 
1220 
1221  SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
1222 
1223  return true;
1224  }
1225 
1226  /*
1227  * a) No transaction were running, we can jump to consistent.
1228  *
1229  * This is not affected by races around xl_running_xacts, because we can
1230  * miss transaction commits, but currently not transactions starting.
1231  *
1232  * NB: We might have already started to incrementally assemble a snapshot,
1233  * so we need to be careful to deal with that.
1234  */
1235  if (running->oldestRunningXid == running->nextXid)
1236  {
1237  if (builder->start_decoding_at == InvalidXLogRecPtr ||
1238  builder->start_decoding_at <= lsn)
1239  /* can decode everything after this */
1240  builder->start_decoding_at = lsn + 1;
1241 
1242  /* As no transactions were running xmin/xmax can be trivially set. */
1243  builder->xmin = running->nextXid; /* < are finished */
1244  builder->xmax = running->nextXid; /* >= are running */
1245 
1246  /* so we can safely use the faster comparisons */
1247  Assert(TransactionIdIsNormal(builder->xmin));
1248  Assert(TransactionIdIsNormal(builder->xmax));
1249 
1250  builder->state = SNAPBUILD_CONSISTENT;
1252 
1253  ereport(LOG,
1254  (errmsg("logical decoding found consistent point at %X/%X",
1255  LSN_FORMAT_ARGS(lsn)),
1256  errdetail("There are no running transactions.")));
1257 
1258  return false;
1259  }
1260  /* b) valid on disk state and not building full snapshot */
1261  else if (!builder->building_full_snapshot &&
1262  SnapBuildRestore(builder, lsn))
1263  {
1264  /* there won't be any state to cleanup */
1265  return false;
1266  }
1267 
1268  /*
1269  * c) transition from START to BUILDING_SNAPSHOT.
1270  *
1271  * In START state, and a xl_running_xacts record with running xacts is
1272  * encountered. In that case, switch to BUILDING_SNAPSHOT state, and
1273  * record xl_running_xacts->nextXid. Once all running xacts have finished
1274  * (i.e. they're all >= nextXid), we have a complete catalog snapshot. It
1275  * might look that we could use xl_running_xact's ->xids information to
1276  * get there quicker, but that is problematic because transactions marked
1277  * as running, might already have inserted their commit record - it's
1278  * infeasible to change that with locking.
1279  */
1280  else if (builder->state == SNAPBUILD_START)
1281  {
1283  builder->next_phase_at = running->nextXid;
1284 
1285  /*
1286  * Start with an xmin/xmax that's correct for future, when all the
1287  * currently running transactions have finished. We'll update both
1288  * while waiting for the pending transactions to finish.
1289  */
1290  builder->xmin = running->nextXid; /* < are finished */
1291  builder->xmax = running->nextXid; /* >= are running */
1292 
1293  /* so we can safely use the faster comparisons */
1294  Assert(TransactionIdIsNormal(builder->xmin));
1295  Assert(TransactionIdIsNormal(builder->xmax));
1296 
1297  ereport(LOG,
1298  (errmsg("logical decoding found initial starting point at %X/%X",
1299  LSN_FORMAT_ARGS(lsn)),
1300  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1301  running->xcnt, running->nextXid)));
1302 
1303  SnapBuildWaitSnapshot(running, running->nextXid);
1304  }
1305 
1306  /*
1307  * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
1308  *
1309  * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
1310  * is >= than nextXid from when we switched to BUILDING_SNAPSHOT. This
1311  * means all transactions starting afterwards have enough information to
1312  * be decoded. Switch to FULL_SNAPSHOT.
1313  */
1314  else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
1316  running->oldestRunningXid))
1317  {
1318  builder->state = SNAPBUILD_FULL_SNAPSHOT;
1319  builder->next_phase_at = running->nextXid;
1320 
1321  ereport(LOG,
1322  (errmsg("logical decoding found initial consistent point at %X/%X",
1323  LSN_FORMAT_ARGS(lsn)),
1324  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1325  running->xcnt, running->nextXid)));
1326 
1327  SnapBuildWaitSnapshot(running, running->nextXid);
1328  }
1329 
1330  /*
1331  * c) transition from FULL_SNAPSHOT to CONSISTENT.
1332  *
1333  * In FULL_SNAPSHOT state (see d) ), and this xl_running_xacts'
1334  * oldestRunningXid is >= than nextXid from when we switched to
1335  * FULL_SNAPSHOT. This means all transactions that are currently in
1336  * progress have a catalog snapshot, and all their changes have been
1337  * collected. Switch to CONSISTENT.
1338  */
1339  else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
1341  running->oldestRunningXid))
1342  {
1343  builder->state = SNAPBUILD_CONSISTENT;
1345 
1346  ereport(LOG,
1347  (errmsg("logical decoding found consistent point at %X/%X",
1348  LSN_FORMAT_ARGS(lsn)),
1349  errdetail("There are no old transactions anymore.")));
1350  }
1351 
1352  /*
1353  * We already started to track running xacts and need to wait for all
1354  * in-progress ones to finish. We fall through to the normal processing of
1355  * records so incremental cleanup can be performed.
1356  */
1357  return true;
1358 
1359 }
1360 
1361 /* ---
1362  * Iterate through xids in record, wait for all older than the cutoff to
1363  * finish. Then, if possible, log a new xl_running_xacts record.
1364  *
1365  * This isn't required for the correctness of decoding, but to:
1366  * a) allow isolationtester to notice that we're currently waiting for
1367  * something.
1368  * b) log a new xl_running_xacts record where it'd be helpful, without having
1369  * to wait for bgwriter or checkpointer.
1370  * ---
1371  */
1372 static void
1374 {
1375  int off;
1376 
1377  for (off = 0; off < running->xcnt; off++)
1378  {
1379  TransactionId xid = running->xids[off];
1380 
1381  /*
1382  * Upper layers should prevent that we ever need to wait on ourselves.
1383  * Check anyway, since failing to do so would either result in an
1384  * endless wait or an Assert() failure.
1385  */
1387  elog(ERROR, "waiting for ourselves");
1388 
1389  if (TransactionIdFollows(xid, cutoff))
1390  continue;
1391 
1392  XactLockTableWait(xid, NULL, NULL, XLTW_None);
1393  }
1394 
1395  /*
1396  * All transactions we needed to finish finished - try to ensure there is
1397  * another xl_running_xacts record in a timely manner, without having to
1398  * wait for bgwriter or checkpointer to log one. During recovery we
1399  * can't enforce that, so we'll have to wait.
1400  */
1401  if (!RecoveryInProgress())
1402  {
1404  }
1405 }
1406 
1407 /* -----------------------------------
1408  * Snapshot serialization support
1409  * -----------------------------------
1410  */
1411 
1412 /*
1413  * We store current state of struct SnapBuild on disk in the following manner:
1414  *
1415  * struct SnapBuildOnDisk;
1416  * TransactionId * running.xcnt_space;
1417  * TransactionId * committed.xcnt; (*not xcnt_space*)
1418  *
1419  */
1420 typedef struct SnapBuildOnDisk
1421 {
1422  /* first part of this struct needs to be version independent */
1423 
1424  /* data not covered by checksum */
1427 
1428  /* data covered by checksum */
1429 
1430  /* version, in case we want to support pg_upgrade */
1432  /* how large is the on disk data, excluding the constant sized part */
1434 
1435  /* version dependent part */
1437 
1438  /* variable amount of TransactionIds follows */
1439 } SnapBuildOnDisk;
1440 
1441 #define SnapBuildOnDiskConstantSize \
1442  offsetof(SnapBuildOnDisk, builder)
1443 #define SnapBuildOnDiskNotChecksummedSize \
1444  offsetof(SnapBuildOnDisk, version)
1445 
1446 #define SNAPBUILD_MAGIC 0x51A1E001
1447 #define SNAPBUILD_VERSION 4
1448 
1449 /*
1450  * Store/Load a snapshot from disk, depending on the snapshot builder's state.
1451  *
1452  * Supposed to be used by external (i.e. not snapbuild.c) code that just read
1453  * a record that's a potential location for a serialized snapshot.
1454  */
1455 void
1457 {
1458  if (builder->state < SNAPBUILD_CONSISTENT)
1459  SnapBuildRestore(builder, lsn);
1460  else
1461  SnapBuildSerialize(builder, lsn);
1462 }
1463 
1464 /*
1465  * Serialize the snapshot 'builder' at the location 'lsn' if it hasn't already
1466  * been done by another decoding process.
1467  */
1468 static void
1470 {
1471  Size needed_length;
1472  SnapBuildOnDisk *ondisk = NULL;
1473  char *ondisk_c;
1474  int fd;
1475  char tmppath[MAXPGPATH];
1476  char path[MAXPGPATH];
1477  int ret;
1478  struct stat stat_buf;
1479  Size sz;
1480 
1481  Assert(lsn != InvalidXLogRecPtr);
1483  builder->last_serialized_snapshot <= lsn);
1484 
1485  /*
1486  * no point in serializing if we cannot continue to work immediately after
1487  * restoring the snapshot
1488  */
1489  if (builder->state < SNAPBUILD_CONSISTENT)
1490  return;
1491 
1492  /* consistent snapshots have no next phase */
1494 
1495  /*
1496  * We identify snapshots by the LSN they are valid for. We don't need to
1497  * include timelines in the name as each LSN maps to exactly one timeline
1498  * unless the user used pg_resetwal or similar. If a user did so, there's
1499  * no hope continuing to decode anyway.
1500  */
1501  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1502  LSN_FORMAT_ARGS(lsn));
1503 
1504  /*
1505  * first check whether some other backend already has written the snapshot
1506  * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1507  * as a valid state. Everything else is an unexpected error.
1508  */
1509  ret = stat(path, &stat_buf);
1510 
1511  if (ret != 0 && errno != ENOENT)
1512  ereport(ERROR,
1514  errmsg("could not stat file \"%s\": %m", path)));
1515 
1516  else if (ret == 0)
1517  {
1518  /*
1519  * somebody else has already serialized to this point, don't overwrite
1520  * but remember location, so we don't need to read old data again.
1521  *
1522  * To be sure it has been synced to disk after the rename() from the
1523  * tempfile filename to the real filename, we just repeat the fsync.
1524  * That ought to be cheap because in most scenarios it should already
1525  * be safely on disk.
1526  */
1527  fsync_fname(path, false);
1528  fsync_fname("pg_logical/snapshots", true);
1529 
1530  builder->last_serialized_snapshot = lsn;
1531  goto out;
1532  }
1533 
1534  /*
1535  * there is an obvious race condition here between the time we stat(2) the
1536  * file and us writing the file. But we rename the file into place
1537  * atomically and all files created need to contain the same data anyway,
1538  * so this is perfectly fine, although a bit of a resource waste. Locking
1539  * seems like pointless complication.
1540  */
1541  elog(DEBUG1, "serializing snapshot to %s", path);
1542 
1543  /* to make sure only we will write to this tempfile, include pid */
1544  sprintf(tmppath, "pg_logical/snapshots/%X-%X.snap.%u.tmp",
1545  LSN_FORMAT_ARGS(lsn), MyProcPid);
1546 
1547  /*
1548  * Unlink temporary file if it already exists, needs to have been before a
1549  * crash/error since we won't enter this function twice from within a
1550  * single decoding slot/backend and the temporary file contains the pid of
1551  * the current process.
1552  */
1553  if (unlink(tmppath) != 0 && errno != ENOENT)
1554  ereport(ERROR,
1556  errmsg("could not remove file \"%s\": %m", tmppath)));
1557 
1558  needed_length = sizeof(SnapBuildOnDisk) +
1559  sizeof(TransactionId) * builder->committed.xcnt;
1560 
1561  ondisk_c = MemoryContextAllocZero(builder->context, needed_length);
1562  ondisk = (SnapBuildOnDisk *) ondisk_c;
1563  ondisk->magic = SNAPBUILD_MAGIC;
1564  ondisk->version = SNAPBUILD_VERSION;
1565  ondisk->length = needed_length;
1566  INIT_CRC32C(ondisk->checksum);
1567  COMP_CRC32C(ondisk->checksum,
1568  ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1570  ondisk_c += sizeof(SnapBuildOnDisk);
1571 
1572  memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1573  /* NULL-ify memory-only data */
1574  ondisk->builder.context = NULL;
1575  ondisk->builder.snapshot = NULL;
1576  ondisk->builder.reorder = NULL;
1577  ondisk->builder.committed.xip = NULL;
1578 
1579  COMP_CRC32C(ondisk->checksum,
1580  &ondisk->builder,
1581  sizeof(SnapBuild));
1582 
1583  /* copy committed xacts */
1584  sz = sizeof(TransactionId) * builder->committed.xcnt;
1585  memcpy(ondisk_c, builder->committed.xip, sz);
1586  COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1587  ondisk_c += sz;
1588 
1589  FIN_CRC32C(ondisk->checksum);
1590 
1591  /* we have valid data now, open tempfile and write it there */
1592  fd = OpenTransientFile(tmppath,
1593  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1594  if (fd < 0)
1595  ereport(ERROR,
1597  errmsg("could not open file \"%s\": %m", tmppath)));
1598 
1599  errno = 0;
1601  if ((write(fd, ondisk, needed_length)) != needed_length)
1602  {
1603  int save_errno = errno;
1604 
1605  CloseTransientFile(fd);
1606 
1607  /* if write didn't set errno, assume problem is no disk space */
1608  errno = save_errno ? save_errno : ENOSPC;
1609  ereport(ERROR,
1611  errmsg("could not write to file \"%s\": %m", tmppath)));
1612  }
1614 
1615  /*
1616  * fsync the file before renaming so that even if we crash after this we
1617  * have either a fully valid file or nothing.
1618  *
1619  * It's safe to just ERROR on fsync() here because we'll retry the whole
1620  * operation including the writes.
1621  *
1622  * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1623  * some noticeable overhead since it's performed synchronously during
1624  * decoding?
1625  */
1627  if (pg_fsync(fd) != 0)
1628  {
1629  int save_errno = errno;
1630 
1631  CloseTransientFile(fd);
1632  errno = save_errno;
1633  ereport(ERROR,
1635  errmsg("could not fsync file \"%s\": %m", tmppath)));
1636  }
1638 
1639  if (CloseTransientFile(fd) != 0)
1640  ereport(ERROR,
1642  errmsg("could not close file \"%s\": %m", tmppath)));
1643 
1644  fsync_fname("pg_logical/snapshots", true);
1645 
1646  /*
1647  * We may overwrite the work from some other backend, but that's ok, our
1648  * snapshot is valid as well, we'll just have done some superfluous work.
1649  */
1650  if (rename(tmppath, path) != 0)
1651  {
1652  ereport(ERROR,
1654  errmsg("could not rename file \"%s\" to \"%s\": %m",
1655  tmppath, path)));
1656  }
1657 
1658  /* make sure we persist */
1659  fsync_fname(path, false);
1660  fsync_fname("pg_logical/snapshots", true);
1661 
1662  /*
1663  * Now there's no way we can loose the dumped state anymore, remember this
1664  * as a serialization point.
1665  */
1666  builder->last_serialized_snapshot = lsn;
1667 
1668 out:
1670  builder->last_serialized_snapshot);
1671  /* be tidy */
1672  if (ondisk)
1673  pfree(ondisk);
1674 }
1675 
1676 /*
1677  * Restore a snapshot into 'builder' if previously one has been stored at the
1678  * location indicated by 'lsn'. Returns true if successful, false otherwise.
1679  */
1680 static bool
1682 {
1683  SnapBuildOnDisk ondisk;
1684  int fd;
1685  char path[MAXPGPATH];
1686  Size sz;
1687  int readBytes;
1688  pg_crc32c checksum;
1689 
1690  /* no point in loading a snapshot if we're already there */
1691  if (builder->state == SNAPBUILD_CONSISTENT)
1692  return false;
1693 
1694  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
1695  LSN_FORMAT_ARGS(lsn));
1696 
1697  fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1698 
1699  if (fd < 0 && errno == ENOENT)
1700  return false;
1701  else if (fd < 0)
1702  ereport(ERROR,
1704  errmsg("could not open file \"%s\": %m", path)));
1705 
1706  /* ----
1707  * Make sure the snapshot had been stored safely to disk, that's normally
1708  * cheap.
1709  * Note that we do not need PANIC here, nobody will be able to use the
1710  * slot without fsyncing, and saving it won't succeed without an fsync()
1711  * either...
1712  * ----
1713  */
1714  fsync_fname(path, false);
1715  fsync_fname("pg_logical/snapshots", true);
1716 
1717 
1718  /* read statically sized portion of snapshot */
1720  readBytes = read(fd, &ondisk, SnapBuildOnDiskConstantSize);
1722  if (readBytes != SnapBuildOnDiskConstantSize)
1723  {
1724  int save_errno = errno;
1725 
1726  CloseTransientFile(fd);
1727 
1728  if (readBytes < 0)
1729  {
1730  errno = save_errno;
1731  ereport(ERROR,
1733  errmsg("could not read file \"%s\": %m", path)));
1734  }
1735  else
1736  ereport(ERROR,
1738  errmsg("could not read file \"%s\": read %d of %zu",
1739  path, readBytes,
1741  }
1742 
1743  if (ondisk.magic != SNAPBUILD_MAGIC)
1744  ereport(ERROR,
1746  errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1747  path, ondisk.magic, SNAPBUILD_MAGIC)));
1748 
1749  if (ondisk.version != SNAPBUILD_VERSION)
1750  ereport(ERROR,
1752  errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1753  path, ondisk.version, SNAPBUILD_VERSION)));
1754 
1755  INIT_CRC32C(checksum);
1756  COMP_CRC32C(checksum,
1757  ((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
1759 
1760  /* read SnapBuild */
1762  readBytes = read(fd, &ondisk.builder, sizeof(SnapBuild));
1764  if (readBytes != sizeof(SnapBuild))
1765  {
1766  int save_errno = errno;
1767 
1768  CloseTransientFile(fd);
1769 
1770  if (readBytes < 0)
1771  {
1772  errno = save_errno;
1773  ereport(ERROR,
1775  errmsg("could not read file \"%s\": %m", path)));
1776  }
1777  else
1778  ereport(ERROR,
1780  errmsg("could not read file \"%s\": read %d of %zu",
1781  path, readBytes, sizeof(SnapBuild))));
1782  }
1783  COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
1784 
1785  /* restore committed xacts information */
1786  sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
1787  ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
1789  readBytes = read(fd, ondisk.builder.committed.xip, sz);
1791  if (readBytes != sz)
1792  {
1793  int save_errno = errno;
1794 
1795  CloseTransientFile(fd);
1796 
1797  if (readBytes < 0)
1798  {
1799  errno = save_errno;
1800  ereport(ERROR,
1802  errmsg("could not read file \"%s\": %m", path)));
1803  }
1804  else
1805  ereport(ERROR,
1807  errmsg("could not read file \"%s\": read %d of %zu",
1808  path, readBytes, sz)));
1809  }
1810  COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
1811 
1812  if (CloseTransientFile(fd) != 0)
1813  ereport(ERROR,
1815  errmsg("could not close file \"%s\": %m", path)));
1816 
1817  FIN_CRC32C(checksum);
1818 
1819  /* verify checksum of what we've read */
1820  if (!EQ_CRC32C(checksum, ondisk.checksum))
1821  ereport(ERROR,
1823  errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1824  path, checksum, ondisk.checksum)));
1825 
1826  /*
1827  * ok, we now have a sensible snapshot here, figure out if it has more
1828  * information than we have.
1829  */
1830 
1831  /*
1832  * We are only interested in consistent snapshots for now, comparing
1833  * whether one incomplete snapshot is more "advanced" seems to be
1834  * unnecessarily complex.
1835  */
1836  if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
1837  goto snapshot_not_interesting;
1838 
1839  /*
1840  * Don't use a snapshot that requires an xmin that we cannot guarantee to
1841  * be available.
1842  */
1843  if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
1844  goto snapshot_not_interesting;
1845 
1846  /* consistent snapshots have no next phase */
1847  Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
1848 
1849  /* ok, we think the snapshot is sensible, copy over everything important */
1850  builder->xmin = ondisk.builder.xmin;
1851  builder->xmax = ondisk.builder.xmax;
1852  builder->state = ondisk.builder.state;
1853 
1854  builder->committed.xcnt = ondisk.builder.committed.xcnt;
1855  /* We only allocated/stored xcnt, not xcnt_space xids ! */
1856  /* don't overwrite preallocated xip, if we don't have anything here */
1857  if (builder->committed.xcnt > 0)
1858  {
1859  pfree(builder->committed.xip);
1860  builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1861  builder->committed.xip = ondisk.builder.committed.xip;
1862  }
1863  ondisk.builder.committed.xip = NULL;
1864 
1865  /* our snapshot is not interesting anymore, build a new one */
1866  if (builder->snapshot != NULL)
1867  {
1869  }
1870  builder->snapshot = SnapBuildBuildSnapshot(builder);
1872 
1873  ReorderBufferSetRestartPoint(builder->reorder, lsn);
1874 
1875  Assert(builder->state == SNAPBUILD_CONSISTENT);
1876 
1877  ereport(LOG,
1878  (errmsg("logical decoding found consistent point at %X/%X",
1879  LSN_FORMAT_ARGS(lsn)),
1880  errdetail("Logical decoding will begin using saved snapshot.")));
1881  return true;
1882 
1883 snapshot_not_interesting:
1884  if (ondisk.builder.committed.xip != NULL)
1885  pfree(ondisk.builder.committed.xip);
1886  return false;
1887 }
1888 
1889 /*
1890  * Remove all serialized snapshots that are not required anymore because no
1891  * slot can need them. This doesn't actually have to run during a checkpoint,
1892  * but it's a convenient point to schedule this.
1893  *
1894  * NB: We run this during checkpoints even if logical decoding is disabled so
1895  * we cleanup old slots at some point after it got disabled.
1896  */
1897 void
1899 {
1900  XLogRecPtr cutoff;
1901  XLogRecPtr redo;
1902  DIR *snap_dir;
1903  struct dirent *snap_de;
1904  char path[MAXPGPATH + 21];
1905 
1906  /*
1907  * We start off with a minimum of the last redo pointer. No new
1908  * replication slot will start before that, so that's a safe upper bound
1909  * for removal.
1910  */
1911  redo = GetRedoRecPtr();
1912 
1913  /* now check for the restart ptrs from existing slots */
1915 
1916  /* don't start earlier than the restart lsn */
1917  if (redo < cutoff)
1918  cutoff = redo;
1919 
1920  snap_dir = AllocateDir("pg_logical/snapshots");
1921  while ((snap_de = ReadDir(snap_dir, "pg_logical/snapshots")) != NULL)
1922  {
1923  uint32 hi;
1924  uint32 lo;
1925  XLogRecPtr lsn;
1926  struct stat statbuf;
1927 
1928  if (strcmp(snap_de->d_name, ".") == 0 ||
1929  strcmp(snap_de->d_name, "..") == 0)
1930  continue;
1931 
1932  snprintf(path, sizeof(path), "pg_logical/snapshots/%s", snap_de->d_name);
1933 
1934  if (lstat(path, &statbuf) == 0 && !S_ISREG(statbuf.st_mode))
1935  {
1936  elog(DEBUG1, "only regular files expected: %s", path);
1937  continue;
1938  }
1939 
1940  /*
1941  * temporary filenames from SnapBuildSerialize() include the LSN and
1942  * everything but are postfixed by .$pid.tmp. We can just remove them
1943  * the same as other files because there can be none that are
1944  * currently being written that are older than cutoff.
1945  *
1946  * We just log a message if a file doesn't fit the pattern, it's
1947  * probably some editors lock/state file or similar...
1948  */
1949  if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
1950  {
1951  ereport(LOG,
1952  (errmsg("could not parse file name \"%s\"", path)));
1953  continue;
1954  }
1955 
1956  lsn = ((uint64) hi) << 32 | lo;
1957 
1958  /* check whether we still need it */
1959  if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
1960  {
1961  elog(DEBUG1, "removing snapbuild snapshot %s", path);
1962 
1963  /*
1964  * It's not particularly harmful, though strange, if we can't
1965  * remove the file here. Don't prevent the checkpoint from
1966  * completing, that'd be a cure worse than the disease.
1967  */
1968  if (unlink(path) < 0)
1969  {
1970  ereport(LOG,
1972  errmsg("could not remove file \"%s\": %m",
1973  path)));
1974  continue;
1975  }
1976  }
1977  }
1978  FreeDir(snap_dir);
1979 }
#define TransactionIdAdvance(dest)
Definition: transam.h:91
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1681
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
uint32 CommandId
Definition: c.h:601
void AbortCurrentTransaction(void)
Definition: xact.c:3210
uint64 snapXactCompletionCount
Definition: snapshot.h:216
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2876
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:212
#define AllocSetContextCreate
Definition: memutils.h:170
#define rbtxn_prepared(txn)
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:41
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:842
#define SNAPBUILD_VERSION
Definition: snapbuild.c:1447
static void test(void)
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
Definition: transam.c:334
uint32 TransactionId
Definition: c.h:587
bool copied
Definition: snapshot.h:185
TransactionId next_phase_at
Definition: snapbuild.c:207
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
Definition: snapbuild.c:1373
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:385
XLogRecPtr initial_consistent_point
Definition: snapbuild.c:176
#define SNAPBUILD_MAGIC
Definition: snapbuild.c:1446
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:912
#define DEBUG3
Definition: elog.h:23
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:367
bool TransactionIdIsCurrentTransactionId(TransactionId xid)
Definition: xact.c:869
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:1015
#define write(a, b, c)
Definition: win32.h:14
PGPROC * MyProc
Definition: proc.c:68
XLogRecPtr SnapBuildInitialConsistentPoint(SnapBuild *builder)
Definition: snapbuild.c:376
CommandId combocid
Definition: heapam_xlog.h:366
pg_crc32c checksum
Definition: snapbuild.c:1426
Snapshot snapshot
Definition: snapbuild.c:190
#define dlist_foreach(iter, lhead)
Definition: ilist.h:507
uint32 pg_crc32c
Definition: pg_crc32c.h:38
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1591
ResourceOwner CurrentResourceOwner
Definition: resowner.c:144
XLogRecPtr current_restart_decoding_lsn
#define XACT_REPEATABLE_READ
Definition: xact.h:38
bool building_full_snapshot
Definition: snapbuild.c:185
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:661
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:349
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
int errcode(int sqlerrcode)
Definition: elog.c:694
CommandId cmax
Definition: heapam_xlog.h:365
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4701
size_t xcnt_space
Definition: snapbuild.c:219
ItemPointerData target_tid
Definition: heapam_xlog.h:372
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1106
bool suboverflowed
Definition: snapshot.h:182
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileNode node, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1181
size_t xcnt
Definition: snapbuild.c:216
#define LOG
Definition: elog.h:26
bool RecoveryInProgress(void)
Definition: xlog.c:8132
Definition: dirent.h:9
uint32 regd_count
Definition: snapshot.h:205
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1271
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:250
#define FirstCommandId
Definition: c.h:603
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1065
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:697
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1808
TransactionId xids[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:56
#define sprintf
Definition: port.h:218
const char * SnapBuildExportSnapshot(SnapBuild *builder)
Definition: snapbuild.c:613
SnapBuildState state
Definition: snapbuild.c:150
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1469
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define dlist_container(type, membername, ptr)
Definition: ilist.h:477
void pfree(void *pointer)
Definition: mcxt.c:1057
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1205
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:319
Definition: dirent.c:25
#define ERROR
Definition: elog.h:45
Definition: lmgr.h:26
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2404
bool FirstSnapshotSet
Definition: snapmgr.c:149
#define MAXPGPATH
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:192
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:789
static bool ExportInProgress
Definition: snapbuild.c:251
TransactionId xmin
Definition: proc.h:138
#define DEBUG2
Definition: elog.h:24
SnapBuild builder
Definition: snapbuild.c:1436
ReorderBuffer * reorder
Definition: snapbuild.c:200
TransactionId initial_xmin_horizon
Definition: snapbuild.c:182
TransactionId * xip
Definition: snapbuild.c:242
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:747
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:342
int errdetail(const char *fmt,...)
Definition: elog.c:1038
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:441
int errcode_for_file_access(void)
Definition: elog.c:717
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:877
struct SnapshotData SnapshotData
bool includes_all_transactions
Definition: snapbuild.c:226
#define InvalidTransactionId
Definition: transam.h:31
void CheckPointSnapBuild(void)
Definition: snapbuild.c:1898
SnapshotType snapshot_type
Definition: snapshot.h:144
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
unsigned int uint32
Definition: c.h:441
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2615
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1079
TransactionId xmax
Definition: snapshot.h:158
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1512
TransactionId xmin
Definition: snapshot.h:157
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
SnapBuildState
Definition: snapbuild.h:18
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:195
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
TransactionId * xip
Definition: snapshot.h:168
#define S_ISREG(m)
Definition: win32_port.h:319
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:45
int CloseTransientFile(int fd)
Definition: fd.c:2581
TransactionId xmax
Definition: snapbuild.c:159
RelFileNode target_node
Definition: heapam_xlog.h:371
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
void * palloc0(Size size)
Definition: mcxt.c:981
#define InvalidCommandId
Definition: c.h:604
unsigned short st_mode
Definition: win32_port.h:260
dlist_head toplevel_by_lsn
TransactionId xid
dlist_node * cur
Definition: ilist.h:161
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:840
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
CommandId curcid
Definition: snapshot.h:187
#define SnapBuildOnDiskConstantSize
Definition: snapbuild.c:1441
#define ereport(elevel,...)
Definition: elog.h:155
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:1998
bool XactReadOnly
Definition: xact.c:78
void FreeSnapshotBuilder(SnapBuild *builder)
Definition: snapbuild.c:323
int errmsg_internal(const char *fmt,...)
Definition: elog.c:992
#define Max(x, y)
Definition: c.h:980
void XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid, XLTW_Oper oper)
Definition: lmgr.c:639
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2681
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1534
void StartTransactionCommand(void)
Definition: xact.c:2838
#define SnapBuildOnDiskNotChecksummedSize
Definition: snapbuild.c:1443
bool takenDuringRecovery
Definition: snapshot.h:184
#define NormalTransactionIdFollows(id1, id2)
Definition: transam.h:152
size_t Size
Definition: c.h:540
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1488
int XactIsoLevel
Definition: xact.c:75
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1206
bool IsTransactionState(void)
Definition: xact.c:371
MemoryContext context
Definition: snapbuild.c:153
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8424
TransactionId nextXid
Definition: standbydefs.h:52
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1070
#define rbtxn_skip_prepared(txn)
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:409
#define lstat(path, sb)
Definition: win32_port.h:276
struct SnapBuild::@16 committed
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:521
uint32 xcnt
Definition: snapshot.h:169
void * palloc(Size size)
Definition: mcxt.c:950
int errmsg(const char *fmt,...)
Definition: elog.c:905
TransactionId xmin
Definition: snapbuild.c:156
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:797
#define elog(elevel,...)
Definition: elog.h:227
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1456
XLogRecPtr restart_decoding_lsn
TransactionId oldestRunningXid
Definition: standbydefs.h:53
CommandId cmin
Definition: heapam_xlog.h:364
int pg_fsync(int fd)
Definition: fd.c:347
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
char d_name[MAX_PATH]
Definition: dirent.h:15
struct SnapBuildOnDisk SnapBuildOnDisk
#define qsort(a, b, c, d)
Definition: port.h:504
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define TransactionIdIsNormal(xid)
Definition: transam.h:42
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:397
#define snprintf
Definition: port.h:216
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
XLogRecPtr start_decoding_at
Definition: snapbuild.c:165
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr initial_consistent_point)
Definition: snapbuild.c:280
static void SnapBuildPurgeCommittedTxn(SnapBuild *builder)
Definition: snapbuild.c:871
#define read(a, b, c)
Definition: win32.h:13
int FreeDir(DIR *dir)
Definition: fd.c:2733
#define stat
Definition: win32_port.h:275
void SnapBuildClearExportedSnapshot(void)
Definition: snapbuild.c:674
TransactionId * subxip
Definition: snapshot.h:180
uint32 active_count
Definition: snapshot.h:204
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:136
int32 subxcnt
Definition: snapshot.h:181
TransactionId top_xid
Definition: heapam_xlog.h:363
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:653