1 /*-------------------------------------------------------------------------
4 * Two-phase commit support functions.
6 * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
7 * Portions Copyright (c) 1994, Regents of the University of California
10 * $PostgreSQL: pgsql/src/backend/access/transam/twophase.c,v 1.51 2009/01/01 17:23:36 momjian Exp $
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.
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.
24 * A global transaction (gxact) also has a dummy PGPROC that is entered
25 * into the ProcArray array; this is what keeps the XID considered
26 * running by TransactionIdIsInProgress. It is also convenient as a
27 * PGPROC to hook the gxact's locks to.
29 * In order to survive crashes and shutdowns, all prepared
30 * transactions must be stored in permanent storage. This includes
31 * locking information, pending notifications etc. All that state
32 * information is written to the per-transaction state file in
33 * the pg_twophase directory.
35 *-------------------------------------------------------------------------
41 #include <sys/types.h>
45 #include "access/htup.h"
46 #include "access/subtrans.h"
47 #include "access/transam.h"
48 #include "access/twophase.h"
49 #include "access/twophase_rmgr.h"
50 #include "access/xact.h"
51 #include "access/xlogutils.h"
52 #include "catalog/pg_type.h"
53 #include "catalog/storage.h"
55 #include "miscadmin.h"
58 #include "storage/fd.h"
59 #include "storage/procarray.h"
60 #include "storage/smgr.h"
61 #include "utils/builtins.h"
62 #include "utils/memutils.h"
66 * Directory where Two-phase commit files reside within PGDATA
68 #define TWOPHASE_DIR "pg_twophase"
70 /* GUC variable, can't be changed after startup */
71 int max_prepared_xacts = 5;
74 * This struct describes one global transaction that is in prepared state
75 * or attempting to become prepared.
77 * The first component of the struct is a dummy PGPROC that is inserted
78 * into the global ProcArray so that the transaction appears to still be
79 * running and holding locks. It must be first because we cast pointers
80 * to PGPROC and pointers to GlobalTransactionData back and forth.
82 * The lifecycle of a global transaction is:
84 * 1. After checking that the requested GID is not in use, set up an
85 * entry in the TwoPhaseState->prepXacts array with the correct XID and GID,
86 * with locking_xid = my own XID and valid = false.
88 * 2. After successfully completing prepare, set valid = true and enter the
89 * contained PGPROC into the global ProcArray.
91 * 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry
92 * is valid and its locking_xid is no longer active, then store my current
93 * XID into locking_xid. This prevents concurrent attempts to commit or
94 * rollback the same prepared xact.
96 * 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
97 * from the ProcArray and the TwoPhaseState->prepXacts array and return it to
100 * Note that if the preparing transaction fails between steps 1 and 2, the
101 * entry will remain in prepXacts until recycled. We can detect recyclable
102 * entries by checking for valid = false and locking_xid no longer active.
104 * typedef struct GlobalTransactionData *GlobalTransaction appears in
109 typedef struct GlobalTransactionData
111 PGPROC proc; /* dummy proc */
112 TimestampTz prepared_at; /* time of preparation */
113 XLogRecPtr prepare_lsn; /* XLOG offset of prepare record */
114 Oid owner; /* ID of user that executed the xact */
115 TransactionId locking_xid; /* top-level XID of backend working on xact */
116 bool valid; /* TRUE if fully prepared */
117 char gid[GIDSIZE]; /* The GID assigned to the prepared xact */
118 } GlobalTransactionData;
121 * Two Phase Commit shared state. Access to this struct is protected
122 * by TwoPhaseStateLock.
124 typedef struct TwoPhaseStateData
126 /* Head of linked list of free GlobalTransactionData structs */
127 GlobalTransaction freeGXacts;
129 /* Number of valid prepXacts entries. */
133 * There are max_prepared_xacts items in this array, but C wants a
136 GlobalTransaction prepXacts[1]; /* VARIABLE LENGTH ARRAY */
137 } TwoPhaseStateData; /* VARIABLE LENGTH STRUCT */
139 static TwoPhaseStateData *TwoPhaseState;
142 static void RecordTransactionCommitPrepared(TransactionId xid,
144 TransactionId *children,
147 static void RecordTransactionAbortPrepared(TransactionId xid,
149 TransactionId *children,
152 static void ProcessRecords(char *bufptr, TransactionId xid,
153 const TwoPhaseCallback callbacks[]);
157 * Initialization of shared memory
160 TwoPhaseShmemSize(void)
164 /* Need the fixed struct, the array of pointers, and the GTD structs */
165 size = offsetof(TwoPhaseStateData, prepXacts);
166 size = add_size(size, mul_size(max_prepared_xacts,
167 sizeof(GlobalTransaction)));
168 size = MAXALIGN(size);
169 size = add_size(size, mul_size(max_prepared_xacts,
170 sizeof(GlobalTransactionData)));
176 TwoPhaseShmemInit(void)
180 TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
183 if (!IsUnderPostmaster)
185 GlobalTransaction gxacts;
189 TwoPhaseState->freeGXacts = NULL;
190 TwoPhaseState->numPrepXacts = 0;
193 * Initialize the linked list of free GlobalTransactionData structs
195 gxacts = (GlobalTransaction)
196 ((char *) TwoPhaseState +
197 MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
198 sizeof(GlobalTransaction) * max_prepared_xacts));
199 for (i = 0; i < max_prepared_xacts; i++)
201 gxacts[i].proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
202 TwoPhaseState->freeGXacts = &gxacts[i];
212 * Reserve the GID for the given transaction.
214 * Internally, this creates a gxact struct and puts it into the active array.
215 * NOTE: this is also used when reloading a gxact after a crash; so avoid
216 * assuming that we can use very much backend context.
219 MarkAsPreparing(TransactionId xid, const char *gid,
220 TimestampTz prepared_at, Oid owner, Oid databaseid)
222 GlobalTransaction gxact;
225 if (strlen(gid) >= GIDSIZE)
227 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
228 errmsg("transaction identifier \"%s\" is too long",
231 LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
234 * First, find and recycle any gxacts that failed during prepare. We do
235 * this partly to ensure we don't mistakenly say their GIDs are still
236 * reserved, and partly so we don't fail on out-of-slots unnecessarily.
238 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
240 gxact = TwoPhaseState->prepXacts[i];
241 if (!gxact->valid && !TransactionIdIsActive(gxact->locking_xid))
243 /* It's dead Jim ... remove from the active array */
244 TwoPhaseState->numPrepXacts--;
245 TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
246 /* and put it back in the freelist */
247 gxact->proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
248 TwoPhaseState->freeGXacts = gxact;
249 /* Back up index count too, so we don't miss scanning one */
254 /* Check for conflicting GID */
255 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
257 gxact = TwoPhaseState->prepXacts[i];
258 if (strcmp(gxact->gid, gid) == 0)
261 (errcode(ERRCODE_DUPLICATE_OBJECT),
262 errmsg("transaction identifier \"%s\" is already in use",
267 /* Get a free gxact from the freelist */
268 if (TwoPhaseState->freeGXacts == NULL)
270 (errcode(ERRCODE_OUT_OF_MEMORY),
271 errmsg("maximum number of prepared transactions reached"),
272 errhint("Increase max_prepared_transactions (currently %d).",
273 max_prepared_xacts)));
274 gxact = TwoPhaseState->freeGXacts;
275 TwoPhaseState->freeGXacts = (GlobalTransaction) gxact->proc.links.next;
278 MemSet(&gxact->proc, 0, sizeof(PGPROC));
279 SHMQueueElemInit(&(gxact->proc.links));
280 gxact->proc.waitStatus = STATUS_OK;
281 /* We set up the gxact's VXID as InvalidBackendId/XID */
282 gxact->proc.lxid = (LocalTransactionId) xid;
283 gxact->proc.xid = xid;
284 gxact->proc.xmin = InvalidTransactionId;
286 gxact->proc.backendId = InvalidBackendId;
287 gxact->proc.databaseId = databaseid;
288 gxact->proc.roleId = owner;
289 gxact->proc.inCommit = false;
290 gxact->proc.vacuumFlags = 0;
291 gxact->proc.lwWaiting = false;
292 gxact->proc.lwExclusive = false;
293 gxact->proc.lwWaitLink = NULL;
294 gxact->proc.waitLock = NULL;
295 gxact->proc.waitProcLock = NULL;
296 for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
297 SHMQueueInit(&(gxact->proc.myProcLocks[i]));
298 /* subxid data must be filled later by GXactLoadSubxactData */
299 gxact->proc.subxids.overflowed = false;
300 gxact->proc.subxids.nxids = 0;
302 gxact->prepared_at = prepared_at;
303 /* initialize LSN to 0 (start of WAL) */
304 gxact->prepare_lsn.xlogid = 0;
305 gxact->prepare_lsn.xrecoff = 0;
306 gxact->owner = owner;
307 gxact->locking_xid = xid;
308 gxact->valid = false;
309 strcpy(gxact->gid, gid);
311 /* And insert it into the active array */
312 Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
313 TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
315 LWLockRelease(TwoPhaseStateLock);
321 * GXactLoadSubxactData
323 * If the transaction being persisted had any subtransactions, this must
324 * be called before MarkAsPrepared() to load information into the dummy
328 GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
329 TransactionId *children)
331 /* We need no extra lock since the GXACT isn't valid yet */
332 if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
334 gxact->proc.subxids.overflowed = true;
335 nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
339 memcpy(gxact->proc.subxids.xids, children,
340 nsubxacts * sizeof(TransactionId));
341 gxact->proc.subxids.nxids = nsubxacts;
347 * Mark the GXACT as fully valid, and enter it into the global ProcArray.
350 MarkAsPrepared(GlobalTransaction gxact)
352 /* Lock here may be overkill, but I'm not convinced of that ... */
353 LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
354 Assert(!gxact->valid);
356 LWLockRelease(TwoPhaseStateLock);
359 * Put it into the global ProcArray so TransactionIdIsInProgress considers
360 * the XID as still running.
362 ProcArrayAdd(&gxact->proc);
367 * Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
369 static GlobalTransaction
370 LockGXact(const char *gid, Oid user)
374 LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
376 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
378 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
380 /* Ignore not-yet-valid GIDs */
383 if (strcmp(gxact->gid, gid) != 0)
386 /* Found it, but has someone else got it locked? */
387 if (TransactionIdIsValid(gxact->locking_xid))
389 if (TransactionIdIsActive(gxact->locking_xid))
391 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
392 errmsg("prepared transaction with identifier \"%s\" is busy",
394 gxact->locking_xid = InvalidTransactionId;
397 if (user != gxact->owner && !superuser_arg(user))
399 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
400 errmsg("permission denied to finish prepared transaction"),
401 errhint("Must be superuser or the user that prepared the transaction.")));
404 * Note: it probably would be possible to allow committing from
405 * another database; but at the moment NOTIFY is known not to work and
406 * there may be some other issues as well. Hence disallow until
407 * someone gets motivated to make it work.
409 if (MyDatabaseId != gxact->proc.databaseId)
411 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
412 errmsg("prepared transaction belongs to another database"),
413 errhint("Connect to the database where the transaction was prepared to finish it.")));
415 /* OK for me to lock it */
416 gxact->locking_xid = GetTopTransactionId();
418 LWLockRelease(TwoPhaseStateLock);
423 LWLockRelease(TwoPhaseStateLock);
426 (errcode(ERRCODE_UNDEFINED_OBJECT),
427 errmsg("prepared transaction with identifier \"%s\" does not exist",
436 * Remove the prepared transaction from the shared memory array.
438 * NB: caller should have already removed it from ProcArray
441 RemoveGXact(GlobalTransaction gxact)
445 LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
447 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
449 if (gxact == TwoPhaseState->prepXacts[i])
451 /* remove from the active array */
452 TwoPhaseState->numPrepXacts--;
453 TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
455 /* and put it back in the freelist */
456 gxact->proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
457 TwoPhaseState->freeGXacts = gxact;
459 LWLockRelease(TwoPhaseStateLock);
465 LWLockRelease(TwoPhaseStateLock);
467 elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
471 * TransactionIdIsPrepared
472 * True iff transaction associated with the identifier is prepared
473 * for two-phase commit
475 * Note: only gxacts marked "valid" are considered; but notice we do not
476 * check the locking status.
478 * This is not currently exported, because it is only needed internally.
481 TransactionIdIsPrepared(TransactionId xid)
486 LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
488 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
490 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
492 if (gxact->valid && gxact->proc.xid == xid)
499 LWLockRelease(TwoPhaseStateLock);
505 * Returns an array of all prepared transactions for the user-level
506 * function pg_prepared_xact.
508 * The returned array and all its elements are copies of internal data
509 * structures, to minimize the time we need to hold the TwoPhaseStateLock.
511 * WARNING -- we return even those transactions that are not fully prepared
512 * yet. The caller should filter them out if he doesn't want them.
514 * The returned array is palloc'd.
517 GetPreparedTransactionList(GlobalTransaction *gxacts)
519 GlobalTransaction array;
523 LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
525 if (TwoPhaseState->numPrepXacts == 0)
527 LWLockRelease(TwoPhaseStateLock);
533 num = TwoPhaseState->numPrepXacts;
534 array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
536 for (i = 0; i < num; i++)
537 memcpy(array + i, TwoPhaseState->prepXacts[i],
538 sizeof(GlobalTransactionData));
540 LWLockRelease(TwoPhaseStateLock);
546 /* Working status for pg_prepared_xact */
549 GlobalTransaction array;
556 * Produce a view with one row per prepared transaction.
558 * This function is here so we don't have to export the
559 * GlobalTransactionData struct definition.
562 pg_prepared_xact(PG_FUNCTION_ARGS)
564 FuncCallContext *funcctx;
565 Working_State *status;
567 if (SRF_IS_FIRSTCALL())
570 MemoryContext oldcontext;
572 /* create a function context for cross-call persistence */
573 funcctx = SRF_FIRSTCALL_INIT();
576 * Switch to memory context appropriate for multiple function calls
578 oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
580 /* build tupdesc for result tuples */
581 /* this had better match pg_prepared_xacts view in system_views.sql */
582 tupdesc = CreateTemplateTupleDesc(5, false);
583 TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
585 TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
587 TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
588 TIMESTAMPTZOID, -1, 0);
589 TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
591 TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
594 funcctx->tuple_desc = BlessTupleDesc(tupdesc);
597 * Collect all the 2PC status information that we will format and send
598 * out as a result set.
600 status = (Working_State *) palloc(sizeof(Working_State));
601 funcctx->user_fctx = (void *) status;
603 status->ngxacts = GetPreparedTransactionList(&status->array);
606 MemoryContextSwitchTo(oldcontext);
609 funcctx = SRF_PERCALL_SETUP();
610 status = (Working_State *) funcctx->user_fctx;
612 while (status->array != NULL && status->currIdx < status->ngxacts)
614 GlobalTransaction gxact = &status->array[status->currIdx++];
624 * Form tuple with appropriate data.
626 MemSet(values, 0, sizeof(values));
627 MemSet(nulls, 0, sizeof(nulls));
629 values[0] = TransactionIdGetDatum(gxact->proc.xid);
630 values[1] = CStringGetTextDatum(gxact->gid);
631 values[2] = TimestampTzGetDatum(gxact->prepared_at);
632 values[3] = ObjectIdGetDatum(gxact->owner);
633 values[4] = ObjectIdGetDatum(gxact->proc.databaseId);
635 tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
636 result = HeapTupleGetDatum(tuple);
637 SRF_RETURN_NEXT(funcctx, result);
640 SRF_RETURN_DONE(funcctx);
644 * TwoPhaseGetDummyProc
645 * Get the PGPROC that represents a prepared transaction specified by XID
648 TwoPhaseGetDummyProc(TransactionId xid)
650 PGPROC *result = NULL;
653 static TransactionId cached_xid = InvalidTransactionId;
654 static PGPROC *cached_proc = NULL;
657 * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
658 * repeatedly for the same XID. We can save work with a simple cache.
660 if (xid == cached_xid)
663 LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
665 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
667 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
669 if (gxact->proc.xid == xid)
671 result = &gxact->proc;
676 LWLockRelease(TwoPhaseStateLock);
678 if (result == NULL) /* should not happen */
679 elog(ERROR, "failed to find dummy PGPROC for xid %u", xid);
682 cached_proc = result;
687 /************************************************************************/
688 /* State file support */
689 /************************************************************************/
691 #define TwoPhaseFilePath(path, xid) \
692 snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)
695 * 2PC state file format:
697 * 1. TwoPhaseFileHeader
698 * 2. TransactionId[] (subtransactions)
699 * 3. RelFileNode[] (files to be deleted at commit)
700 * 4. RelFileNode[] (files to be deleted at abort)
701 * 5. TwoPhaseRecordOnDisk
703 * 7. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
706 * Each segment except the final CRC32 is MAXALIGN'd.
710 * Header for a 2PC state file
712 #define TWOPHASE_MAGIC 0x57F94531 /* format identifier */
714 typedef struct TwoPhaseFileHeader
716 uint32 magic; /* format identifier */
717 uint32 total_len; /* actual file length */
718 TransactionId xid; /* original transaction XID */
719 Oid database; /* OID of database it was in */
720 TimestampTz prepared_at; /* time of preparation */
721 Oid owner; /* user running the transaction */
722 int32 nsubxacts; /* number of following subxact XIDs */
723 int32 ncommitrels; /* number of delete-on-commit rels */
724 int32 nabortrels; /* number of delete-on-abort rels */
725 char gid[GIDSIZE]; /* GID for transaction */
726 } TwoPhaseFileHeader;
729 * Header for each record in a state file
731 * NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
732 * The rmgr data will be stored starting on a MAXALIGN boundary.
734 typedef struct TwoPhaseRecordOnDisk
736 uint32 len; /* length of rmgr data */
737 TwoPhaseRmgrId rmid; /* resource manager for this record */
738 uint16 info; /* flag bits for use by rmgr */
739 } TwoPhaseRecordOnDisk;
742 * During prepare, the state file is assembled in memory before writing it
743 * to WAL and the actual state file. We use a chain of XLogRecData blocks
744 * so that we will be able to pass the state file contents directly to
749 XLogRecData *head; /* first data block in the chain */
750 XLogRecData *tail; /* last block in chain */
751 uint32 bytes_free; /* free bytes left in tail block */
752 uint32 total_len; /* total data bytes in chain */
757 * Append a block of data to records data structure.
759 * NB: each block is padded to a MAXALIGN multiple. This must be
760 * accounted for when the file is later read!
762 * The data is copied, so the caller is free to modify it afterwards.
765 save_state_data(const void *data, uint32 len)
767 uint32 padlen = MAXALIGN(len);
769 if (padlen > records.bytes_free)
771 records.tail->next = palloc0(sizeof(XLogRecData));
772 records.tail = records.tail->next;
773 records.tail->buffer = InvalidBuffer;
774 records.tail->len = 0;
775 records.tail->next = NULL;
777 records.bytes_free = Max(padlen, 512);
778 records.tail->data = palloc(records.bytes_free);
781 memcpy(((char *) records.tail->data) + records.tail->len, data, len);
782 records.tail->len += padlen;
783 records.bytes_free -= padlen;
784 records.total_len += padlen;
788 * Start preparing a state file.
790 * Initializes data structure and inserts the 2PC file header record.
793 StartPrepare(GlobalTransaction gxact)
795 TransactionId xid = gxact->proc.xid;
796 TwoPhaseFileHeader hdr;
797 TransactionId *children;
798 RelFileNode *commitrels;
799 RelFileNode *abortrels;
801 /* Initialize linked list */
802 records.head = palloc0(sizeof(XLogRecData));
803 records.head->buffer = InvalidBuffer;
804 records.head->len = 0;
805 records.head->next = NULL;
807 records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
808 records.head->data = palloc(records.bytes_free);
810 records.tail = records.head;
812 records.total_len = 0;
815 hdr.magic = TWOPHASE_MAGIC;
816 hdr.total_len = 0; /* EndPrepare will fill this in */
818 hdr.database = gxact->proc.databaseId;
819 hdr.prepared_at = gxact->prepared_at;
820 hdr.owner = gxact->owner;
821 hdr.nsubxacts = xactGetCommittedChildren(&children);
822 hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels, NULL);
823 hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels, NULL);
824 StrNCpy(hdr.gid, gxact->gid, GIDSIZE);
826 save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
828 /* Add the additional info about subxacts and deletable files */
829 if (hdr.nsubxacts > 0)
831 save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
832 /* While we have the child-xact data, stuff it in the gxact too */
833 GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
835 if (hdr.ncommitrels > 0)
837 save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileNode));
840 if (hdr.nabortrels > 0)
842 save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileNode));
848 * Finish preparing state file.
850 * Calculates CRC and writes state file to WAL and in pg_twophase directory.
853 EndPrepare(GlobalTransaction gxact)
855 TransactionId xid = gxact->proc.xid;
856 TwoPhaseFileHeader *hdr;
857 char path[MAXPGPATH];
859 pg_crc32 statefile_crc;
863 /* Add the end sentinel to the list of 2PC records */
864 RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
867 /* Go back and fill in total_len in the file header record */
868 hdr = (TwoPhaseFileHeader *) records.head->data;
869 Assert(hdr->magic == TWOPHASE_MAGIC);
870 hdr->total_len = records.total_len + sizeof(pg_crc32);
873 * If the file size exceeds MaxAllocSize, we won't be able to read it in
874 * ReadTwoPhaseFile. Check for that now, rather than fail at commit time.
876 if (hdr->total_len > MaxAllocSize)
878 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
879 errmsg("two-phase state file maximum length exceeded")));
882 * Create the 2PC state file.
884 * Note: because we use BasicOpenFile(), we are responsible for ensuring
885 * the FD gets closed in any error exit path. Once we get into the
886 * critical section, though, it doesn't matter since any failure causes
889 TwoPhaseFilePath(path, xid);
891 fd = BasicOpenFile(path,
892 O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
896 (errcode_for_file_access(),
897 errmsg("could not create two-phase state file \"%s\": %m",
900 /* Write data to file, and calculate CRC as we pass over it */
901 INIT_CRC32(statefile_crc);
903 for (record = records.head; record != NULL; record = record->next)
905 COMP_CRC32(statefile_crc, record->data, record->len);
906 if ((write(fd, record->data, record->len)) != record->len)
910 (errcode_for_file_access(),
911 errmsg("could not write two-phase state file: %m")));
915 FIN_CRC32(statefile_crc);
918 * Write a deliberately bogus CRC to the state file; this is just paranoia
919 * to catch the case where four more bytes will run us out of disk space.
921 bogus_crc = ~statefile_crc;
923 if ((write(fd, &bogus_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
927 (errcode_for_file_access(),
928 errmsg("could not write two-phase state file: %m")));
931 /* Back up to prepare for rewriting the CRC */
932 if (lseek(fd, -((off_t) sizeof(pg_crc32)), SEEK_CUR) < 0)
936 (errcode_for_file_access(),
937 errmsg("could not seek in two-phase state file: %m")));
941 * The state file isn't valid yet, because we haven't written the correct
942 * CRC yet. Before we do that, insert entry in WAL and flush it to disk.
944 * Between the time we have written the WAL entry and the time we write
945 * out the correct state file CRC, we have an inconsistency: the xact is
946 * prepared according to WAL but not according to our on-disk state. We
947 * use a critical section to force a PANIC if we are unable to complete
948 * the write --- then, WAL replay should repair the inconsistency. The
949 * odds of a PANIC actually occurring should be very tiny given that we
950 * were able to write the bogus CRC above.
952 * We have to set inCommit here, too; otherwise a checkpoint starting
953 * immediately after the WAL record is inserted could complete without
954 * fsync'ing our state file. (This is essentially the same kind of race
955 * condition as the COMMIT-to-clog-write case that RecordTransactionCommit
956 * uses inCommit for; see notes there.)
958 * We save the PREPARE record's location in the gxact for later use by
959 * CheckPointTwoPhase.
961 START_CRIT_SECTION();
963 MyProc->inCommit = true;
965 gxact->prepare_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE,
967 XLogFlush(gxact->prepare_lsn);
969 /* If we crash now, we have prepared: WAL replay will fix things */
971 /* write correct CRC and close file */
972 if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
976 (errcode_for_file_access(),
977 errmsg("could not write two-phase state file: %m")));
982 (errcode_for_file_access(),
983 errmsg("could not close two-phase state file: %m")));
986 * Mark the prepared transaction as valid. As soon as xact.c marks MyProc
987 * as not running our XID (which it will do immediately after this
988 * function returns), others can commit/rollback the xact.
990 * NB: a side effect of this is to make a dummy ProcArray entry for the
991 * prepared XID. This must happen before we clear the XID from MyProc,
992 * else there is a window where the XID is not running according to
993 * TransactionIdIsInProgress, and onlookers would be entitled to assume
994 * the xact crashed. Instead we have a window where the same XID appears
995 * twice in ProcArray, which is OK.
997 MarkAsPrepared(gxact);
1000 * Now we can mark ourselves as out of the commit critical section: a
1001 * checkpoint starting after this will certainly see the gxact as a
1002 * candidate for fsyncing.
1004 MyProc->inCommit = false;
1008 records.tail = records.head = NULL;
1012 * Register a 2PC record to be written to state file.
1015 RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
1016 const void *data, uint32 len)
1018 TwoPhaseRecordOnDisk record;
1023 save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
1025 save_state_data(data, len);
1030 * Read and validate the state file for xid.
1032 * If it looks OK (has a valid magic number and CRC), return the palloc'd
1033 * contents of the file. Otherwise return NULL.
1036 ReadTwoPhaseFile(TransactionId xid)
1038 char path[MAXPGPATH];
1040 TwoPhaseFileHeader *hdr;
1047 TwoPhaseFilePath(path, xid);
1049 fd = BasicOpenFile(path, O_RDONLY | PG_BINARY, 0);
1053 (errcode_for_file_access(),
1054 errmsg("could not open two-phase state file \"%s\": %m",
1060 * Check file length. We can determine a lower bound pretty easily. We
1061 * set an upper bound to avoid palloc() failure on a corrupt file, though
1062 * we can't guarantee that we won't get an out of memory error anyway,
1063 * even on a valid file.
1065 if (fstat(fd, &stat))
1069 (errcode_for_file_access(),
1070 errmsg("could not stat two-phase state file \"%s\": %m",
1075 if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
1076 MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
1077 sizeof(pg_crc32)) ||
1078 stat.st_size > MaxAllocSize)
1084 crc_offset = stat.st_size - sizeof(pg_crc32);
1085 if (crc_offset != MAXALIGN(crc_offset))
1092 * OK, slurp in the file.
1094 buf = (char *) palloc(stat.st_size);
1096 if (read(fd, buf, stat.st_size) != stat.st_size)
1100 (errcode_for_file_access(),
1101 errmsg("could not read two-phase state file \"%s\": %m",
1109 hdr = (TwoPhaseFileHeader *) buf;
1110 if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
1116 INIT_CRC32(calc_crc);
1117 COMP_CRC32(calc_crc, buf, crc_offset);
1118 FIN_CRC32(calc_crc);
1120 file_crc = *((pg_crc32 *) (buf + crc_offset));
1122 if (!EQ_CRC32(calc_crc, file_crc))
1133 * FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
1136 FinishPreparedTransaction(const char *gid, bool isCommit)
1138 GlobalTransaction gxact;
1142 TwoPhaseFileHeader *hdr;
1143 TransactionId latestXid;
1144 TransactionId *children;
1145 RelFileNode *commitrels;
1146 RelFileNode *abortrels;
1147 RelFileNode *delrels;
1152 * Validate the GID, and lock the GXACT to ensure that two backends do not
1153 * try to commit the same GID at once.
1155 gxact = LockGXact(gid, GetUserId());
1156 xid = gxact->proc.xid;
1159 * Read and validate the state file
1161 buf = ReadTwoPhaseFile(xid);
1164 (errcode(ERRCODE_DATA_CORRUPTED),
1165 errmsg("two-phase state file for transaction %u is corrupt",
1169 * Disassemble the header area
1171 hdr = (TwoPhaseFileHeader *) buf;
1172 Assert(TransactionIdEquals(hdr->xid, xid));
1173 bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
1174 children = (TransactionId *) bufptr;
1175 bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
1176 commitrels = (RelFileNode *) bufptr;
1177 bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
1178 abortrels = (RelFileNode *) bufptr;
1179 bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
1181 /* compute latestXid among all children */
1182 latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
1185 * The order of operations here is critical: make the XLOG entry for
1186 * commit or abort, then mark the transaction committed or aborted in
1187 * pg_clog, then remove its PGPROC from the global ProcArray (which means
1188 * TransactionIdIsInProgress will stop saying the prepared xact is in
1189 * progress), then run the post-commit or post-abort callbacks. The
1190 * callbacks will release the locks the transaction held.
1193 RecordTransactionCommitPrepared(xid,
1194 hdr->nsubxacts, children,
1195 hdr->ncommitrels, commitrels);
1197 RecordTransactionAbortPrepared(xid,
1198 hdr->nsubxacts, children,
1199 hdr->nabortrels, abortrels);
1201 ProcArrayRemove(&gxact->proc, latestXid);
1204 * In case we fail while running the callbacks, mark the gxact invalid so
1205 * no one else will try to commit/rollback, and so it can be recycled
1206 * properly later. It is still locked by our XID so it won't go away yet.
1208 * (We assume it's safe to do this without taking TwoPhaseStateLock.)
1210 gxact->valid = false;
1213 * We have to remove any files that were supposed to be dropped. For
1214 * consistency with the regular xact.c code paths, must do this before
1215 * releasing locks, so do it before running the callbacks.
1217 * NB: this code knows that we couldn't be dropping any temp rels ...
1221 delrels = commitrels;
1222 ndelrels = hdr->ncommitrels;
1226 delrels = abortrels;
1227 ndelrels = hdr->nabortrels;
1229 for (i = 0; i < ndelrels; i++)
1231 SMgrRelation srel = smgropen(delrels[i]);
1234 for (fork = 0; fork <= MAX_FORKNUM; fork++)
1236 if (smgrexists(srel, fork))
1238 XLogDropRelation(delrels[i], fork);
1239 smgrdounlink(srel, fork, false, true);
1245 /* And now do the callbacks */
1247 ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
1249 ProcessRecords(bufptr, xid, twophase_postabort_callbacks);
1251 /* Count the prepared xact as committed or aborted */
1252 AtEOXact_PgStat(isCommit);
1255 * And now we can clean up our mess.
1257 RemoveTwoPhaseFile(xid, true);
1265 * Scan a 2PC state file (already read into memory by ReadTwoPhaseFile)
1266 * and call the indicated callbacks for each 2PC record.
1269 ProcessRecords(char *bufptr, TransactionId xid,
1270 const TwoPhaseCallback callbacks[])
1274 TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;
1276 Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
1277 if (record->rmid == TWOPHASE_RM_END_ID)
1280 bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));
1282 if (callbacks[record->rmid] != NULL)
1283 callbacks[record->rmid] (xid, record->info,
1284 (void *) bufptr, record->len);
1286 bufptr += MAXALIGN(record->len);
1291 * Remove the 2PC file for the specified XID.
1293 * If giveWarning is false, do not complain about file-not-present;
1294 * this is an expected case during WAL replay.
1297 RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
1299 char path[MAXPGPATH];
1301 TwoPhaseFilePath(path, xid);
1303 if (errno != ENOENT || giveWarning)
1305 (errcode_for_file_access(),
1306 errmsg("could not remove two-phase state file \"%s\": %m",
1311 * Recreates a state file. This is used in WAL replay.
1313 * Note: content and len don't include CRC.
1316 RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
1318 char path[MAXPGPATH];
1319 pg_crc32 statefile_crc;
1323 INIT_CRC32(statefile_crc);
1324 COMP_CRC32(statefile_crc, content, len);
1325 FIN_CRC32(statefile_crc);
1327 TwoPhaseFilePath(path, xid);
1329 fd = BasicOpenFile(path,
1330 O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY,
1334 (errcode_for_file_access(),
1335 errmsg("could not recreate two-phase state file \"%s\": %m",
1338 /* Write content and CRC */
1339 if (write(fd, content, len) != len)
1343 (errcode_for_file_access(),
1344 errmsg("could not write two-phase state file: %m")));
1346 if (write(fd, &statefile_crc, sizeof(pg_crc32)) != sizeof(pg_crc32))
1350 (errcode_for_file_access(),
1351 errmsg("could not write two-phase state file: %m")));
1355 * We must fsync the file because the end-of-replay checkpoint will not do
1356 * so, there being no GXACT in shared memory yet to tell it to.
1358 if (pg_fsync(fd) != 0)
1362 (errcode_for_file_access(),
1363 errmsg("could not fsync two-phase state file: %m")));
1368 (errcode_for_file_access(),
1369 errmsg("could not close two-phase state file: %m")));
1373 * CheckPointTwoPhase -- handle 2PC component of checkpointing.
1375 * We must fsync the state file of any GXACT that is valid and has a PREPARE
1376 * LSN <= the checkpoint's redo horizon. (If the gxact isn't valid yet or
1377 * has a later LSN, this checkpoint is not responsible for fsyncing it.)
1379 * This is deliberately run as late as possible in the checkpoint sequence,
1380 * because GXACTs ordinarily have short lifespans, and so it is quite
1381 * possible that GXACTs that were valid at checkpoint start will no longer
1382 * exist if we wait a little bit.
1384 * If a GXACT remains valid across multiple checkpoints, it'll be fsynced
1385 * each time. This is considered unusual enough that we don't bother to
1386 * expend any extra code to avoid the redundant fsyncs. (They should be
1387 * reasonably cheap anyway, since they won't cause I/O.)
1390 CheckPointTwoPhase(XLogRecPtr redo_horizon)
1392 TransactionId *xids;
1394 char path[MAXPGPATH];
1398 * We don't want to hold the TwoPhaseStateLock while doing I/O, so we grab
1399 * it just long enough to make a list of the XIDs that require fsyncing,
1400 * and then do the I/O afterwards.
1402 * This approach creates a race condition: someone else could delete a
1403 * GXACT between the time we release TwoPhaseStateLock and the time we try
1404 * to open its state file. We handle this by special-casing ENOENT
1405 * failures: if we see that, we verify that the GXACT is no longer valid,
1406 * and if so ignore the failure.
1408 if (max_prepared_xacts <= 0)
1409 return; /* nothing to do */
1411 TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();
1413 xids = (TransactionId *) palloc(max_prepared_xacts * sizeof(TransactionId));
1416 LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
1418 for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
1420 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
1423 XLByteLE(gxact->prepare_lsn, redo_horizon))
1424 xids[nxids++] = gxact->proc.xid;
1427 LWLockRelease(TwoPhaseStateLock);
1429 for (i = 0; i < nxids; i++)
1431 TransactionId xid = xids[i];
1434 TwoPhaseFilePath(path, xid);
1436 fd = BasicOpenFile(path, O_RDWR | PG_BINARY, 0);
1439 if (errno == ENOENT)
1441 /* OK if gxact is no longer valid */
1442 if (!TransactionIdIsPrepared(xid))
1444 /* Restore errno in case it was changed */
1448 (errcode_for_file_access(),
1449 errmsg("could not open two-phase state file \"%s\": %m",
1453 if (pg_fsync(fd) != 0)
1457 (errcode_for_file_access(),
1458 errmsg("could not fsync two-phase state file \"%s\": %m",
1464 (errcode_for_file_access(),
1465 errmsg("could not close two-phase state file \"%s\": %m",
1471 TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
1475 * PrescanPreparedTransactions
1477 * Scan the pg_twophase directory and determine the range of valid XIDs
1478 * present. This is run during database startup, after we have completed
1479 * reading WAL. ShmemVariableCache->nextXid has been set to one more than
1480 * the highest XID for which evidence exists in WAL.
1482 * We throw away any prepared xacts with main XID beyond nextXid --- if any
1483 * are present, it suggests that the DBA has done a PITR recovery to an
1484 * earlier point in time without cleaning out pg_twophase. We dare not
1485 * try to recover such prepared xacts since they likely depend on database
1486 * state that doesn't exist now.
1488 * However, we will advance nextXid beyond any subxact XIDs belonging to
1489 * valid prepared xacts. We need to do this since subxact commit doesn't
1490 * write a WAL entry, and so there might be no evidence in WAL of those
1493 * Our other responsibility is to determine and return the oldest valid XID
1494 * among the prepared xacts (if none, return ShmemVariableCache->nextXid).
1495 * This is needed to synchronize pg_subtrans startup properly.
1498 PrescanPreparedTransactions(void)
1500 TransactionId origNextXid = ShmemVariableCache->nextXid;
1501 TransactionId result = origNextXid;
1503 struct dirent *clde;
1505 cldir = AllocateDir(TWOPHASE_DIR);
1506 while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
1508 if (strlen(clde->d_name) == 8 &&
1509 strspn(clde->d_name, "0123456789ABCDEF") == 8)
1513 TwoPhaseFileHeader *hdr;
1514 TransactionId *subxids;
1517 xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
1519 /* Reject XID if too new */
1520 if (TransactionIdFollowsOrEquals(xid, origNextXid))
1523 (errmsg("removing future two-phase state file \"%s\"",
1525 RemoveTwoPhaseFile(xid, true);
1530 * Note: we can't check if already processed because clog
1531 * subsystem isn't up yet.
1534 /* Read and validate file */
1535 buf = ReadTwoPhaseFile(xid);
1539 (errmsg("removing corrupt two-phase state file \"%s\"",
1541 RemoveTwoPhaseFile(xid, true);
1545 /* Deconstruct header */
1546 hdr = (TwoPhaseFileHeader *) buf;
1547 if (!TransactionIdEquals(hdr->xid, xid))
1550 (errmsg("removing corrupt two-phase state file \"%s\"",
1552 RemoveTwoPhaseFile(xid, true);
1558 * OK, we think this file is valid. Incorporate xid into the
1559 * running-minimum result.
1561 if (TransactionIdPrecedes(xid, result))
1565 * Examine subtransaction XIDs ... they should all follow main
1566 * XID, and they may force us to advance nextXid.
1568 subxids = (TransactionId *)
1569 (buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
1570 for (i = 0; i < hdr->nsubxacts; i++)
1572 TransactionId subxid = subxids[i];
1574 Assert(TransactionIdFollows(subxid, xid));
1575 if (TransactionIdFollowsOrEquals(subxid,
1576 ShmemVariableCache->nextXid))
1578 ShmemVariableCache->nextXid = subxid;
1579 TransactionIdAdvance(ShmemVariableCache->nextXid);
1592 * RecoverPreparedTransactions
1594 * Scan the pg_twophase directory and reload shared-memory state for each
1595 * prepared transaction (reacquire locks, etc). This is run during database
1599 RecoverPreparedTransactions(void)
1601 char dir[MAXPGPATH];
1603 struct dirent *clde;
1605 snprintf(dir, MAXPGPATH, "%s", TWOPHASE_DIR);
1607 cldir = AllocateDir(dir);
1608 while ((clde = ReadDir(cldir, dir)) != NULL)
1610 if (strlen(clde->d_name) == 8 &&
1611 strspn(clde->d_name, "0123456789ABCDEF") == 8)
1616 TwoPhaseFileHeader *hdr;
1617 TransactionId *subxids;
1618 GlobalTransaction gxact;
1621 xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
1623 /* Already processed? */
1624 if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
1627 (errmsg("removing stale two-phase state file \"%s\"",
1629 RemoveTwoPhaseFile(xid, true);
1633 /* Read and validate file */
1634 buf = ReadTwoPhaseFile(xid);
1638 (errmsg("removing corrupt two-phase state file \"%s\"",
1640 RemoveTwoPhaseFile(xid, true);
1645 (errmsg("recovering prepared transaction %u", xid)));
1647 /* Deconstruct header */
1648 hdr = (TwoPhaseFileHeader *) buf;
1649 Assert(TransactionIdEquals(hdr->xid, xid));
1650 bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
1651 subxids = (TransactionId *) bufptr;
1652 bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
1653 bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
1654 bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
1657 * Reconstruct subtrans state for the transaction --- needed
1658 * because pg_subtrans is not preserved over a restart. Note that
1659 * we are linking all the subtransactions directly to the
1660 * top-level XID; there may originally have been a more complex
1661 * hierarchy, but there's no need to restore that exactly.
1663 for (i = 0; i < hdr->nsubxacts; i++)
1664 SubTransSetParent(subxids[i], xid);
1667 * Recreate its GXACT and dummy PGPROC
1669 * Note: since we don't have the PREPARE record's WAL location at
1670 * hand, we leave prepare_lsn zeroes. This means the GXACT will
1671 * be fsync'd on every future checkpoint. We assume this
1672 * situation is infrequent enough that the performance cost is
1673 * negligible (especially since we know the state file has already
1676 gxact = MarkAsPreparing(xid, hdr->gid,
1678 hdr->owner, hdr->database);
1679 GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
1680 MarkAsPrepared(gxact);
1683 * Recover other state (notably locks) using resource managers
1685 ProcessRecords(bufptr, xid, twophase_recover_callbacks);
1694 * RecordTransactionCommitPrepared
1696 * This is basically the same as RecordTransactionCommit: in particular,
1697 * we must set the inCommit flag to avoid a race condition.
1699 * We know the transaction made at least one XLOG entry (its PREPARE),
1700 * so it is never possible to optimize out the commit record.
1703 RecordTransactionCommitPrepared(TransactionId xid,
1705 TransactionId *children,
1709 XLogRecData rdata[3];
1711 xl_xact_commit_prepared xlrec;
1714 START_CRIT_SECTION();
1716 /* See notes in RecordTransactionCommit */
1717 MyProc->inCommit = true;
1719 /* Emit the XLOG commit record */
1721 xlrec.crec.xact_time = GetCurrentTimestamp();
1722 xlrec.crec.nrels = nrels;
1723 xlrec.crec.nsubxacts = nchildren;
1724 rdata[0].data = (char *) (&xlrec);
1725 rdata[0].len = MinSizeOfXactCommitPrepared;
1726 rdata[0].buffer = InvalidBuffer;
1727 /* dump rels to delete */
1730 rdata[0].next = &(rdata[1]);
1731 rdata[1].data = (char *) rels;
1732 rdata[1].len = nrels * sizeof(RelFileNode);
1733 rdata[1].buffer = InvalidBuffer;
1736 /* dump committed child Xids */
1739 rdata[lastrdata].next = &(rdata[2]);
1740 rdata[2].data = (char *) children;
1741 rdata[2].len = nchildren * sizeof(TransactionId);
1742 rdata[2].buffer = InvalidBuffer;
1745 rdata[lastrdata].next = NULL;
1747 recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT_PREPARED, rdata);
1750 * We don't currently try to sleep before flush here ... nor is there any
1751 * support for async commit of a prepared xact (the very idea is probably
1755 /* Flush XLOG to disk */
1758 /* Mark the transaction committed in pg_clog */
1759 TransactionIdCommitTree(xid, nchildren, children);
1761 /* Checkpoint can proceed now */
1762 MyProc->inCommit = false;
1768 * RecordTransactionAbortPrepared
1770 * This is basically the same as RecordTransactionAbort.
1772 * We know the transaction made at least one XLOG entry (its PREPARE),
1773 * so it is never possible to optimize out the abort record.
1776 RecordTransactionAbortPrepared(TransactionId xid,
1778 TransactionId *children,
1782 XLogRecData rdata[3];
1784 xl_xact_abort_prepared xlrec;
1788 * Catch the scenario where we aborted partway through
1789 * RecordTransactionCommitPrepared ...
1791 if (TransactionIdDidCommit(xid))
1792 elog(PANIC, "cannot abort transaction %u, it was already committed",
1795 START_CRIT_SECTION();
1797 /* Emit the XLOG abort record */
1799 xlrec.arec.xact_time = GetCurrentTimestamp();
1800 xlrec.arec.nrels = nrels;
1801 xlrec.arec.nsubxacts = nchildren;
1802 rdata[0].data = (char *) (&xlrec);
1803 rdata[0].len = MinSizeOfXactAbortPrepared;
1804 rdata[0].buffer = InvalidBuffer;
1805 /* dump rels to delete */
1808 rdata[0].next = &(rdata[1]);
1809 rdata[1].data = (char *) rels;
1810 rdata[1].len = nrels * sizeof(RelFileNode);
1811 rdata[1].buffer = InvalidBuffer;
1814 /* dump committed child Xids */
1817 rdata[lastrdata].next = &(rdata[2]);
1818 rdata[2].data = (char *) children;
1819 rdata[2].len = nchildren * sizeof(TransactionId);
1820 rdata[2].buffer = InvalidBuffer;
1823 rdata[lastrdata].next = NULL;
1825 recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ABORT_PREPARED, rdata);
1827 /* Always flush, since we're about to remove the 2PC state file */
1831 * Mark the transaction aborted in clog. This is not absolutely necessary
1832 * but we may as well do it while we are here.
1834 TransactionIdAbortTree(xid, nchildren, children);