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