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