Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * twophase.c
4 : * Two-phase commit support functions.
5 : *
6 : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : * IDENTIFICATION
10 : * src/backend/access/transam/twophase.c
11 : *
12 : * NOTES
13 : * Each global transaction is associated with a global transaction
14 : * identifier (GID). The client assigns a GID to a postgres
15 : * transaction with the PREPARE TRANSACTION command.
16 : *
17 : * We keep all active global transactions in a shared memory array.
18 : * When the PREPARE TRANSACTION command is issued, the GID is
19 : * reserved for the transaction in the array. This is done before
20 : * a WAL entry is made, because the reservation checks for duplicate
21 : * GIDs and aborts the transaction if there already is a global
22 : * transaction in prepared state with the same GID.
23 : *
24 : * A global transaction (gxact) also has dummy PGPROC; this is what keeps
25 : * the XID considered running by TransactionIdIsInProgress. It is also
26 : * convenient as a PGPROC to hook the gxact's locks to.
27 : *
28 : * Information to recover prepared transactions in case of crash is
29 : * now stored in WAL for the common case. In some cases there will be
30 : * an extended period between preparing a GXACT and commit/abort, in
31 : * which case we need to separately record prepared transaction data
32 : * in permanent storage. This includes locking information, pending
33 : * notifications etc. All that state information is written to the
34 : * per-transaction state file in the pg_twophase directory.
35 : * All prepared transactions will be written prior to shutdown.
36 : *
37 : * Life track of state data is following:
38 : *
39 : * * On PREPARE TRANSACTION backend writes state data only to the WAL and
40 : * stores pointer to the start of the WAL record in
41 : * gxact->prepare_start_lsn.
42 : * * If COMMIT occurs before checkpoint then backend reads data from WAL
43 : * using prepare_start_lsn.
44 : * * On checkpoint state data copied to files in pg_twophase directory and
45 : * fsynced
46 : * * If COMMIT happens after checkpoint then backend reads state data from
47 : * files
48 : *
49 : * During replay and replication, TwoPhaseState also holds information
50 : * about active prepared transactions that haven't been moved to disk yet.
51 : *
52 : * Replay of twophase records happens by the following rules:
53 : *
54 : * * At the beginning of recovery, pg_twophase is scanned once, filling
55 : * TwoPhaseState with entries marked with gxact->inredo and
56 : * gxact->ondisk. Two-phase file data older than the XID horizon of
57 : * the redo position are discarded.
58 : * * On PREPARE redo, the transaction is added to TwoPhaseState->prepXacts.
59 : * gxact->inredo is set to true for such entries.
60 : * * On Checkpoint we iterate through TwoPhaseState->prepXacts entries
61 : * that have gxact->inredo set and are behind the redo_horizon. We
62 : * save them to disk and then switch gxact->ondisk to true.
63 : * * On COMMIT/ABORT we delete the entry from TwoPhaseState->prepXacts.
64 : * If gxact->ondisk is true, the corresponding entry from the disk
65 : * is additionally deleted.
66 : * * RecoverPreparedTransactions(), StandbyRecoverPreparedTransactions()
67 : * and PrescanPreparedTransactions() have been modified to go through
68 : * gxact->inredo entries that have not made it to disk.
69 : *
70 : *-------------------------------------------------------------------------
71 : */
72 : #include "postgres.h"
73 :
74 : #include <fcntl.h>
75 : #include <sys/stat.h>
76 : #include <time.h>
77 : #include <unistd.h>
78 :
79 : #include "access/commit_ts.h"
80 : #include "access/htup_details.h"
81 : #include "access/subtrans.h"
82 : #include "access/transam.h"
83 : #include "access/twophase.h"
84 : #include "access/twophase_rmgr.h"
85 : #include "access/xact.h"
86 : #include "access/xlog.h"
87 : #include "access/xloginsert.h"
88 : #include "access/xlogreader.h"
89 : #include "access/xlogrecovery.h"
90 : #include "access/xlogutils.h"
91 : #include "catalog/pg_type.h"
92 : #include "catalog/storage.h"
93 : #include "funcapi.h"
94 : #include "miscadmin.h"
95 : #include "pg_trace.h"
96 : #include "pgstat.h"
97 : #include "replication/origin.h"
98 : #include "replication/syncrep.h"
99 : #include "storage/fd.h"
100 : #include "storage/ipc.h"
101 : #include "storage/md.h"
102 : #include "storage/predicate.h"
103 : #include "storage/proc.h"
104 : #include "storage/procarray.h"
105 : #include "utils/builtins.h"
106 : #include "utils/memutils.h"
107 : #include "utils/timestamp.h"
108 :
109 : /*
110 : * Directory where Two-phase commit files reside within PGDATA
111 : */
112 : #define TWOPHASE_DIR "pg_twophase"
113 :
114 : /* GUC variable, can't be changed after startup */
115 : int max_prepared_xacts = 0;
116 :
117 : /*
118 : * This struct describes one global transaction that is in prepared state
119 : * or attempting to become prepared.
120 : *
121 : * The lifecycle of a global transaction is:
122 : *
123 : * 1. After checking that the requested GID is not in use, set up an entry in
124 : * the TwoPhaseState->prepXacts array with the correct GID and valid = false,
125 : * and mark it as locked by my backend.
126 : *
127 : * 2. After successfully completing prepare, set valid = true and enter the
128 : * referenced PGPROC into the global ProcArray.
129 : *
130 : * 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry is
131 : * valid and not locked, then mark the entry as locked by storing my current
132 : * proc number into locking_backend. This prevents concurrent attempts to
133 : * commit or rollback the same prepared xact.
134 : *
135 : * 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
136 : * from the ProcArray and the TwoPhaseState->prepXacts array and return it to
137 : * the freelist.
138 : *
139 : * Note that if the preparing transaction fails between steps 1 and 2, the
140 : * entry must be removed so that the GID and the GlobalTransaction struct
141 : * can be reused. See AtAbort_Twophase().
142 : *
143 : * typedef struct GlobalTransactionData *GlobalTransaction appears in
144 : * twophase.h
145 : */
146 :
147 : typedef struct GlobalTransactionData
148 : {
149 : GlobalTransaction next; /* list link for free list */
150 : int pgprocno; /* ID of associated dummy PGPROC */
151 : TimestampTz prepared_at; /* time of preparation */
152 :
153 : /*
154 : * Note that we need to keep track of two LSNs for each GXACT. We keep
155 : * track of the start LSN because this is the address we must use to read
156 : * state data back from WAL when committing a prepared GXACT. We keep
157 : * track of the end LSN because that is the LSN we need to wait for prior
158 : * to commit.
159 : */
160 : XLogRecPtr prepare_start_lsn; /* XLOG offset of prepare record start */
161 : XLogRecPtr prepare_end_lsn; /* XLOG offset of prepare record end */
162 : FullTransactionId fxid; /* The GXACT full xid */
163 :
164 : Oid owner; /* ID of user that executed the xact */
165 : ProcNumber locking_backend; /* backend currently working on the xact */
166 : bool valid; /* true if PGPROC entry is in proc array */
167 : bool ondisk; /* true if prepare state file is on disk */
168 : bool inredo; /* true if entry was added via xlog_redo */
169 : char gid[GIDSIZE]; /* The GID assigned to the prepared xact */
170 : } GlobalTransactionData;
171 :
172 : /*
173 : * Two Phase Commit shared state. Access to this struct is protected
174 : * by TwoPhaseStateLock.
175 : */
176 : typedef struct TwoPhaseStateData
177 : {
178 : /* Head of linked list of free GlobalTransactionData structs */
179 : GlobalTransaction freeGXacts;
180 :
181 : /* Number of valid prepXacts entries. */
182 : int numPrepXacts;
183 :
184 : /* There are max_prepared_xacts items in this array */
185 : GlobalTransaction prepXacts[FLEXIBLE_ARRAY_MEMBER];
186 : } TwoPhaseStateData;
187 :
188 : static TwoPhaseStateData *TwoPhaseState;
189 :
190 : /*
191 : * Global transaction entry currently locked by us, if any. Note that any
192 : * access to the entry pointed to by this variable must be protected by
193 : * TwoPhaseStateLock, though obviously the pointer itself doesn't need to be
194 : * (since it's just local memory).
195 : */
196 : static GlobalTransaction MyLockedGxact = NULL;
197 :
198 : static bool twophaseExitRegistered = false;
199 :
200 : static void PrepareRedoRemoveFull(FullTransactionId fxid, bool giveWarning);
201 : static void RecordTransactionCommitPrepared(TransactionId xid,
202 : int nchildren,
203 : TransactionId *children,
204 : int nrels,
205 : RelFileLocator *rels,
206 : int nstats,
207 : xl_xact_stats_item *stats,
208 : int ninvalmsgs,
209 : SharedInvalidationMessage *invalmsgs,
210 : bool initfileinval,
211 : const char *gid);
212 : static void RecordTransactionAbortPrepared(TransactionId xid,
213 : int nchildren,
214 : TransactionId *children,
215 : int nrels,
216 : RelFileLocator *rels,
217 : int nstats,
218 : xl_xact_stats_item *stats,
219 : const char *gid);
220 : static void ProcessRecords(char *bufptr, FullTransactionId fxid,
221 : const TwoPhaseCallback callbacks[]);
222 : static void RemoveGXact(GlobalTransaction gxact);
223 :
224 : static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len);
225 : static char *ProcessTwoPhaseBuffer(FullTransactionId fxid,
226 : XLogRecPtr prepare_start_lsn,
227 : bool fromdisk, bool setParent, bool setNextXid);
228 : static void MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
229 : const char *gid, TimestampTz prepared_at, Oid owner,
230 : Oid databaseid);
231 : static void RemoveTwoPhaseFile(FullTransactionId fxid, bool giveWarning);
232 : static void RecreateTwoPhaseFile(FullTransactionId fxid, void *content, int len);
233 :
234 : /*
235 : * Initialization of shared memory
236 : */
237 : Size
238 6150 : TwoPhaseShmemSize(void)
239 : {
240 : Size size;
241 :
242 : /* Need the fixed struct, the array of pointers, and the GTD structs */
243 6150 : size = offsetof(TwoPhaseStateData, prepXacts);
244 6150 : size = add_size(size, mul_size(max_prepared_xacts,
245 : sizeof(GlobalTransaction)));
246 6150 : size = MAXALIGN(size);
247 6150 : size = add_size(size, mul_size(max_prepared_xacts,
248 : sizeof(GlobalTransactionData)));
249 :
250 6150 : return size;
251 : }
252 :
253 : void
254 2152 : TwoPhaseShmemInit(void)
255 : {
256 : bool found;
257 :
258 2152 : TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
259 : TwoPhaseShmemSize(),
260 : &found);
261 2152 : if (!IsUnderPostmaster)
262 : {
263 : GlobalTransaction gxacts;
264 : int i;
265 :
266 : Assert(!found);
267 2152 : TwoPhaseState->freeGXacts = NULL;
268 2152 : TwoPhaseState->numPrepXacts = 0;
269 :
270 : /*
271 : * Initialize the linked list of free GlobalTransactionData structs
272 : */
273 2152 : gxacts = (GlobalTransaction)
274 2152 : ((char *) TwoPhaseState +
275 2152 : MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
276 : sizeof(GlobalTransaction) * max_prepared_xacts));
277 3838 : for (i = 0; i < max_prepared_xacts; i++)
278 : {
279 : /* insert into linked list */
280 1686 : gxacts[i].next = TwoPhaseState->freeGXacts;
281 1686 : TwoPhaseState->freeGXacts = &gxacts[i];
282 :
283 : /* associate it with a PGPROC assigned by InitProcGlobal */
284 1686 : gxacts[i].pgprocno = GetNumberFromPGProc(&PreparedXactProcs[i]);
285 : }
286 : }
287 : else
288 : Assert(found);
289 2152 : }
290 :
291 : /*
292 : * Exit hook to unlock the global transaction entry we're working on.
293 : */
294 : static void
295 268 : AtProcExit_Twophase(int code, Datum arg)
296 : {
297 : /* same logic as abort */
298 268 : AtAbort_Twophase();
299 268 : }
300 :
301 : /*
302 : * Abort hook to unlock the global transaction entry we're working on.
303 : */
304 : void
305 49594 : AtAbort_Twophase(void)
306 : {
307 49594 : if (MyLockedGxact == NULL)
308 49590 : return;
309 :
310 : /*
311 : * What to do with the locked global transaction entry? If we were in the
312 : * process of preparing the transaction, but haven't written the WAL
313 : * record and state file yet, the transaction must not be considered as
314 : * prepared. Likewise, if we are in the process of finishing an
315 : * already-prepared transaction, and fail after having already written the
316 : * 2nd phase commit or rollback record to the WAL, the transaction should
317 : * not be considered as prepared anymore. In those cases, just remove the
318 : * entry from shared memory.
319 : *
320 : * Otherwise, the entry must be left in place so that the transaction can
321 : * be finished later, so just unlock it.
322 : *
323 : * If we abort during prepare, after having written the WAL record, we
324 : * might not have transferred all locks and other state to the prepared
325 : * transaction yet. Likewise, if we abort during commit or rollback,
326 : * after having written the WAL record, we might not have released all the
327 : * resources held by the transaction yet. In those cases, the in-memory
328 : * state can be wrong, but it's too late to back out.
329 : */
330 4 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
331 4 : if (!MyLockedGxact->valid)
332 4 : RemoveGXact(MyLockedGxact);
333 : else
334 0 : MyLockedGxact->locking_backend = INVALID_PROC_NUMBER;
335 4 : LWLockRelease(TwoPhaseStateLock);
336 :
337 4 : MyLockedGxact = NULL;
338 : }
339 :
340 : /*
341 : * This is called after we have finished transferring state to the prepared
342 : * PGPROC entry.
343 : */
344 : void
345 636 : PostPrepare_Twophase(void)
346 : {
347 636 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
348 636 : MyLockedGxact->locking_backend = INVALID_PROC_NUMBER;
349 636 : LWLockRelease(TwoPhaseStateLock);
350 :
351 636 : MyLockedGxact = NULL;
352 636 : }
353 :
354 :
355 : /*
356 : * MarkAsPreparing
357 : * Reserve the GID for the given transaction.
358 : */
359 : GlobalTransaction
360 598 : MarkAsPreparing(FullTransactionId fxid, const char *gid,
361 : TimestampTz prepared_at, Oid owner, Oid databaseid)
362 : {
363 : GlobalTransaction gxact;
364 : int i;
365 :
366 598 : if (strlen(gid) >= GIDSIZE)
367 0 : ereport(ERROR,
368 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
369 : errmsg("transaction identifier \"%s\" is too long",
370 : gid)));
371 :
372 : /* fail immediately if feature is disabled */
373 598 : if (max_prepared_xacts == 0)
374 20 : ereport(ERROR,
375 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
376 : errmsg("prepared transactions are disabled"),
377 : errhint("Set \"max_prepared_transactions\" to a nonzero value.")));
378 :
379 : /* on first call, register the exit hook */
380 578 : if (!twophaseExitRegistered)
381 : {
382 148 : before_shmem_exit(AtProcExit_Twophase, 0);
383 148 : twophaseExitRegistered = true;
384 : }
385 :
386 578 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
387 :
388 : /* Check for conflicting GID */
389 1034 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
390 : {
391 460 : gxact = TwoPhaseState->prepXacts[i];
392 460 : if (strcmp(gxact->gid, gid) == 0)
393 : {
394 4 : ereport(ERROR,
395 : (errcode(ERRCODE_DUPLICATE_OBJECT),
396 : errmsg("transaction identifier \"%s\" is already in use",
397 : gid)));
398 : }
399 : }
400 :
401 : /* Get a free gxact from the freelist */
402 574 : if (TwoPhaseState->freeGXacts == NULL)
403 0 : ereport(ERROR,
404 : (errcode(ERRCODE_OUT_OF_MEMORY),
405 : errmsg("maximum number of prepared transactions reached"),
406 : errhint("Increase \"max_prepared_transactions\" (currently %d).",
407 : max_prepared_xacts)));
408 574 : gxact = TwoPhaseState->freeGXacts;
409 574 : TwoPhaseState->freeGXacts = gxact->next;
410 :
411 574 : MarkAsPreparingGuts(gxact, fxid, gid, prepared_at, owner, databaseid);
412 :
413 574 : gxact->ondisk = false;
414 :
415 : /* And insert it into the active array */
416 : Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
417 574 : TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
418 :
419 574 : LWLockRelease(TwoPhaseStateLock);
420 :
421 574 : return gxact;
422 : }
423 :
424 : /*
425 : * MarkAsPreparingGuts
426 : *
427 : * This uses a gxact struct and puts it into the active array.
428 : * NOTE: this is also used when reloading a gxact after a crash; so avoid
429 : * assuming that we can use very much backend context.
430 : *
431 : * Note: This function should be called with appropriate locks held.
432 : */
433 : static void
434 640 : MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
435 : const char *gid, TimestampTz prepared_at, Oid owner,
436 : Oid databaseid)
437 : {
438 : PGPROC *proc;
439 : int i;
440 640 : TransactionId xid = XidFromFullTransactionId(fxid);
441 :
442 : Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
443 :
444 : Assert(gxact != NULL);
445 640 : proc = GetPGProcByNumber(gxact->pgprocno);
446 :
447 : /* Initialize the PGPROC entry */
448 67200 : MemSet(proc, 0, sizeof(PGPROC));
449 640 : dlist_node_init(&proc->links);
450 640 : proc->waitStatus = PROC_WAIT_STATUS_OK;
451 640 : if (LocalTransactionIdIsValid(MyProc->vxid.lxid))
452 : {
453 : /* clone VXID, for TwoPhaseGetXidByVirtualXID() to find */
454 574 : proc->vxid.lxid = MyProc->vxid.lxid;
455 574 : proc->vxid.procNumber = MyProcNumber;
456 : }
457 : else
458 : {
459 : Assert(AmStartupProcess() || !IsPostmasterEnvironment);
460 : /* GetLockConflicts() uses this to specify a wait on the XID */
461 66 : proc->vxid.lxid = xid;
462 66 : proc->vxid.procNumber = INVALID_PROC_NUMBER;
463 : }
464 640 : proc->xid = xid;
465 : Assert(proc->xmin == InvalidTransactionId);
466 640 : proc->delayChkptFlags = 0;
467 640 : proc->statusFlags = 0;
468 640 : proc->pid = 0;
469 640 : proc->databaseId = databaseid;
470 640 : proc->roleId = owner;
471 640 : proc->tempNamespaceId = InvalidOid;
472 640 : proc->isRegularBackend = false;
473 640 : proc->lwWaiting = LW_WS_NOT_WAITING;
474 640 : proc->lwWaitMode = 0;
475 640 : proc->waitLock = NULL;
476 640 : proc->waitProcLock = NULL;
477 640 : pg_atomic_init_u64(&proc->waitStart, 0);
478 10880 : for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
479 10240 : dlist_init(&proc->myProcLocks[i]);
480 : /* subxid data must be filled later by GXactLoadSubxactData */
481 640 : proc->subxidStatus.overflowed = false;
482 640 : proc->subxidStatus.count = 0;
483 :
484 640 : gxact->prepared_at = prepared_at;
485 640 : gxact->fxid = fxid;
486 640 : gxact->owner = owner;
487 640 : gxact->locking_backend = MyProcNumber;
488 640 : gxact->valid = false;
489 640 : gxact->inredo = false;
490 640 : strcpy(gxact->gid, gid);
491 :
492 : /*
493 : * Remember that we have this GlobalTransaction entry locked for us. If we
494 : * abort after this, we must release it.
495 : */
496 640 : MyLockedGxact = gxact;
497 640 : }
498 :
499 : /*
500 : * GXactLoadSubxactData
501 : *
502 : * If the transaction being persisted had any subtransactions, this must
503 : * be called before MarkAsPrepared() to load information into the dummy
504 : * PGPROC.
505 : */
506 : static void
507 268 : GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
508 : TransactionId *children)
509 : {
510 268 : PGPROC *proc = GetPGProcByNumber(gxact->pgprocno);
511 :
512 : /* We need no extra lock since the GXACT isn't valid yet */
513 268 : if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
514 : {
515 8 : proc->subxidStatus.overflowed = true;
516 8 : nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
517 : }
518 268 : if (nsubxacts > 0)
519 : {
520 234 : memcpy(proc->subxids.xids, children,
521 : nsubxacts * sizeof(TransactionId));
522 234 : proc->subxidStatus.count = nsubxacts;
523 : }
524 268 : }
525 :
526 : /*
527 : * MarkAsPrepared
528 : * Mark the GXACT as fully valid, and enter it into the global ProcArray.
529 : *
530 : * lock_held indicates whether caller already holds TwoPhaseStateLock.
531 : */
532 : static void
533 636 : MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
534 : {
535 : /* Lock here may be overkill, but I'm not convinced of that ... */
536 636 : if (!lock_held)
537 570 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
538 : Assert(!gxact->valid);
539 636 : gxact->valid = true;
540 636 : if (!lock_held)
541 570 : LWLockRelease(TwoPhaseStateLock);
542 :
543 : /*
544 : * Put it into the global ProcArray so TransactionIdIsInProgress considers
545 : * the XID as still running.
546 : */
547 636 : ProcArrayAdd(GetPGProcByNumber(gxact->pgprocno));
548 636 : }
549 :
550 : /*
551 : * LockGXact
552 : * Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
553 : */
554 : static GlobalTransaction
555 590 : LockGXact(const char *gid, Oid user)
556 : {
557 : int i;
558 :
559 : /* on first call, register the exit hook */
560 590 : if (!twophaseExitRegistered)
561 : {
562 120 : before_shmem_exit(AtProcExit_Twophase, 0);
563 120 : twophaseExitRegistered = true;
564 : }
565 :
566 590 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
567 :
568 980 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
569 : {
570 968 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
571 968 : PGPROC *proc = GetPGProcByNumber(gxact->pgprocno);
572 :
573 : /* Ignore not-yet-valid GIDs */
574 968 : if (!gxact->valid)
575 6 : continue;
576 962 : if (strcmp(gxact->gid, gid) != 0)
577 384 : continue;
578 :
579 : /* Found it, but has someone else got it locked? */
580 578 : if (gxact->locking_backend != INVALID_PROC_NUMBER)
581 0 : ereport(ERROR,
582 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
583 : errmsg("prepared transaction with identifier \"%s\" is busy",
584 : gid)));
585 :
586 578 : if (user != gxact->owner && !superuser_arg(user))
587 0 : ereport(ERROR,
588 : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
589 : errmsg("permission denied to finish prepared transaction"),
590 : errhint("Must be superuser or the user that prepared the transaction.")));
591 :
592 : /*
593 : * Note: it probably would be possible to allow committing from
594 : * another database; but at the moment NOTIFY is known not to work and
595 : * there may be some other issues as well. Hence disallow until
596 : * someone gets motivated to make it work.
597 : */
598 578 : if (MyDatabaseId != proc->databaseId)
599 0 : ereport(ERROR,
600 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
601 : errmsg("prepared transaction belongs to another database"),
602 : errhint("Connect to the database where the transaction was prepared to finish it.")));
603 :
604 : /* OK for me to lock it */
605 578 : gxact->locking_backend = MyProcNumber;
606 578 : MyLockedGxact = gxact;
607 :
608 578 : LWLockRelease(TwoPhaseStateLock);
609 :
610 578 : return gxact;
611 : }
612 :
613 12 : LWLockRelease(TwoPhaseStateLock);
614 :
615 12 : ereport(ERROR,
616 : (errcode(ERRCODE_UNDEFINED_OBJECT),
617 : errmsg("prepared transaction with identifier \"%s\" does not exist",
618 : gid)));
619 :
620 : /* NOTREACHED */
621 : return NULL;
622 : }
623 :
624 : /*
625 : * RemoveGXact
626 : * Remove the prepared transaction from the shared memory array.
627 : *
628 : * NB: caller should have already removed it from ProcArray
629 : */
630 : static void
631 694 : RemoveGXact(GlobalTransaction gxact)
632 : {
633 : int i;
634 :
635 : Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
636 :
637 1082 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
638 : {
639 1082 : if (gxact == TwoPhaseState->prepXacts[i])
640 : {
641 : /* remove from the active array */
642 694 : TwoPhaseState->numPrepXacts--;
643 694 : TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
644 :
645 : /* and put it back in the freelist */
646 694 : gxact->next = TwoPhaseState->freeGXacts;
647 694 : TwoPhaseState->freeGXacts = gxact;
648 :
649 694 : return;
650 : }
651 : }
652 :
653 0 : elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
654 : }
655 :
656 : /*
657 : * Returns an array of all prepared transactions for the user-level
658 : * function pg_prepared_xact.
659 : *
660 : * The returned array and all its elements are copies of internal data
661 : * structures, to minimize the time we need to hold the TwoPhaseStateLock.
662 : *
663 : * WARNING -- we return even those transactions that are not fully prepared
664 : * yet. The caller should filter them out if he doesn't want them.
665 : *
666 : * The returned array is palloc'd.
667 : */
668 : static int
669 230 : GetPreparedTransactionList(GlobalTransaction *gxacts)
670 : {
671 : GlobalTransaction array;
672 : int num;
673 : int i;
674 :
675 230 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
676 :
677 230 : if (TwoPhaseState->numPrepXacts == 0)
678 : {
679 148 : LWLockRelease(TwoPhaseStateLock);
680 :
681 148 : *gxacts = NULL;
682 148 : return 0;
683 : }
684 :
685 82 : num = TwoPhaseState->numPrepXacts;
686 82 : array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
687 82 : *gxacts = array;
688 174 : for (i = 0; i < num; i++)
689 92 : memcpy(array + i, TwoPhaseState->prepXacts[i],
690 : sizeof(GlobalTransactionData));
691 :
692 82 : LWLockRelease(TwoPhaseStateLock);
693 :
694 82 : return num;
695 : }
696 :
697 :
698 : /* Working status for pg_prepared_xact */
699 : typedef struct
700 : {
701 : GlobalTransaction array;
702 : int ngxacts;
703 : int currIdx;
704 : } Working_State;
705 :
706 : /*
707 : * pg_prepared_xact
708 : * Produce a view with one row per prepared transaction.
709 : *
710 : * This function is here so we don't have to export the
711 : * GlobalTransactionData struct definition.
712 : */
713 : Datum
714 322 : pg_prepared_xact(PG_FUNCTION_ARGS)
715 : {
716 : FuncCallContext *funcctx;
717 : Working_State *status;
718 :
719 322 : if (SRF_IS_FIRSTCALL())
720 : {
721 : TupleDesc tupdesc;
722 : MemoryContext oldcontext;
723 :
724 : /* create a function context for cross-call persistence */
725 230 : funcctx = SRF_FIRSTCALL_INIT();
726 :
727 : /*
728 : * Switch to memory context appropriate for multiple function calls
729 : */
730 230 : oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
731 :
732 : /* build tupdesc for result tuples */
733 : /* this had better match pg_prepared_xacts view in system_views.sql */
734 230 : tupdesc = CreateTemplateTupleDesc(5);
735 230 : TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
736 : XIDOID, -1, 0);
737 230 : TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
738 : TEXTOID, -1, 0);
739 230 : TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
740 : TIMESTAMPTZOID, -1, 0);
741 230 : TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
742 : OIDOID, -1, 0);
743 230 : TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
744 : OIDOID, -1, 0);
745 :
746 230 : funcctx->tuple_desc = BlessTupleDesc(tupdesc);
747 :
748 : /*
749 : * Collect all the 2PC status information that we will format and send
750 : * out as a result set.
751 : */
752 230 : status = (Working_State *) palloc(sizeof(Working_State));
753 230 : funcctx->user_fctx = status;
754 :
755 230 : status->ngxacts = GetPreparedTransactionList(&status->array);
756 230 : status->currIdx = 0;
757 :
758 230 : MemoryContextSwitchTo(oldcontext);
759 : }
760 :
761 322 : funcctx = SRF_PERCALL_SETUP();
762 322 : status = (Working_State *) funcctx->user_fctx;
763 :
764 322 : while (status->array != NULL && status->currIdx < status->ngxacts)
765 : {
766 92 : GlobalTransaction gxact = &status->array[status->currIdx++];
767 92 : PGPROC *proc = GetPGProcByNumber(gxact->pgprocno);
768 92 : Datum values[5] = {0};
769 92 : bool nulls[5] = {0};
770 : HeapTuple tuple;
771 : Datum result;
772 :
773 92 : if (!gxact->valid)
774 0 : continue;
775 :
776 : /*
777 : * Form tuple with appropriate data.
778 : */
779 :
780 92 : values[0] = TransactionIdGetDatum(proc->xid);
781 92 : values[1] = CStringGetTextDatum(gxact->gid);
782 92 : values[2] = TimestampTzGetDatum(gxact->prepared_at);
783 92 : values[3] = ObjectIdGetDatum(gxact->owner);
784 92 : values[4] = ObjectIdGetDatum(proc->databaseId);
785 :
786 92 : tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
787 92 : result = HeapTupleGetDatum(tuple);
788 92 : SRF_RETURN_NEXT(funcctx, result);
789 : }
790 :
791 230 : SRF_RETURN_DONE(funcctx);
792 : }
793 :
794 : /*
795 : * TwoPhaseGetGXact
796 : * Get the GlobalTransaction struct for a prepared transaction
797 : * specified by XID
798 : *
799 : * If lock_held is set to true, TwoPhaseStateLock will not be taken, so the
800 : * caller had better hold it.
801 : */
802 : static GlobalTransaction
803 2376 : TwoPhaseGetGXact(FullTransactionId fxid, bool lock_held)
804 : {
805 2376 : GlobalTransaction result = NULL;
806 : int i;
807 :
808 : static FullTransactionId cached_fxid = {InvalidTransactionId};
809 : static GlobalTransaction cached_gxact = NULL;
810 :
811 : Assert(!lock_held || LWLockHeldByMe(TwoPhaseStateLock));
812 :
813 : /*
814 : * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
815 : * repeatedly for the same XID. We can save work with a simple cache.
816 : */
817 2376 : if (FullTransactionIdEquals(fxid, cached_fxid))
818 1602 : return cached_gxact;
819 :
820 774 : if (!lock_held)
821 636 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
822 :
823 1242 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
824 : {
825 1242 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
826 :
827 1242 : if (FullTransactionIdEquals(gxact->fxid, fxid))
828 : {
829 774 : result = gxact;
830 774 : break;
831 : }
832 : }
833 :
834 774 : if (!lock_held)
835 636 : LWLockRelease(TwoPhaseStateLock);
836 :
837 774 : if (result == NULL) /* should not happen */
838 0 : elog(ERROR, "failed to find GlobalTransaction for xid %u",
839 : XidFromFullTransactionId(fxid));
840 :
841 774 : cached_fxid = fxid;
842 774 : cached_gxact = result;
843 :
844 774 : return result;
845 : }
846 :
847 : /*
848 : * TwoPhaseGetXidByVirtualXID
849 : * Lookup VXID among xacts prepared since last startup.
850 : *
851 : * (This won't find recovered xacts.) If more than one matches, return any
852 : * and set "have_more" to true. To witness multiple matches, a single
853 : * proc number must consume 2^32 LXIDs, with no intervening database restart.
854 : */
855 : TransactionId
856 178 : TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid,
857 : bool *have_more)
858 : {
859 : int i;
860 178 : TransactionId result = InvalidTransactionId;
861 :
862 : Assert(VirtualTransactionIdIsValid(vxid));
863 178 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
864 :
865 286 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
866 : {
867 108 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
868 : PGPROC *proc;
869 : VirtualTransactionId proc_vxid;
870 :
871 108 : if (!gxact->valid)
872 0 : continue;
873 108 : proc = GetPGProcByNumber(gxact->pgprocno);
874 108 : GET_VXID_FROM_PGPROC(proc_vxid, *proc);
875 108 : if (VirtualTransactionIdEquals(vxid, proc_vxid))
876 : {
877 : /*
878 : * Startup process sets proc->vxid.procNumber to
879 : * INVALID_PROC_NUMBER.
880 : */
881 : Assert(!gxact->inredo);
882 :
883 34 : if (result != InvalidTransactionId)
884 : {
885 0 : *have_more = true;
886 0 : break;
887 : }
888 34 : result = XidFromFullTransactionId(gxact->fxid);
889 : }
890 : }
891 :
892 178 : LWLockRelease(TwoPhaseStateLock);
893 :
894 178 : return result;
895 : }
896 :
897 : /*
898 : * TwoPhaseGetDummyProcNumber
899 : * Get the dummy proc number for prepared transaction
900 : *
901 : * Dummy proc numbers are similar to proc numbers of real backends. They
902 : * start at MaxBackends, and are unique across all currently active real
903 : * backends and prepared transactions. If lock_held is set to true,
904 : * TwoPhaseStateLock will not be taken, so the caller had better hold it.
905 : */
906 : ProcNumber
907 220 : TwoPhaseGetDummyProcNumber(FullTransactionId fxid, bool lock_held)
908 : {
909 220 : GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
910 :
911 220 : return gxact->pgprocno;
912 : }
913 :
914 : /*
915 : * TwoPhaseGetDummyProc
916 : * Get the PGPROC that represents a prepared transaction
917 : *
918 : * If lock_held is set to true, TwoPhaseStateLock will not be taken, so the
919 : * caller had better hold it.
920 : */
921 : PGPROC *
922 2156 : TwoPhaseGetDummyProc(FullTransactionId fxid, bool lock_held)
923 : {
924 2156 : GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
925 :
926 2156 : return GetPGProcByNumber(gxact->pgprocno);
927 : }
928 :
929 : /************************************************************************/
930 : /* State file support */
931 : /************************************************************************/
932 :
933 : /*
934 : * Compute the FullTransactionId for the given TransactionId.
935 : *
936 : * This is safe if the xid has not yet reached COMMIT PREPARED or ROLLBACK
937 : * PREPARED. After those commands, concurrent vac_truncate_clog() may make
938 : * the xid cease to qualify as allowable. XXX Not all callers limit their
939 : * calls accordingly.
940 : */
941 : static inline FullTransactionId
942 646 : AdjustToFullTransactionId(TransactionId xid)
943 : {
944 : Assert(TransactionIdIsValid(xid));
945 646 : return FullTransactionIdFromAllowableAt(ReadNextFullTransactionId(), xid);
946 : }
947 :
948 : static inline int
949 1052 : TwoPhaseFilePath(char *path, FullTransactionId fxid)
950 : {
951 2104 : return snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X%08X",
952 1052 : EpochFromFullTransactionId(fxid),
953 1052 : XidFromFullTransactionId(fxid));
954 : }
955 :
956 : /*
957 : * 2PC state file format:
958 : *
959 : * 1. TwoPhaseFileHeader
960 : * 2. TransactionId[] (subtransactions)
961 : * 3. RelFileLocator[] (files to be deleted at commit)
962 : * 4. RelFileLocator[] (files to be deleted at abort)
963 : * 5. SharedInvalidationMessage[] (inval messages to be sent at commit)
964 : * 6. TwoPhaseRecordOnDisk
965 : * 7. ...
966 : * 8. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
967 : * 9. checksum (CRC-32C)
968 : *
969 : * Each segment except the final checksum is MAXALIGN'd.
970 : */
971 :
972 : /*
973 : * Header for a 2PC state file
974 : */
975 : #define TWOPHASE_MAGIC 0x57F94534 /* format identifier */
976 :
977 : typedef xl_xact_prepare TwoPhaseFileHeader;
978 :
979 : /*
980 : * Header for each record in a state file
981 : *
982 : * NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
983 : * The rmgr data will be stored starting on a MAXALIGN boundary.
984 : */
985 : typedef struct TwoPhaseRecordOnDisk
986 : {
987 : uint32 len; /* length of rmgr data */
988 : TwoPhaseRmgrId rmid; /* resource manager for this record */
989 : uint16 info; /* flag bits for use by rmgr */
990 : } TwoPhaseRecordOnDisk;
991 :
992 : /*
993 : * During prepare, the state file is assembled in memory before writing it
994 : * to WAL and the actual state file. We use a chain of StateFileChunk blocks
995 : * for that.
996 : */
997 : typedef struct StateFileChunk
998 : {
999 : char *data;
1000 : uint32 len;
1001 : struct StateFileChunk *next;
1002 : } StateFileChunk;
1003 :
1004 : static struct xllist
1005 : {
1006 : StateFileChunk *head; /* first data block in the chain */
1007 : StateFileChunk *tail; /* last block in chain */
1008 : uint32 num_chunks;
1009 : uint32 bytes_free; /* free bytes left in tail block */
1010 : uint32 total_len; /* total data bytes in chain */
1011 : } records;
1012 :
1013 :
1014 : /*
1015 : * Append a block of data to records data structure.
1016 : *
1017 : * NB: each block is padded to a MAXALIGN multiple. This must be
1018 : * accounted for when the file is later read!
1019 : *
1020 : * The data is copied, so the caller is free to modify it afterwards.
1021 : */
1022 : static void
1023 6490 : save_state_data(const void *data, uint32 len)
1024 : {
1025 6490 : uint32 padlen = MAXALIGN(len);
1026 :
1027 6490 : if (padlen > records.bytes_free)
1028 : {
1029 86 : records.tail->next = palloc0(sizeof(StateFileChunk));
1030 86 : records.tail = records.tail->next;
1031 86 : records.tail->len = 0;
1032 86 : records.tail->next = NULL;
1033 86 : records.num_chunks++;
1034 :
1035 86 : records.bytes_free = Max(padlen, 512);
1036 86 : records.tail->data = palloc(records.bytes_free);
1037 : }
1038 :
1039 6490 : memcpy(((char *) records.tail->data) + records.tail->len, data, len);
1040 6490 : records.tail->len += padlen;
1041 6490 : records.bytes_free -= padlen;
1042 6490 : records.total_len += padlen;
1043 6490 : }
1044 :
1045 : /*
1046 : * Start preparing a state file.
1047 : *
1048 : * Initializes data structure and inserts the 2PC file header record.
1049 : */
1050 : void
1051 574 : StartPrepare(GlobalTransaction gxact)
1052 : {
1053 574 : PGPROC *proc = GetPGProcByNumber(gxact->pgprocno);
1054 574 : TransactionId xid = XidFromFullTransactionId(gxact->fxid);
1055 : TwoPhaseFileHeader hdr;
1056 : TransactionId *children;
1057 : RelFileLocator *commitrels;
1058 : RelFileLocator *abortrels;
1059 574 : xl_xact_stats_item *abortstats = NULL;
1060 574 : xl_xact_stats_item *commitstats = NULL;
1061 : SharedInvalidationMessage *invalmsgs;
1062 :
1063 : /* Initialize linked list */
1064 574 : records.head = palloc0(sizeof(StateFileChunk));
1065 574 : records.head->len = 0;
1066 574 : records.head->next = NULL;
1067 :
1068 574 : records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
1069 574 : records.head->data = palloc(records.bytes_free);
1070 :
1071 574 : records.tail = records.head;
1072 574 : records.num_chunks = 1;
1073 :
1074 574 : records.total_len = 0;
1075 :
1076 : /* Create header */
1077 574 : hdr.magic = TWOPHASE_MAGIC;
1078 574 : hdr.total_len = 0; /* EndPrepare will fill this in */
1079 574 : hdr.xid = xid;
1080 574 : hdr.database = proc->databaseId;
1081 574 : hdr.prepared_at = gxact->prepared_at;
1082 574 : hdr.owner = gxact->owner;
1083 574 : hdr.nsubxacts = xactGetCommittedChildren(&children);
1084 574 : hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels);
1085 574 : hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
1086 574 : hdr.ncommitstats =
1087 574 : pgstat_get_transactional_drops(true, &commitstats);
1088 574 : hdr.nabortstats =
1089 574 : pgstat_get_transactional_drops(false, &abortstats);
1090 574 : hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
1091 : &hdr.initfileinval);
1092 574 : hdr.gidlen = strlen(gxact->gid) + 1; /* Include '\0' */
1093 : /* EndPrepare will fill the origin data, if necessary */
1094 574 : hdr.origin_lsn = InvalidXLogRecPtr;
1095 574 : hdr.origin_timestamp = 0;
1096 :
1097 574 : save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
1098 574 : save_state_data(gxact->gid, hdr.gidlen);
1099 :
1100 : /*
1101 : * Add the additional info about subxacts, deletable files and cache
1102 : * invalidation messages.
1103 : */
1104 574 : if (hdr.nsubxacts > 0)
1105 : {
1106 202 : save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
1107 : /* While we have the child-xact data, stuff it in the gxact too */
1108 202 : GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
1109 : }
1110 574 : if (hdr.ncommitrels > 0)
1111 : {
1112 18 : save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileLocator));
1113 18 : pfree(commitrels);
1114 : }
1115 574 : if (hdr.nabortrels > 0)
1116 : {
1117 34 : save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileLocator));
1118 34 : pfree(abortrels);
1119 : }
1120 574 : if (hdr.ncommitstats > 0)
1121 : {
1122 18 : save_state_data(commitstats,
1123 18 : hdr.ncommitstats * sizeof(xl_xact_stats_item));
1124 18 : pfree(commitstats);
1125 : }
1126 574 : if (hdr.nabortstats > 0)
1127 : {
1128 26 : save_state_data(abortstats,
1129 26 : hdr.nabortstats * sizeof(xl_xact_stats_item));
1130 26 : pfree(abortstats);
1131 : }
1132 574 : if (hdr.ninvalmsgs > 0)
1133 : {
1134 46 : save_state_data(invalmsgs,
1135 46 : hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
1136 46 : pfree(invalmsgs);
1137 : }
1138 574 : }
1139 :
1140 : /*
1141 : * Finish preparing state data and writing it to WAL.
1142 : */
1143 : void
1144 570 : EndPrepare(GlobalTransaction gxact)
1145 : {
1146 : TwoPhaseFileHeader *hdr;
1147 : StateFileChunk *record;
1148 : bool replorigin;
1149 :
1150 : /* Add the end sentinel to the list of 2PC records */
1151 570 : RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
1152 : NULL, 0);
1153 :
1154 : /* Go back and fill in total_len in the file header record */
1155 570 : hdr = (TwoPhaseFileHeader *) records.head->data;
1156 : Assert(hdr->magic == TWOPHASE_MAGIC);
1157 570 : hdr->total_len = records.total_len + sizeof(pg_crc32c);
1158 :
1159 618 : replorigin = (replorigin_session_origin != InvalidRepOriginId &&
1160 48 : replorigin_session_origin != DoNotReplicateId);
1161 :
1162 570 : if (replorigin)
1163 : {
1164 48 : hdr->origin_lsn = replorigin_session_origin_lsn;
1165 48 : hdr->origin_timestamp = replorigin_session_origin_timestamp;
1166 : }
1167 :
1168 : /*
1169 : * If the data size exceeds MaxAllocSize, we won't be able to read it in
1170 : * ReadTwoPhaseFile. Check for that now, rather than fail in the case
1171 : * where we write data to file and then re-read at commit time.
1172 : */
1173 570 : if (hdr->total_len > MaxAllocSize)
1174 0 : ereport(ERROR,
1175 : (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
1176 : errmsg("two-phase state file maximum length exceeded")));
1177 :
1178 : /*
1179 : * Now writing 2PC state data to WAL. We let the WAL's CRC protection
1180 : * cover us, so no need to calculate a separate CRC.
1181 : *
1182 : * We have to set DELAY_CHKPT_START here, too; otherwise a checkpoint
1183 : * starting immediately after the WAL record is inserted could complete
1184 : * without fsync'ing our state file. (This is essentially the same kind
1185 : * of race condition as the COMMIT-to-clog-write case that
1186 : * RecordTransactionCommit uses DELAY_CHKPT_IN_COMMIT for; see notes
1187 : * there.) Note that DELAY_CHKPT_IN_COMMIT is used to find transactions in
1188 : * the critical commit section. We need to know about such transactions
1189 : * for conflict detection in logical replication. See
1190 : * GetOldestActiveTransactionId(true, false) and its use.
1191 : *
1192 : * We save the PREPARE record's location in the gxact for later use by
1193 : * CheckPointTwoPhase.
1194 : */
1195 570 : XLogEnsureRecordSpace(0, records.num_chunks);
1196 :
1197 570 : START_CRIT_SECTION();
1198 :
1199 : Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
1200 570 : MyProc->delayChkptFlags |= DELAY_CHKPT_START;
1201 :
1202 570 : XLogBeginInsert();
1203 1226 : for (record = records.head; record != NULL; record = record->next)
1204 656 : XLogRegisterData(record->data, record->len);
1205 :
1206 570 : XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
1207 :
1208 570 : gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
1209 :
1210 570 : if (replorigin)
1211 : {
1212 : /* Move LSNs forward for this replication origin */
1213 48 : replorigin_session_advance(replorigin_session_origin_lsn,
1214 : gxact->prepare_end_lsn);
1215 : }
1216 :
1217 570 : XLogFlush(gxact->prepare_end_lsn);
1218 :
1219 : /* If we crash now, we have prepared: WAL replay will fix things */
1220 :
1221 : /* Store record's start location to read that later on Commit */
1222 570 : gxact->prepare_start_lsn = ProcLastRecPtr;
1223 :
1224 : /*
1225 : * Mark the prepared transaction as valid. As soon as xact.c marks MyProc
1226 : * as not running our XID (which it will do immediately after this
1227 : * function returns), others can commit/rollback the xact.
1228 : *
1229 : * NB: a side effect of this is to make a dummy ProcArray entry for the
1230 : * prepared XID. This must happen before we clear the XID from MyProc /
1231 : * ProcGlobal->xids[], else there is a window where the XID is not running
1232 : * according to TransactionIdIsInProgress, and onlookers would be entitled
1233 : * to assume the xact crashed. Instead we have a window where the same
1234 : * XID appears twice in ProcArray, which is OK.
1235 : */
1236 570 : MarkAsPrepared(gxact, false);
1237 :
1238 : /*
1239 : * Now we can mark ourselves as out of the commit critical section: a
1240 : * checkpoint starting after this will certainly see the gxact as a
1241 : * candidate for fsyncing.
1242 : */
1243 570 : MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
1244 :
1245 : /*
1246 : * Remember that we have this GlobalTransaction entry locked for us. If
1247 : * we crash after this point, it's too late to abort, but we must unlock
1248 : * it so that the prepared transaction can be committed or rolled back.
1249 : */
1250 570 : MyLockedGxact = gxact;
1251 :
1252 570 : END_CRIT_SECTION();
1253 :
1254 : /*
1255 : * Wait for synchronous replication, if required.
1256 : *
1257 : * Note that at this stage we have marked the prepare, but still show as
1258 : * running in the procarray (twice!) and continue to hold locks.
1259 : */
1260 570 : SyncRepWaitForLSN(gxact->prepare_end_lsn, false);
1261 :
1262 570 : records.tail = records.head = NULL;
1263 570 : records.num_chunks = 0;
1264 570 : }
1265 :
1266 : /*
1267 : * Register a 2PC record to be written to state file.
1268 : */
1269 : void
1270 2784 : RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
1271 : const void *data, uint32 len)
1272 : {
1273 : TwoPhaseRecordOnDisk record;
1274 :
1275 2784 : record.rmid = rmid;
1276 2784 : record.info = info;
1277 2784 : record.len = len;
1278 2784 : save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
1279 2784 : if (len > 0)
1280 2214 : save_state_data(data, len);
1281 2784 : }
1282 :
1283 :
1284 : /*
1285 : * Read and validate the state file for xid.
1286 : *
1287 : * If it looks OK (has a valid magic number and CRC), return the palloc'd
1288 : * contents of the file, issuing an error when finding corrupted data. If
1289 : * missing_ok is true, which indicates that missing files can be safely
1290 : * ignored, then return NULL. This state can be reached when doing recovery
1291 : * after discarding two-phase files from frozen epochs.
1292 : */
1293 : static char *
1294 788 : ReadTwoPhaseFile(FullTransactionId fxid, bool missing_ok)
1295 : {
1296 : char path[MAXPGPATH];
1297 : char *buf;
1298 : TwoPhaseFileHeader *hdr;
1299 : int fd;
1300 : struct stat stat;
1301 : uint32 crc_offset;
1302 : pg_crc32c calc_crc,
1303 : file_crc;
1304 : int r;
1305 :
1306 788 : TwoPhaseFilePath(path, fxid);
1307 :
1308 788 : fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1309 788 : if (fd < 0)
1310 : {
1311 646 : if (missing_ok && errno == ENOENT)
1312 646 : return NULL;
1313 :
1314 0 : ereport(ERROR,
1315 : (errcode_for_file_access(),
1316 : errmsg("could not open file \"%s\": %m", path)));
1317 : }
1318 :
1319 : /*
1320 : * Check file length. We can determine a lower bound pretty easily. We
1321 : * set an upper bound to avoid palloc() failure on a corrupt file, though
1322 : * we can't guarantee that we won't get an out of memory error anyway,
1323 : * even on a valid file.
1324 : */
1325 142 : if (fstat(fd, &stat))
1326 0 : ereport(ERROR,
1327 : (errcode_for_file_access(),
1328 : errmsg("could not stat file \"%s\": %m", path)));
1329 :
1330 142 : if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
1331 : MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
1332 142 : sizeof(pg_crc32c)) ||
1333 142 : stat.st_size > MaxAllocSize)
1334 0 : ereport(ERROR,
1335 : (errcode(ERRCODE_DATA_CORRUPTED),
1336 : errmsg_plural("incorrect size of file \"%s\": %lld byte",
1337 : "incorrect size of file \"%s\": %lld bytes",
1338 : (long long int) stat.st_size, path,
1339 : (long long int) stat.st_size)));
1340 :
1341 142 : crc_offset = stat.st_size - sizeof(pg_crc32c);
1342 142 : if (crc_offset != MAXALIGN(crc_offset))
1343 0 : ereport(ERROR,
1344 : (errcode(ERRCODE_DATA_CORRUPTED),
1345 : errmsg("incorrect alignment of CRC offset for file \"%s\"",
1346 : path)));
1347 :
1348 : /*
1349 : * OK, slurp in the file.
1350 : */
1351 142 : buf = (char *) palloc(stat.st_size);
1352 :
1353 142 : pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_READ);
1354 142 : r = read(fd, buf, stat.st_size);
1355 142 : if (r != stat.st_size)
1356 : {
1357 0 : if (r < 0)
1358 0 : ereport(ERROR,
1359 : (errcode_for_file_access(),
1360 : errmsg("could not read file \"%s\": %m", path)));
1361 : else
1362 0 : ereport(ERROR,
1363 : (errmsg("could not read file \"%s\": read %d of %lld",
1364 : path, r, (long long int) stat.st_size)));
1365 : }
1366 :
1367 142 : pgstat_report_wait_end();
1368 :
1369 142 : if (CloseTransientFile(fd) != 0)
1370 0 : ereport(ERROR,
1371 : (errcode_for_file_access(),
1372 : errmsg("could not close file \"%s\": %m", path)));
1373 :
1374 142 : hdr = (TwoPhaseFileHeader *) buf;
1375 142 : if (hdr->magic != TWOPHASE_MAGIC)
1376 0 : ereport(ERROR,
1377 : (errcode(ERRCODE_DATA_CORRUPTED),
1378 : errmsg("invalid magic number stored in file \"%s\"",
1379 : path)));
1380 :
1381 142 : if (hdr->total_len != stat.st_size)
1382 0 : ereport(ERROR,
1383 : (errcode(ERRCODE_DATA_CORRUPTED),
1384 : errmsg("invalid size stored in file \"%s\"",
1385 : path)));
1386 :
1387 142 : INIT_CRC32C(calc_crc);
1388 142 : COMP_CRC32C(calc_crc, buf, crc_offset);
1389 142 : FIN_CRC32C(calc_crc);
1390 :
1391 142 : file_crc = *((pg_crc32c *) (buf + crc_offset));
1392 :
1393 142 : if (!EQ_CRC32C(calc_crc, file_crc))
1394 0 : ereport(ERROR,
1395 : (errcode(ERRCODE_DATA_CORRUPTED),
1396 : errmsg("calculated CRC checksum does not match value stored in file \"%s\"",
1397 : path)));
1398 :
1399 142 : return buf;
1400 : }
1401 :
1402 :
1403 : /*
1404 : * Reads 2PC data from xlog. During checkpoint this data will be moved to
1405 : * twophase files and ReadTwoPhaseFile should be used instead.
1406 : *
1407 : * Note clearly that this function can access WAL during normal operation,
1408 : * similarly to the way WALSender or Logical Decoding would do.
1409 : */
1410 : static void
1411 738 : XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
1412 : {
1413 : XLogRecord *record;
1414 : XLogReaderState *xlogreader;
1415 : char *errormsg;
1416 :
1417 738 : xlogreader = XLogReaderAllocate(wal_segment_size, NULL,
1418 738 : XL_ROUTINE(.page_read = &read_local_xlog_page,
1419 : .segment_open = &wal_segment_open,
1420 : .segment_close = &wal_segment_close),
1421 : NULL);
1422 738 : if (!xlogreader)
1423 0 : ereport(ERROR,
1424 : (errcode(ERRCODE_OUT_OF_MEMORY),
1425 : errmsg("out of memory"),
1426 : errdetail("Failed while allocating a WAL reading processor.")));
1427 :
1428 738 : XLogBeginRead(xlogreader, lsn);
1429 738 : record = XLogReadRecord(xlogreader, &errormsg);
1430 :
1431 738 : if (record == NULL)
1432 : {
1433 0 : if (errormsg)
1434 0 : ereport(ERROR,
1435 : (errcode_for_file_access(),
1436 : errmsg("could not read two-phase state from WAL at %X/%08X: %s",
1437 : LSN_FORMAT_ARGS(lsn), errormsg)));
1438 : else
1439 0 : ereport(ERROR,
1440 : (errcode_for_file_access(),
1441 : errmsg("could not read two-phase state from WAL at %X/%08X",
1442 : LSN_FORMAT_ARGS(lsn))));
1443 : }
1444 :
1445 738 : if (XLogRecGetRmid(xlogreader) != RM_XACT_ID ||
1446 738 : (XLogRecGetInfo(xlogreader) & XLOG_XACT_OPMASK) != XLOG_XACT_PREPARE)
1447 0 : ereport(ERROR,
1448 : (errcode_for_file_access(),
1449 : errmsg("expected two-phase state data is not present in WAL at %X/%08X",
1450 : LSN_FORMAT_ARGS(lsn))));
1451 :
1452 738 : if (len != NULL)
1453 54 : *len = XLogRecGetDataLen(xlogreader);
1454 :
1455 738 : *buf = palloc(sizeof(char) * XLogRecGetDataLen(xlogreader));
1456 738 : memcpy(*buf, XLogRecGetData(xlogreader), sizeof(char) * XLogRecGetDataLen(xlogreader));
1457 :
1458 738 : XLogReaderFree(xlogreader);
1459 738 : }
1460 :
1461 :
1462 : /*
1463 : * Confirms an xid is prepared, during recovery
1464 : */
1465 : bool
1466 646 : StandbyTransactionIdIsPrepared(TransactionId xid)
1467 : {
1468 : char *buf;
1469 : TwoPhaseFileHeader *hdr;
1470 : bool result;
1471 : FullTransactionId fxid;
1472 :
1473 : Assert(TransactionIdIsValid(xid));
1474 :
1475 646 : if (max_prepared_xacts <= 0)
1476 0 : return false; /* nothing to do */
1477 :
1478 : /* Read and validate file */
1479 646 : fxid = AdjustToFullTransactionId(xid);
1480 646 : buf = ReadTwoPhaseFile(fxid, true);
1481 646 : if (buf == NULL)
1482 646 : return false;
1483 :
1484 : /* Check header also */
1485 0 : hdr = (TwoPhaseFileHeader *) buf;
1486 0 : result = TransactionIdEquals(hdr->xid, xid);
1487 0 : pfree(buf);
1488 :
1489 0 : return result;
1490 : }
1491 :
1492 : /*
1493 : * FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
1494 : */
1495 : void
1496 590 : FinishPreparedTransaction(const char *gid, bool isCommit)
1497 : {
1498 : GlobalTransaction gxact;
1499 : PGPROC *proc;
1500 : FullTransactionId fxid;
1501 : TransactionId xid;
1502 : bool ondisk;
1503 : char *buf;
1504 : char *bufptr;
1505 : TwoPhaseFileHeader *hdr;
1506 : TransactionId latestXid;
1507 : TransactionId *children;
1508 : RelFileLocator *commitrels;
1509 : RelFileLocator *abortrels;
1510 : RelFileLocator *delrels;
1511 : int ndelrels;
1512 : xl_xact_stats_item *commitstats;
1513 : xl_xact_stats_item *abortstats;
1514 : SharedInvalidationMessage *invalmsgs;
1515 :
1516 : /*
1517 : * Validate the GID, and lock the GXACT to ensure that two backends do not
1518 : * try to commit the same GID at once.
1519 : */
1520 590 : gxact = LockGXact(gid, GetUserId());
1521 578 : proc = GetPGProcByNumber(gxact->pgprocno);
1522 578 : fxid = gxact->fxid;
1523 578 : xid = XidFromFullTransactionId(fxid);
1524 :
1525 : /*
1526 : * Read and validate 2PC state data. State data will typically be stored
1527 : * in WAL files if the LSN is after the last checkpoint record, or moved
1528 : * to disk if for some reason they have lived for a long time.
1529 : */
1530 578 : if (gxact->ondisk)
1531 48 : buf = ReadTwoPhaseFile(fxid, false);
1532 : else
1533 530 : XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
1534 :
1535 :
1536 : /*
1537 : * Disassemble the header area
1538 : */
1539 578 : hdr = (TwoPhaseFileHeader *) buf;
1540 : Assert(TransactionIdEquals(hdr->xid, xid));
1541 578 : bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
1542 578 : bufptr += MAXALIGN(hdr->gidlen);
1543 578 : children = (TransactionId *) bufptr;
1544 578 : bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
1545 578 : commitrels = (RelFileLocator *) bufptr;
1546 578 : bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileLocator));
1547 578 : abortrels = (RelFileLocator *) bufptr;
1548 578 : bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileLocator));
1549 578 : commitstats = (xl_xact_stats_item *) bufptr;
1550 578 : bufptr += MAXALIGN(hdr->ncommitstats * sizeof(xl_xact_stats_item));
1551 578 : abortstats = (xl_xact_stats_item *) bufptr;
1552 578 : bufptr += MAXALIGN(hdr->nabortstats * sizeof(xl_xact_stats_item));
1553 578 : invalmsgs = (SharedInvalidationMessage *) bufptr;
1554 578 : bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
1555 :
1556 : /* compute latestXid among all children */
1557 578 : latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
1558 :
1559 : /* Prevent cancel/die interrupt while cleaning up */
1560 578 : HOLD_INTERRUPTS();
1561 :
1562 : /*
1563 : * The order of operations here is critical: make the XLOG entry for
1564 : * commit or abort, then mark the transaction committed or aborted in
1565 : * pg_xact, then remove its PGPROC from the global ProcArray (which means
1566 : * TransactionIdIsInProgress will stop saying the prepared xact is in
1567 : * progress), then run the post-commit or post-abort callbacks. The
1568 : * callbacks will release the locks the transaction held.
1569 : */
1570 578 : if (isCommit)
1571 502 : RecordTransactionCommitPrepared(xid,
1572 : hdr->nsubxacts, children,
1573 : hdr->ncommitrels, commitrels,
1574 : hdr->ncommitstats,
1575 : commitstats,
1576 : hdr->ninvalmsgs, invalmsgs,
1577 502 : hdr->initfileinval, gid);
1578 : else
1579 76 : RecordTransactionAbortPrepared(xid,
1580 : hdr->nsubxacts, children,
1581 : hdr->nabortrels, abortrels,
1582 : hdr->nabortstats,
1583 : abortstats,
1584 : gid);
1585 :
1586 578 : ProcArrayRemove(proc, latestXid);
1587 :
1588 : /*
1589 : * In case we fail while running the callbacks, mark the gxact invalid so
1590 : * no one else will try to commit/rollback, and so it will be recycled if
1591 : * we fail after this point. It is still locked by our backend so it
1592 : * won't go away yet.
1593 : *
1594 : * (We assume it's safe to do this without taking TwoPhaseStateLock.)
1595 : */
1596 578 : gxact->valid = false;
1597 :
1598 : /*
1599 : * We have to remove any files that were supposed to be dropped. For
1600 : * consistency with the regular xact.c code paths, must do this before
1601 : * releasing locks, so do it before running the callbacks.
1602 : *
1603 : * NB: this code knows that we couldn't be dropping any temp rels ...
1604 : */
1605 578 : if (isCommit)
1606 : {
1607 502 : delrels = commitrels;
1608 502 : ndelrels = hdr->ncommitrels;
1609 : }
1610 : else
1611 : {
1612 76 : delrels = abortrels;
1613 76 : ndelrels = hdr->nabortrels;
1614 : }
1615 :
1616 : /* Make sure files supposed to be dropped are dropped */
1617 578 : DropRelationFiles(delrels, ndelrels, false);
1618 :
1619 578 : if (isCommit)
1620 502 : pgstat_execute_transactional_drops(hdr->ncommitstats, commitstats, false);
1621 : else
1622 76 : pgstat_execute_transactional_drops(hdr->nabortstats, abortstats, false);
1623 :
1624 : /*
1625 : * Handle cache invalidation messages.
1626 : *
1627 : * Relcache init file invalidation requires processing both before and
1628 : * after we send the SI messages, only when committing. See
1629 : * AtEOXact_Inval().
1630 : */
1631 578 : if (isCommit)
1632 : {
1633 502 : if (hdr->initfileinval)
1634 0 : RelationCacheInitFilePreInvalidate();
1635 502 : SendSharedInvalidMessages(invalmsgs, hdr->ninvalmsgs);
1636 502 : if (hdr->initfileinval)
1637 0 : RelationCacheInitFilePostInvalidate();
1638 : }
1639 :
1640 : /*
1641 : * Acquire the two-phase lock. We want to work on the two-phase callbacks
1642 : * while holding it to avoid potential conflicts with other transactions
1643 : * attempting to use the same GID, so the lock is released once the shared
1644 : * memory state is cleared.
1645 : */
1646 578 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
1647 :
1648 : /* And now do the callbacks */
1649 578 : if (isCommit)
1650 502 : ProcessRecords(bufptr, fxid, twophase_postcommit_callbacks);
1651 : else
1652 76 : ProcessRecords(bufptr, fxid, twophase_postabort_callbacks);
1653 :
1654 578 : PredicateLockTwoPhaseFinish(fxid, isCommit);
1655 :
1656 : /*
1657 : * Read this value while holding the two-phase lock, as the on-disk 2PC
1658 : * file is physically removed after the lock is released.
1659 : */
1660 578 : ondisk = gxact->ondisk;
1661 :
1662 : /* Clear shared memory state */
1663 578 : RemoveGXact(gxact);
1664 :
1665 : /*
1666 : * Release the lock as all callbacks are called and shared memory cleanup
1667 : * is done.
1668 : */
1669 578 : LWLockRelease(TwoPhaseStateLock);
1670 :
1671 : /* Count the prepared xact as committed or aborted */
1672 578 : AtEOXact_PgStat(isCommit, false);
1673 :
1674 : /*
1675 : * And now we can clean up any files we may have left.
1676 : */
1677 578 : if (ondisk)
1678 50 : RemoveTwoPhaseFile(fxid, true);
1679 :
1680 578 : MyLockedGxact = NULL;
1681 :
1682 578 : RESUME_INTERRUPTS();
1683 :
1684 578 : pfree(buf);
1685 578 : }
1686 :
1687 : /*
1688 : * Scan 2PC state data in memory and call the indicated callbacks for each 2PC record.
1689 : */
1690 : static void
1691 644 : ProcessRecords(char *bufptr, FullTransactionId fxid,
1692 : const TwoPhaseCallback callbacks[])
1693 : {
1694 : for (;;)
1695 2588 : {
1696 3232 : TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;
1697 :
1698 : Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
1699 3232 : if (record->rmid == TWOPHASE_RM_END_ID)
1700 644 : break;
1701 :
1702 2588 : bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));
1703 :
1704 2588 : if (callbacks[record->rmid] != NULL)
1705 2430 : callbacks[record->rmid] (fxid, record->info, bufptr, record->len);
1706 :
1707 2588 : bufptr += MAXALIGN(record->len);
1708 : }
1709 644 : }
1710 :
1711 : /*
1712 : * Remove the 2PC file.
1713 : *
1714 : * If giveWarning is false, do not complain about file-not-present;
1715 : * this is an expected case during WAL replay.
1716 : *
1717 : * This routine is used at early stages at recovery where future and
1718 : * past orphaned files are checked, hence the FullTransactionId to build
1719 : * a complete file name fit for the removal.
1720 : */
1721 : static void
1722 58 : RemoveTwoPhaseFile(FullTransactionId fxid, bool giveWarning)
1723 : {
1724 : char path[MAXPGPATH];
1725 :
1726 58 : TwoPhaseFilePath(path, fxid);
1727 58 : if (unlink(path))
1728 0 : if (errno != ENOENT || giveWarning)
1729 0 : ereport(WARNING,
1730 : (errcode_for_file_access(),
1731 : errmsg("could not remove file \"%s\": %m", path)));
1732 58 : }
1733 :
1734 : /*
1735 : * Recreates a state file. This is used in WAL replay and during
1736 : * checkpoint creation.
1737 : *
1738 : * Note: content and len don't include CRC.
1739 : */
1740 : static void
1741 54 : RecreateTwoPhaseFile(FullTransactionId fxid, void *content, int len)
1742 : {
1743 : char path[MAXPGPATH];
1744 : pg_crc32c statefile_crc;
1745 : int fd;
1746 :
1747 : /* Recompute CRC */
1748 54 : INIT_CRC32C(statefile_crc);
1749 54 : COMP_CRC32C(statefile_crc, content, len);
1750 54 : FIN_CRC32C(statefile_crc);
1751 :
1752 54 : TwoPhaseFilePath(path, fxid);
1753 :
1754 54 : fd = OpenTransientFile(path,
1755 : O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY);
1756 54 : if (fd < 0)
1757 0 : ereport(ERROR,
1758 : (errcode_for_file_access(),
1759 : errmsg("could not recreate file \"%s\": %m", path)));
1760 :
1761 : /* Write content and CRC */
1762 54 : errno = 0;
1763 54 : pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_WRITE);
1764 54 : if (write(fd, content, len) != len)
1765 : {
1766 : /* if write didn't set errno, assume problem is no disk space */
1767 0 : if (errno == 0)
1768 0 : errno = ENOSPC;
1769 0 : ereport(ERROR,
1770 : (errcode_for_file_access(),
1771 : errmsg("could not write file \"%s\": %m", path)));
1772 : }
1773 54 : if (write(fd, &statefile_crc, sizeof(pg_crc32c)) != sizeof(pg_crc32c))
1774 : {
1775 : /* if write didn't set errno, assume problem is no disk space */
1776 0 : if (errno == 0)
1777 0 : errno = ENOSPC;
1778 0 : ereport(ERROR,
1779 : (errcode_for_file_access(),
1780 : errmsg("could not write file \"%s\": %m", path)));
1781 : }
1782 54 : pgstat_report_wait_end();
1783 :
1784 : /*
1785 : * We must fsync the file because the end-of-replay checkpoint will not do
1786 : * so, there being no GXACT in shared memory yet to tell it to.
1787 : */
1788 54 : pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_SYNC);
1789 54 : if (pg_fsync(fd) != 0)
1790 0 : ereport(ERROR,
1791 : (errcode_for_file_access(),
1792 : errmsg("could not fsync file \"%s\": %m", path)));
1793 54 : pgstat_report_wait_end();
1794 :
1795 54 : if (CloseTransientFile(fd) != 0)
1796 0 : ereport(ERROR,
1797 : (errcode_for_file_access(),
1798 : errmsg("could not close file \"%s\": %m", path)));
1799 54 : }
1800 :
1801 : /*
1802 : * CheckPointTwoPhase -- handle 2PC component of checkpointing.
1803 : *
1804 : * We must fsync the state file of any GXACT that is valid or has been
1805 : * generated during redo and has a PREPARE LSN <= the checkpoint's redo
1806 : * horizon. (If the gxact isn't valid yet, has not been generated in
1807 : * redo, or has a later LSN, this checkpoint is not responsible for
1808 : * fsyncing it.)
1809 : *
1810 : * This is deliberately run as late as possible in the checkpoint sequence,
1811 : * because GXACTs ordinarily have short lifespans, and so it is quite
1812 : * possible that GXACTs that were valid at checkpoint start will no longer
1813 : * exist if we wait a little bit. With typical checkpoint settings this
1814 : * will be about 3 minutes for an online checkpoint, so as a result we
1815 : * expect that there will be no GXACTs that need to be copied to disk.
1816 : *
1817 : * If a GXACT remains valid across multiple checkpoints, it will already
1818 : * be on disk so we don't bother to repeat that write.
1819 : */
1820 : void
1821 3380 : CheckPointTwoPhase(XLogRecPtr redo_horizon)
1822 : {
1823 : int i;
1824 3380 : int serialized_xacts = 0;
1825 :
1826 3380 : if (max_prepared_xacts <= 0)
1827 2360 : return; /* nothing to do */
1828 :
1829 : TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();
1830 :
1831 : /*
1832 : * We are expecting there to be zero GXACTs that need to be copied to
1833 : * disk, so we perform all I/O while holding TwoPhaseStateLock for
1834 : * simplicity. This prevents any new xacts from preparing while this
1835 : * occurs, which shouldn't be a problem since the presence of long-lived
1836 : * prepared xacts indicates the transaction manager isn't active.
1837 : *
1838 : * It's also possible to move I/O out of the lock, but on every error we
1839 : * should check whether somebody committed our transaction in different
1840 : * backend. Let's leave this optimization for future, if somebody will
1841 : * spot that this place cause bottleneck.
1842 : *
1843 : * Note that it isn't possible for there to be a GXACT with a
1844 : * prepare_end_lsn set prior to the last checkpoint yet is marked invalid,
1845 : * because of the efforts with delayChkptFlags.
1846 : */
1847 1020 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
1848 1086 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
1849 : {
1850 : /*
1851 : * Note that we are using gxact not PGPROC so this works in recovery
1852 : * also
1853 : */
1854 66 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
1855 :
1856 66 : if ((gxact->valid || gxact->inredo) &&
1857 66 : !gxact->ondisk &&
1858 58 : gxact->prepare_end_lsn <= redo_horizon)
1859 : {
1860 : char *buf;
1861 : int len;
1862 :
1863 54 : XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len);
1864 54 : RecreateTwoPhaseFile(gxact->fxid, buf, len);
1865 54 : gxact->ondisk = true;
1866 54 : gxact->prepare_start_lsn = InvalidXLogRecPtr;
1867 54 : gxact->prepare_end_lsn = InvalidXLogRecPtr;
1868 54 : pfree(buf);
1869 54 : serialized_xacts++;
1870 : }
1871 : }
1872 1020 : LWLockRelease(TwoPhaseStateLock);
1873 :
1874 : /*
1875 : * Flush unconditionally the parent directory to make any information
1876 : * durable on disk. Two-phase files could have been removed and those
1877 : * removals need to be made persistent as well as any files newly created
1878 : * previously since the last checkpoint.
1879 : */
1880 1020 : fsync_fname(TWOPHASE_DIR, true);
1881 :
1882 : TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
1883 :
1884 1020 : if (log_checkpoints && serialized_xacts > 0)
1885 46 : ereport(LOG,
1886 : (errmsg_plural("%u two-phase state file was written "
1887 : "for a long-running prepared transaction",
1888 : "%u two-phase state files were written "
1889 : "for long-running prepared transactions",
1890 : serialized_xacts,
1891 : serialized_xacts)));
1892 : }
1893 :
1894 : /*
1895 : * restoreTwoPhaseData
1896 : *
1897 : * Scan pg_twophase and fill TwoPhaseState depending on the on-disk data.
1898 : * This is called once at the beginning of recovery, saving any extra
1899 : * lookups in the future. Two-phase files that are newer than the
1900 : * minimum XID horizon are discarded on the way.
1901 : */
1902 : void
1903 1862 : restoreTwoPhaseData(void)
1904 : {
1905 : DIR *cldir;
1906 : struct dirent *clde;
1907 :
1908 1862 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
1909 1862 : cldir = AllocateDir(TWOPHASE_DIR);
1910 5616 : while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
1911 : {
1912 3754 : if (strlen(clde->d_name) == 16 &&
1913 30 : strspn(clde->d_name, "0123456789ABCDEF") == 16)
1914 : {
1915 : FullTransactionId fxid;
1916 : char *buf;
1917 :
1918 30 : fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16));
1919 :
1920 30 : buf = ProcessTwoPhaseBuffer(fxid, InvalidXLogRecPtr,
1921 : true, false, false);
1922 30 : if (buf == NULL)
1923 0 : continue;
1924 :
1925 30 : PrepareRedoAdd(fxid, buf, InvalidXLogRecPtr,
1926 : InvalidXLogRecPtr, InvalidRepOriginId);
1927 : }
1928 : }
1929 1862 : LWLockRelease(TwoPhaseStateLock);
1930 1862 : FreeDir(cldir);
1931 1862 : }
1932 :
1933 : /*
1934 : * PrescanPreparedTransactions
1935 : *
1936 : * Scan the shared memory entries of TwoPhaseState and determine the range
1937 : * of valid XIDs present. This is run during database startup, after we
1938 : * have completed reading WAL. TransamVariables->nextXid has been set to
1939 : * one more than the highest XID for which evidence exists in WAL.
1940 : *
1941 : * We throw away any prepared xacts with main XID beyond nextXid --- if any
1942 : * are present, it suggests that the DBA has done a PITR recovery to an
1943 : * earlier point in time without cleaning out pg_twophase. We dare not
1944 : * try to recover such prepared xacts since they likely depend on database
1945 : * state that doesn't exist now.
1946 : *
1947 : * However, we will advance nextXid beyond any subxact XIDs belonging to
1948 : * valid prepared xacts. We need to do this since subxact commit doesn't
1949 : * write a WAL entry, and so there might be no evidence in WAL of those
1950 : * subxact XIDs.
1951 : *
1952 : * On corrupted two-phase files, fail immediately. Keeping around broken
1953 : * entries and let replay continue causes harm on the system, and a new
1954 : * backup should be rolled in.
1955 : *
1956 : * Our other responsibility is to determine and return the oldest valid XID
1957 : * among the prepared xacts (if none, return TransamVariables->nextXid).
1958 : * This is needed to synchronize pg_subtrans startup properly.
1959 : *
1960 : * If xids_p and nxids_p are not NULL, pointer to a palloc'd array of all
1961 : * top-level xids is stored in *xids_p. The number of entries in the array
1962 : * is returned in *nxids_p.
1963 : */
1964 : TransactionId
1965 1862 : PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
1966 : {
1967 1862 : FullTransactionId nextXid = TransamVariables->nextXid;
1968 1862 : TransactionId origNextXid = XidFromFullTransactionId(nextXid);
1969 1862 : TransactionId result = origNextXid;
1970 1862 : TransactionId *xids = NULL;
1971 1862 : int nxids = 0;
1972 1862 : int allocsize = 0;
1973 : int i;
1974 :
1975 1862 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
1976 1966 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
1977 : {
1978 : TransactionId xid;
1979 : char *buf;
1980 104 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
1981 :
1982 : Assert(gxact->inredo);
1983 :
1984 104 : buf = ProcessTwoPhaseBuffer(gxact->fxid,
1985 : gxact->prepare_start_lsn,
1986 104 : gxact->ondisk, false, true);
1987 :
1988 104 : if (buf == NULL)
1989 0 : continue;
1990 :
1991 : /*
1992 : * OK, we think this file is valid. Incorporate xid into the
1993 : * running-minimum result.
1994 : */
1995 104 : xid = XidFromFullTransactionId(gxact->fxid);
1996 104 : if (TransactionIdPrecedes(xid, result))
1997 90 : result = xid;
1998 :
1999 104 : if (xids_p)
2000 : {
2001 38 : if (nxids == allocsize)
2002 : {
2003 30 : if (nxids == 0)
2004 : {
2005 30 : allocsize = 10;
2006 30 : xids = palloc(allocsize * sizeof(TransactionId));
2007 : }
2008 : else
2009 : {
2010 0 : allocsize = allocsize * 2;
2011 0 : xids = repalloc(xids, allocsize * sizeof(TransactionId));
2012 : }
2013 : }
2014 38 : xids[nxids++] = xid;
2015 : }
2016 :
2017 104 : pfree(buf);
2018 : }
2019 1862 : LWLockRelease(TwoPhaseStateLock);
2020 :
2021 1862 : if (xids_p)
2022 : {
2023 116 : *xids_p = xids;
2024 116 : *nxids_p = nxids;
2025 : }
2026 :
2027 1862 : return result;
2028 : }
2029 :
2030 : /*
2031 : * StandbyRecoverPreparedTransactions
2032 : *
2033 : * Scan the shared memory entries of TwoPhaseState and setup all the required
2034 : * information to allow standby queries to treat prepared transactions as still
2035 : * active.
2036 : *
2037 : * This is never called at the end of recovery - we use
2038 : * RecoverPreparedTransactions() at that point.
2039 : *
2040 : * This updates pg_subtrans, so that any subtransactions will be correctly
2041 : * seen as in-progress in snapshots taken during recovery.
2042 : */
2043 : void
2044 116 : StandbyRecoverPreparedTransactions(void)
2045 : {
2046 : int i;
2047 :
2048 116 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
2049 154 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
2050 : {
2051 : char *buf;
2052 38 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
2053 :
2054 : Assert(gxact->inredo);
2055 :
2056 38 : buf = ProcessTwoPhaseBuffer(gxact->fxid,
2057 : gxact->prepare_start_lsn,
2058 38 : gxact->ondisk, true, false);
2059 38 : if (buf != NULL)
2060 38 : pfree(buf);
2061 : }
2062 116 : LWLockRelease(TwoPhaseStateLock);
2063 116 : }
2064 :
2065 : /*
2066 : * RecoverPreparedTransactions
2067 : *
2068 : * Scan the shared memory entries of TwoPhaseState and reload the state for
2069 : * each prepared transaction (reacquire locks, etc).
2070 : *
2071 : * This is run at the end of recovery, but before we allow backends to write
2072 : * WAL.
2073 : *
2074 : * At the end of recovery the way we take snapshots will change. We now need
2075 : * to mark all running transactions with their full SubTransSetParent() info
2076 : * to allow normal snapshots to work correctly if snapshots overflow.
2077 : * We do this here because by definition prepared transactions are the only
2078 : * type of write transaction still running, so this is necessary and
2079 : * complete.
2080 : */
2081 : void
2082 1746 : RecoverPreparedTransactions(void)
2083 : {
2084 : int i;
2085 :
2086 1746 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
2087 1812 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
2088 : {
2089 : char *buf;
2090 66 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
2091 66 : FullTransactionId fxid = gxact->fxid;
2092 : char *bufptr;
2093 : TwoPhaseFileHeader *hdr;
2094 : TransactionId *subxids;
2095 : const char *gid;
2096 :
2097 : /*
2098 : * Reconstruct subtrans state for the transaction --- needed because
2099 : * pg_subtrans is not preserved over a restart. Note that we are
2100 : * linking all the subtransactions directly to the top-level XID;
2101 : * there may originally have been a more complex hierarchy, but
2102 : * there's no need to restore that exactly. It's possible that
2103 : * SubTransSetParent has been set before, if the prepared transaction
2104 : * generated xid assignment records.
2105 : */
2106 66 : buf = ProcessTwoPhaseBuffer(gxact->fxid,
2107 : gxact->prepare_start_lsn,
2108 66 : gxact->ondisk, true, false);
2109 66 : if (buf == NULL)
2110 0 : continue;
2111 :
2112 66 : ereport(LOG,
2113 : (errmsg("recovering prepared transaction %u of epoch %u from shared memory",
2114 : XidFromFullTransactionId(gxact->fxid),
2115 : EpochFromFullTransactionId(gxact->fxid))));
2116 :
2117 66 : hdr = (TwoPhaseFileHeader *) buf;
2118 : Assert(TransactionIdEquals(hdr->xid,
2119 : XidFromFullTransactionId(gxact->fxid)));
2120 66 : bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
2121 66 : gid = (const char *) bufptr;
2122 66 : bufptr += MAXALIGN(hdr->gidlen);
2123 66 : subxids = (TransactionId *) bufptr;
2124 66 : bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
2125 66 : bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileLocator));
2126 66 : bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileLocator));
2127 66 : bufptr += MAXALIGN(hdr->ncommitstats * sizeof(xl_xact_stats_item));
2128 66 : bufptr += MAXALIGN(hdr->nabortstats * sizeof(xl_xact_stats_item));
2129 66 : bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
2130 :
2131 : /*
2132 : * Recreate its GXACT and dummy PGPROC. But, check whether it was
2133 : * added in redo and already has a shmem entry for it.
2134 : */
2135 66 : MarkAsPreparingGuts(gxact, gxact->fxid, gid,
2136 : hdr->prepared_at,
2137 : hdr->owner, hdr->database);
2138 :
2139 : /* recovered, so reset the flag for entries generated by redo */
2140 66 : gxact->inredo = false;
2141 :
2142 66 : GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
2143 66 : MarkAsPrepared(gxact, true);
2144 :
2145 66 : LWLockRelease(TwoPhaseStateLock);
2146 :
2147 : /*
2148 : * Recover other state (notably locks) using resource managers.
2149 : */
2150 66 : ProcessRecords(bufptr, fxid, twophase_recover_callbacks);
2151 :
2152 : /*
2153 : * Release locks held by the standby process after we process each
2154 : * prepared transaction. As a result, we don't need too many
2155 : * additional locks at any one time.
2156 : */
2157 66 : if (InHotStandby)
2158 14 : StandbyReleaseLockTree(hdr->xid, hdr->nsubxacts, subxids);
2159 :
2160 : /*
2161 : * We're done with recovering this transaction. Clear MyLockedGxact,
2162 : * like we do in PrepareTransaction() during normal operation.
2163 : */
2164 66 : PostPrepare_Twophase();
2165 :
2166 66 : pfree(buf);
2167 :
2168 66 : LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
2169 : }
2170 :
2171 1746 : LWLockRelease(TwoPhaseStateLock);
2172 1746 : }
2173 :
2174 : /*
2175 : * ProcessTwoPhaseBuffer
2176 : *
2177 : * Given a FullTransactionId, read it either from disk or read it directly
2178 : * via shmem xlog record pointer using the provided "prepare_start_lsn".
2179 : *
2180 : * If setParent is true, set up subtransaction parent linkages.
2181 : *
2182 : * If setNextXid is true, set TransamVariables->nextXid to the newest
2183 : * value scanned.
2184 : */
2185 : static char *
2186 238 : ProcessTwoPhaseBuffer(FullTransactionId fxid,
2187 : XLogRecPtr prepare_start_lsn,
2188 : bool fromdisk,
2189 : bool setParent, bool setNextXid)
2190 : {
2191 238 : FullTransactionId nextXid = TransamVariables->nextXid;
2192 : TransactionId *subxids;
2193 : char *buf;
2194 : TwoPhaseFileHeader *hdr;
2195 : int i;
2196 :
2197 : Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
2198 :
2199 238 : if (!fromdisk)
2200 : Assert(prepare_start_lsn != InvalidXLogRecPtr);
2201 :
2202 : /* Already processed? */
2203 476 : if (TransactionIdDidCommit(XidFromFullTransactionId(fxid)) ||
2204 238 : TransactionIdDidAbort(XidFromFullTransactionId(fxid)))
2205 : {
2206 0 : if (fromdisk)
2207 : {
2208 0 : ereport(WARNING,
2209 : (errmsg("removing stale two-phase state file for transaction %u of epoch %u",
2210 : XidFromFullTransactionId(fxid),
2211 : EpochFromFullTransactionId(fxid))));
2212 0 : RemoveTwoPhaseFile(fxid, true);
2213 : }
2214 : else
2215 : {
2216 0 : ereport(WARNING,
2217 : (errmsg("removing stale two-phase state from memory for transaction %u of epoch %u",
2218 : XidFromFullTransactionId(fxid),
2219 : EpochFromFullTransactionId(fxid))));
2220 0 : PrepareRedoRemoveFull(fxid, true);
2221 : }
2222 0 : return NULL;
2223 : }
2224 :
2225 : /* Reject XID if too new */
2226 238 : if (FullTransactionIdFollowsOrEquals(fxid, nextXid))
2227 : {
2228 0 : if (fromdisk)
2229 : {
2230 0 : ereport(WARNING,
2231 : (errmsg("removing future two-phase state file for transaction %u of epoch %u",
2232 : XidFromFullTransactionId(fxid),
2233 : EpochFromFullTransactionId(fxid))));
2234 0 : RemoveTwoPhaseFile(fxid, true);
2235 : }
2236 : else
2237 : {
2238 0 : ereport(WARNING,
2239 : (errmsg("removing future two-phase state from memory for transaction %u of epoch %u",
2240 : XidFromFullTransactionId(fxid),
2241 : EpochFromFullTransactionId(fxid))));
2242 0 : PrepareRedoRemoveFull(fxid, true);
2243 : }
2244 0 : return NULL;
2245 : }
2246 :
2247 238 : if (fromdisk)
2248 : {
2249 : /* Read and validate file */
2250 94 : buf = ReadTwoPhaseFile(fxid, false);
2251 : }
2252 : else
2253 : {
2254 : /* Read xlog data */
2255 144 : XlogReadTwoPhaseData(prepare_start_lsn, &buf, NULL);
2256 : }
2257 :
2258 : /* Deconstruct header */
2259 238 : hdr = (TwoPhaseFileHeader *) buf;
2260 238 : if (!TransactionIdEquals(hdr->xid, XidFromFullTransactionId(fxid)))
2261 : {
2262 0 : if (fromdisk)
2263 0 : ereport(ERROR,
2264 : (errcode(ERRCODE_DATA_CORRUPTED),
2265 : errmsg("corrupted two-phase state file for transaction %u of epoch %u",
2266 : XidFromFullTransactionId(fxid),
2267 : EpochFromFullTransactionId(fxid))));
2268 : else
2269 0 : ereport(ERROR,
2270 : (errcode(ERRCODE_DATA_CORRUPTED),
2271 : errmsg("corrupted two-phase state in memory for transaction %u of epoch %u",
2272 : XidFromFullTransactionId(fxid),
2273 : EpochFromFullTransactionId(fxid))));
2274 : }
2275 :
2276 : /*
2277 : * Examine subtransaction XIDs ... they should all follow main XID, and
2278 : * they may force us to advance nextXid.
2279 : */
2280 238 : subxids = (TransactionId *) (buf +
2281 238 : MAXALIGN(sizeof(TwoPhaseFileHeader)) +
2282 238 : MAXALIGN(hdr->gidlen));
2283 3802 : for (i = 0; i < hdr->nsubxacts; i++)
2284 : {
2285 3564 : TransactionId subxid = subxids[i];
2286 :
2287 : Assert(TransactionIdFollows(subxid, XidFromFullTransactionId(fxid)));
2288 :
2289 : /* update nextXid if needed */
2290 3564 : if (setNextXid)
2291 1642 : AdvanceNextFullTransactionIdPastXid(subxid);
2292 :
2293 3564 : if (setParent)
2294 1642 : SubTransSetParent(subxid, XidFromFullTransactionId(fxid));
2295 : }
2296 :
2297 238 : return buf;
2298 : }
2299 :
2300 :
2301 : /*
2302 : * RecordTransactionCommitPrepared
2303 : *
2304 : * This is basically the same as RecordTransactionCommit (q.v. if you change
2305 : * this function): in particular, we must set DELAY_CHKPT_IN_COMMIT to avoid a
2306 : * race condition.
2307 : *
2308 : * We know the transaction made at least one XLOG entry (its PREPARE),
2309 : * so it is never possible to optimize out the commit record.
2310 : */
2311 : static void
2312 502 : RecordTransactionCommitPrepared(TransactionId xid,
2313 : int nchildren,
2314 : TransactionId *children,
2315 : int nrels,
2316 : RelFileLocator *rels,
2317 : int nstats,
2318 : xl_xact_stats_item *stats,
2319 : int ninvalmsgs,
2320 : SharedInvalidationMessage *invalmsgs,
2321 : bool initfileinval,
2322 : const char *gid)
2323 : {
2324 : XLogRecPtr recptr;
2325 : TimestampTz committs;
2326 : bool replorigin;
2327 :
2328 : /*
2329 : * Are we using the replication origins feature? Or, in other words, are
2330 : * we replaying remote actions?
2331 : */
2332 544 : replorigin = (replorigin_session_origin != InvalidRepOriginId &&
2333 42 : replorigin_session_origin != DoNotReplicateId);
2334 :
2335 502 : START_CRIT_SECTION();
2336 :
2337 : /* See notes in RecordTransactionCommit */
2338 : Assert((MyProc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) == 0);
2339 502 : MyProc->delayChkptFlags |= DELAY_CHKPT_IN_COMMIT;
2340 :
2341 : /*
2342 : * Ensures the DELAY_CHKPT_IN_COMMIT flag write is globally visible before
2343 : * commit time is written.
2344 : */
2345 502 : pg_write_barrier();
2346 :
2347 : /*
2348 : * Note it is important to set committs value after marking ourselves as
2349 : * in the commit critical section (DELAY_CHKPT_IN_COMMIT). This is because
2350 : * we want to ensure all transactions that have acquired commit timestamp
2351 : * are finished before we allow the logical replication client to advance
2352 : * its xid which is used to hold back dead rows for conflict detection.
2353 : * See comments atop worker.c.
2354 : */
2355 502 : committs = GetCurrentTimestamp();
2356 :
2357 : /*
2358 : * Emit the XLOG commit record. Note that we mark 2PC commits as
2359 : * potentially having AccessExclusiveLocks since we don't know whether or
2360 : * not they do.
2361 : */
2362 502 : recptr = XactLogCommitRecord(committs,
2363 : nchildren, children, nrels, rels,
2364 : nstats, stats,
2365 : ninvalmsgs, invalmsgs,
2366 : initfileinval,
2367 502 : MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
2368 : xid, gid);
2369 :
2370 :
2371 502 : if (replorigin)
2372 : /* Move LSNs forward for this replication origin */
2373 42 : replorigin_session_advance(replorigin_session_origin_lsn,
2374 : XactLastRecEnd);
2375 :
2376 : /*
2377 : * Record commit timestamp. The value comes from plain commit timestamp
2378 : * if replorigin is not enabled, or replorigin already set a value for us
2379 : * in replorigin_session_origin_timestamp otherwise.
2380 : *
2381 : * We don't need to WAL-log anything here, as the commit record written
2382 : * above already contains the data.
2383 : */
2384 502 : if (!replorigin || replorigin_session_origin_timestamp == 0)
2385 460 : replorigin_session_origin_timestamp = committs;
2386 :
2387 502 : TransactionTreeSetCommitTsData(xid, nchildren, children,
2388 : replorigin_session_origin_timestamp,
2389 : replorigin_session_origin);
2390 :
2391 : /*
2392 : * We don't currently try to sleep before flush here ... nor is there any
2393 : * support for async commit of a prepared xact (the very idea is probably
2394 : * a contradiction)
2395 : */
2396 :
2397 : /* Flush XLOG to disk */
2398 502 : XLogFlush(recptr);
2399 :
2400 : /* Mark the transaction committed in pg_xact */
2401 502 : TransactionIdCommitTree(xid, nchildren, children);
2402 :
2403 : /* Checkpoint can proceed now */
2404 502 : MyProc->delayChkptFlags &= ~DELAY_CHKPT_IN_COMMIT;
2405 :
2406 502 : END_CRIT_SECTION();
2407 :
2408 : /*
2409 : * Wait for synchronous replication, if required.
2410 : *
2411 : * Note that at this stage we have marked clog, but still show as running
2412 : * in the procarray and continue to hold locks.
2413 : */
2414 502 : SyncRepWaitForLSN(recptr, true);
2415 502 : }
2416 :
2417 : /*
2418 : * RecordTransactionAbortPrepared
2419 : *
2420 : * This is basically the same as RecordTransactionAbort.
2421 : *
2422 : * We know the transaction made at least one XLOG entry (its PREPARE),
2423 : * so it is never possible to optimize out the abort record.
2424 : */
2425 : static void
2426 76 : RecordTransactionAbortPrepared(TransactionId xid,
2427 : int nchildren,
2428 : TransactionId *children,
2429 : int nrels,
2430 : RelFileLocator *rels,
2431 : int nstats,
2432 : xl_xact_stats_item *stats,
2433 : const char *gid)
2434 : {
2435 : XLogRecPtr recptr;
2436 : bool replorigin;
2437 :
2438 : /*
2439 : * Are we using the replication origins feature? Or, in other words, are
2440 : * we replaying remote actions?
2441 : */
2442 88 : replorigin = (replorigin_session_origin != InvalidRepOriginId &&
2443 12 : replorigin_session_origin != DoNotReplicateId);
2444 :
2445 : /*
2446 : * Catch the scenario where we aborted partway through
2447 : * RecordTransactionCommitPrepared ...
2448 : */
2449 76 : if (TransactionIdDidCommit(xid))
2450 0 : elog(PANIC, "cannot abort transaction %u, it was already committed",
2451 : xid);
2452 :
2453 76 : START_CRIT_SECTION();
2454 :
2455 : /*
2456 : * Emit the XLOG commit record. Note that we mark 2PC aborts as
2457 : * potentially having AccessExclusiveLocks since we don't know whether or
2458 : * not they do.
2459 : */
2460 76 : recptr = XactLogAbortRecord(GetCurrentTimestamp(),
2461 : nchildren, children,
2462 : nrels, rels,
2463 : nstats, stats,
2464 76 : MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
2465 : xid, gid);
2466 :
2467 76 : if (replorigin)
2468 : /* Move LSNs forward for this replication origin */
2469 12 : replorigin_session_advance(replorigin_session_origin_lsn,
2470 : XactLastRecEnd);
2471 :
2472 : /* Always flush, since we're about to remove the 2PC state file */
2473 76 : XLogFlush(recptr);
2474 :
2475 : /*
2476 : * Mark the transaction aborted in clog. This is not absolutely necessary
2477 : * but we may as well do it while we are here.
2478 : */
2479 76 : TransactionIdAbortTree(xid, nchildren, children);
2480 :
2481 76 : END_CRIT_SECTION();
2482 :
2483 : /*
2484 : * Wait for synchronous replication, if required.
2485 : *
2486 : * Note that at this stage we have marked clog, but still show as running
2487 : * in the procarray and continue to hold locks.
2488 : */
2489 76 : SyncRepWaitForLSN(recptr, false);
2490 76 : }
2491 :
2492 : /*
2493 : * PrepareRedoAdd
2494 : *
2495 : * Store pointers to the start/end of the WAL record along with the xid in
2496 : * a gxact entry in shared memory TwoPhaseState structure. If caller
2497 : * specifies InvalidXLogRecPtr as WAL location to fetch the two-phase
2498 : * data, the entry is marked as located on disk.
2499 : */
2500 : void
2501 182 : PrepareRedoAdd(FullTransactionId fxid, char *buf,
2502 : XLogRecPtr start_lsn, XLogRecPtr end_lsn,
2503 : RepOriginId origin_id)
2504 : {
2505 182 : TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf;
2506 : char *bufptr;
2507 : const char *gid;
2508 : GlobalTransaction gxact;
2509 :
2510 : Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
2511 : Assert(RecoveryInProgress());
2512 :
2513 182 : if (!FullTransactionIdIsValid(fxid))
2514 : {
2515 : Assert(InRecovery);
2516 152 : fxid = FullTransactionIdFromAllowableAt(TransamVariables->nextXid,
2517 : hdr->xid);
2518 : }
2519 :
2520 182 : bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
2521 182 : gid = (const char *) bufptr;
2522 :
2523 : /*
2524 : * Reserve the GID for the given transaction in the redo code path.
2525 : *
2526 : * This creates a gxact struct and puts it into the active array.
2527 : *
2528 : * In redo, this struct is mainly used to track PREPARE/COMMIT entries in
2529 : * shared memory. Hence, we only fill up the bare minimum contents here.
2530 : * The gxact also gets marked with gxact->inredo set to true to indicate
2531 : * that it got added in the redo phase
2532 : */
2533 :
2534 : /*
2535 : * In the event of a crash while a checkpoint was running, it may be
2536 : * possible that some two-phase data found its way to disk while its
2537 : * corresponding record needs to be replayed in the follow-up recovery. As
2538 : * the 2PC data was on disk, it has already been restored at the beginning
2539 : * of recovery with restoreTwoPhaseData(), so skip this record to avoid
2540 : * duplicates in TwoPhaseState. If a consistent state has been reached,
2541 : * the record is added to TwoPhaseState and it should have no
2542 : * corresponding file in pg_twophase.
2543 : */
2544 182 : if (!XLogRecPtrIsInvalid(start_lsn))
2545 : {
2546 : char path[MAXPGPATH];
2547 :
2548 : Assert(InRecovery);
2549 152 : TwoPhaseFilePath(path, fxid);
2550 :
2551 152 : if (access(path, F_OK) == 0)
2552 : {
2553 0 : ereport(reachedConsistency ? ERROR : WARNING,
2554 : (errmsg("could not recover two-phase state file for transaction %u",
2555 : hdr->xid),
2556 : errdetail("Two-phase state file has been found in WAL record %X/%08X, but this transaction has already been restored from disk.",
2557 : LSN_FORMAT_ARGS(start_lsn))));
2558 0 : return;
2559 : }
2560 :
2561 152 : if (errno != ENOENT)
2562 0 : ereport(ERROR,
2563 : (errcode_for_file_access(),
2564 : errmsg("could not access file \"%s\": %m", path)));
2565 : }
2566 :
2567 : /* Get a free gxact from the freelist */
2568 182 : if (TwoPhaseState->freeGXacts == NULL)
2569 0 : ereport(ERROR,
2570 : (errcode(ERRCODE_OUT_OF_MEMORY),
2571 : errmsg("maximum number of prepared transactions reached"),
2572 : errhint("Increase \"max_prepared_transactions\" (currently %d).",
2573 : max_prepared_xacts)));
2574 182 : gxact = TwoPhaseState->freeGXacts;
2575 182 : TwoPhaseState->freeGXacts = gxact->next;
2576 :
2577 182 : gxact->prepared_at = hdr->prepared_at;
2578 182 : gxact->prepare_start_lsn = start_lsn;
2579 182 : gxact->prepare_end_lsn = end_lsn;
2580 182 : gxact->fxid = fxid;
2581 182 : gxact->owner = hdr->owner;
2582 182 : gxact->locking_backend = INVALID_PROC_NUMBER;
2583 182 : gxact->valid = false;
2584 182 : gxact->ondisk = XLogRecPtrIsInvalid(start_lsn);
2585 182 : gxact->inredo = true; /* yes, added in redo */
2586 182 : strcpy(gxact->gid, gid);
2587 :
2588 : /* And insert it into the active array */
2589 : Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
2590 182 : TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
2591 :
2592 182 : if (origin_id != InvalidRepOriginId)
2593 : {
2594 : /* recover apply progress */
2595 26 : replorigin_advance(origin_id, hdr->origin_lsn, end_lsn,
2596 : false /* backward */ , false /* WAL */ );
2597 : }
2598 :
2599 182 : elog(DEBUG2, "added 2PC data in shared memory for transaction %u of epoch %u",
2600 : XidFromFullTransactionId(gxact->fxid),
2601 : EpochFromFullTransactionId(gxact->fxid));
2602 : }
2603 :
2604 : /*
2605 : * PrepareRedoRemoveFull
2606 : *
2607 : * Remove the corresponding gxact entry from TwoPhaseState. Also remove
2608 : * the 2PC file if a prepared transaction was saved via an earlier checkpoint.
2609 : *
2610 : * Caller must hold TwoPhaseStateLock in exclusive mode, because TwoPhaseState
2611 : * is updated.
2612 : */
2613 : static void
2614 130 : PrepareRedoRemoveFull(FullTransactionId fxid, bool giveWarning)
2615 : {
2616 130 : GlobalTransaction gxact = NULL;
2617 : int i;
2618 130 : bool found = false;
2619 :
2620 : Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
2621 : Assert(RecoveryInProgress());
2622 :
2623 130 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
2624 : {
2625 112 : gxact = TwoPhaseState->prepXacts[i];
2626 :
2627 112 : if (FullTransactionIdEquals(gxact->fxid, fxid))
2628 : {
2629 : Assert(gxact->inredo);
2630 112 : found = true;
2631 112 : break;
2632 : }
2633 : }
2634 :
2635 : /*
2636 : * Just leave if there is nothing, this is expected during WAL replay.
2637 : */
2638 130 : if (!found)
2639 18 : return;
2640 :
2641 : /*
2642 : * And now we can clean up any files we may have left.
2643 : */
2644 112 : elog(DEBUG2, "removing 2PC data for transaction %u of epoch %u ",
2645 : XidFromFullTransactionId(fxid),
2646 : EpochFromFullTransactionId(fxid));
2647 :
2648 112 : if (gxact->ondisk)
2649 8 : RemoveTwoPhaseFile(fxid, giveWarning);
2650 :
2651 112 : RemoveGXact(gxact);
2652 : }
2653 :
2654 : /*
2655 : * Wrapper of PrepareRedoRemoveFull(), for TransactionIds.
2656 : */
2657 : void
2658 130 : PrepareRedoRemove(TransactionId xid, bool giveWarning)
2659 : {
2660 : FullTransactionId fxid =
2661 130 : FullTransactionIdFromAllowableAt(TransamVariables->nextXid, xid);
2662 :
2663 130 : PrepareRedoRemoveFull(fxid, giveWarning);
2664 130 : }
2665 :
2666 : /*
2667 : * LookupGXact
2668 : * Check if the prepared transaction with the given GID, lsn and timestamp
2669 : * exists.
2670 : *
2671 : * Note that we always compare with the LSN where prepare ends because that is
2672 : * what is stored as origin_lsn in the 2PC file.
2673 : *
2674 : * This function is primarily used to check if the prepared transaction
2675 : * received from the upstream (remote node) already exists. Checking only GID
2676 : * is not sufficient because a different prepared xact with the same GID can
2677 : * exist on the same node. So, we are ensuring to match origin_lsn and
2678 : * origin_timestamp of prepared xact to avoid the possibility of a match of
2679 : * prepared xact from two different nodes.
2680 : */
2681 : bool
2682 10 : LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
2683 : TimestampTz origin_prepare_timestamp)
2684 : {
2685 : int i;
2686 10 : bool found = false;
2687 :
2688 10 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
2689 10 : for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
2690 : {
2691 10 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
2692 :
2693 : /* Ignore not-yet-valid GIDs. */
2694 10 : if (gxact->valid && strcmp(gxact->gid, gid) == 0)
2695 : {
2696 : char *buf;
2697 : TwoPhaseFileHeader *hdr;
2698 :
2699 : /*
2700 : * We are not expecting collisions of GXACTs (same gid) between
2701 : * publisher and subscribers, so we perform all I/O while holding
2702 : * TwoPhaseStateLock for simplicity.
2703 : *
2704 : * To move the I/O out of the lock, we need to ensure that no
2705 : * other backend commits the prepared xact in the meantime. We can
2706 : * do this optimization if we encounter many collisions in GID
2707 : * between publisher and subscriber.
2708 : */
2709 10 : if (gxact->ondisk)
2710 0 : buf = ReadTwoPhaseFile(gxact->fxid, false);
2711 : else
2712 : {
2713 : Assert(gxact->prepare_start_lsn);
2714 10 : XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
2715 : }
2716 :
2717 10 : hdr = (TwoPhaseFileHeader *) buf;
2718 :
2719 10 : if (hdr->origin_lsn == prepare_end_lsn &&
2720 10 : hdr->origin_timestamp == origin_prepare_timestamp)
2721 : {
2722 10 : found = true;
2723 10 : pfree(buf);
2724 10 : break;
2725 : }
2726 :
2727 0 : pfree(buf);
2728 : }
2729 : }
2730 10 : LWLockRelease(TwoPhaseStateLock);
2731 10 : return found;
2732 : }
2733 :
2734 : /*
2735 : * TwoPhaseTransactionGid
2736 : * Form the prepared transaction GID for two_phase transactions.
2737 : *
2738 : * Return the GID in the supplied buffer.
2739 : */
2740 : void
2741 96 : TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid_res, int szgid)
2742 : {
2743 : Assert(OidIsValid(subid));
2744 :
2745 96 : if (!TransactionIdIsValid(xid))
2746 0 : ereport(ERROR,
2747 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
2748 : errmsg_internal("invalid two-phase transaction ID")));
2749 :
2750 96 : snprintf(gid_res, szgid, "pg_gid_%u_%u", subid, xid);
2751 96 : }
2752 :
2753 : /*
2754 : * IsTwoPhaseTransactionGidForSubid
2755 : * Check whether the given GID (as formed by TwoPhaseTransactionGid) is
2756 : * for the specified 'subid'.
2757 : */
2758 : static bool
2759 0 : IsTwoPhaseTransactionGidForSubid(Oid subid, char *gid)
2760 : {
2761 : int ret;
2762 : Oid subid_from_gid;
2763 : TransactionId xid_from_gid;
2764 : char gid_tmp[GIDSIZE];
2765 :
2766 : /* Extract the subid and xid from the given GID */
2767 0 : ret = sscanf(gid, "pg_gid_%u_%u", &subid_from_gid, &xid_from_gid);
2768 :
2769 : /*
2770 : * Check that the given GID has expected format, and at least the subid
2771 : * matches.
2772 : */
2773 0 : if (ret != 2 || subid != subid_from_gid)
2774 0 : return false;
2775 :
2776 : /*
2777 : * Reconstruct a temporary GID based on the subid and xid extracted from
2778 : * the given GID and check whether the temporary GID and the given GID
2779 : * match.
2780 : */
2781 0 : TwoPhaseTransactionGid(subid, xid_from_gid, gid_tmp, sizeof(gid_tmp));
2782 :
2783 0 : return strcmp(gid, gid_tmp) == 0;
2784 : }
2785 :
2786 : /*
2787 : * LookupGXactBySubid
2788 : * Check if the prepared transaction done by apply worker exists.
2789 : */
2790 : bool
2791 2 : LookupGXactBySubid(Oid subid)
2792 : {
2793 2 : bool found = false;
2794 :
2795 2 : LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
2796 2 : for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
2797 : {
2798 0 : GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
2799 :
2800 : /* Ignore not-yet-valid GIDs. */
2801 0 : if (gxact->valid &&
2802 0 : IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
2803 : {
2804 0 : found = true;
2805 0 : break;
2806 : }
2807 : }
2808 2 : LWLockRelease(TwoPhaseStateLock);
2809 :
2810 2 : return found;
2811 : }
|
