granicus.if.org Git - postgresql/blob - src/backend/access/transam/twophase.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * twophase.c
   4  *              Two-phase commit support functions.
   5  *
   6  * Portions Copyright (c) 1996-2012, 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 PGXACT and PGPROC; this is
  25  *              what keeps the XID considered running by TransactionIdIsInProgress.
  26  *              It is also convenient as a PGPROC to hook the gxact's locks to.
  27  *
  28  *              In order to survive crashes and shutdowns, all prepared
  29  *              transactions must be stored in permanent storage. This includes
  30  *              locking information, pending notifications etc. All that state
  31  *              information is written to the per-transaction state file in
  32  *              the pg_twophase directory.
  33  *
  34  *-------------------------------------------------------------------------
  35  */
  36 #include "postgres.h"
  37
  38 #include <fcntl.h>
  39 #include <sys/stat.h>
  40 #include <sys/types.h>
  41 #include <time.h>
  42 #include <unistd.h>
  43
  44 #include "access/htup_details.h"
  45 #include "access/subtrans.h"
  46 #include "access/transam.h"
  47 #include "access/twophase.h"
  48 #include "access/twophase_rmgr.h"
  49 #include "access/xact.h"
  50 #include "access/xlogutils.h"
  51 #include "catalog/pg_type.h"
  52 #include "catalog/storage.h"
  53 #include "funcapi.h"
  54 #include "miscadmin.h"
  55 #include "pg_trace.h"
  56 #include "pgstat.h"
  57 #include "replication/walsender.h"
  58 #include "replication/syncrep.h"
  59 #include "storage/fd.h"
  60 #include "storage/predicate.h"
  61 #include "storage/proc.h"
  62 #include "storage/procarray.h"
  63 #include "storage/sinvaladt.h"
  64 #include "storage/smgr.h"
  65 #include "utils/builtins.h"
  66 #include "utils/memutils.h"
  67 #include "utils/timestamp.h"
  68
  69
  70 /*
  71  * Directory where Two-phase commit files reside within PGDATA
  72  */
  73 #define TWOPHASE_DIR "pg_twophase"
  74
  75 /* GUC variable, can't be changed after startup */
  76 int                     max_prepared_xacts = 0;
  77
  78 /*
  79  * This struct describes one global transaction that is in prepared state
  80  * or attempting to become prepared.
  81  *
  82  * The lifecycle of a global transaction is:
  83  *
  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.
  87  *
  88  * 2. After successfully completing prepare, set valid = true and enter the
  89  * referenced PGPROC into the global ProcArray.
  90  *
  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.
  95  *
  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
  98  * the freelist.
  99  *
 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.
 103  *
 104  * typedef struct GlobalTransactionData *GlobalTransaction appears in
 105  * twophase.h
 106  */
 107 #define GIDSIZE 200
 108
 109 typedef struct GlobalTransactionData
 110 {
 111         GlobalTransaction next;         /* list link for free list */
 112         int                     pgprocno;               /* ID of associated dummy PGPROC */
 113         BackendId       dummyBackendId; /* similar to backend id for backends */
 114         TimestampTz prepared_at;        /* time of preparation */
 115         XLogRecPtr      prepare_lsn;    /* XLOG offset of prepare record */
 116         Oid                     owner;                  /* ID of user that executed the xact */
 117         TransactionId locking_xid;      /* top-level XID of backend working on xact */
 118         bool            valid;                  /* TRUE if fully prepared */
 119         char            gid[GIDSIZE];   /* The GID assigned to the prepared xact */
 120 }       GlobalTransactionData;
 121
 122 /*
 123  * Two Phase Commit shared state.  Access to this struct is protected
 124  * by TwoPhaseStateLock.
 125  */
 126 typedef struct TwoPhaseStateData
 127 {
 128         /* Head of linked list of free GlobalTransactionData structs */
 129         GlobalTransaction freeGXacts;
 130
 131         /* Number of valid prepXacts entries. */
 132         int                     numPrepXacts;
 133
 134         /*
 135          * There are max_prepared_xacts items in this array, but C wants a
 136          * fixed-size array.
 137          */
 138         GlobalTransaction prepXacts[1];         /* VARIABLE LENGTH ARRAY */
 139 } TwoPhaseStateData;                    /* VARIABLE LENGTH STRUCT */
 140
 141 static TwoPhaseStateData *TwoPhaseState;
 142
 143
 144 static void RecordTransactionCommitPrepared(TransactionId xid,
 145                                                                 int nchildren,
 146                                                                 TransactionId *children,
 147                                                                 int nrels,
 148                                                                 RelFileNode *rels,
 149                                                                 int ninvalmsgs,
 150                                                                 SharedInvalidationMessage *invalmsgs,
 151                                                                 bool initfileinval);
 152 static void RecordTransactionAbortPrepared(TransactionId xid,
 153                                                            int nchildren,
 154                                                            TransactionId *children,
 155                                                            int nrels,
 156                                                            RelFileNode *rels);
 157 static void ProcessRecords(char *bufptr, TransactionId xid,
 158                            const TwoPhaseCallback callbacks[]);
 159
 160
 161 /*
 162  * Initialization of shared memory
 163  */
 164 Size
 165 TwoPhaseShmemSize(void)
 166 {
 167         Size            size;
 168
 169         /* Need the fixed struct, the array of pointers, and the GTD structs */
 170         size = offsetof(TwoPhaseStateData, prepXacts);
 171         size = add_size(size, mul_size(max_prepared_xacts,
 172                                                                    sizeof(GlobalTransaction)));
 173         size = MAXALIGN(size);
 174         size = add_size(size, mul_size(max_prepared_xacts,
 175                                                                    sizeof(GlobalTransactionData)));
 176
 177         return size;
 178 }
 179
 180 void
 181 TwoPhaseShmemInit(void)
 182 {
 183         bool            found;
 184
 185         TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
 186                                                                         TwoPhaseShmemSize(),
 187                                                                         &found);
 188         if (!IsUnderPostmaster)
 189         {
 190                 GlobalTransaction gxacts;
 191                 int                     i;
 192
 193                 Assert(!found);
 194                 TwoPhaseState->freeGXacts = NULL;
 195                 TwoPhaseState->numPrepXacts = 0;
 196
 197                 /*
 198                  * Initialize the linked list of free GlobalTransactionData structs
 199                  */
 200                 gxacts = (GlobalTransaction)
 201                         ((char *) TwoPhaseState +
 202                          MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
 203                                           sizeof(GlobalTransaction) * max_prepared_xacts));
 204                 for (i = 0; i < max_prepared_xacts; i++)
 205                 {
 206                         /* insert into linked list */
 207                         gxacts[i].next = TwoPhaseState->freeGXacts;
 208                         TwoPhaseState->freeGXacts = &gxacts[i];
 209
 210                         /* associate it with a PGPROC assigned by InitProcGlobal */
 211                         gxacts[i].pgprocno = PreparedXactProcs[i].pgprocno;
 212
 213                         /*
 214                          * Assign a unique ID for each dummy proc, so that the range of
 215                          * dummy backend IDs immediately follows the range of normal
 216                          * backend IDs. We don't dare to assign a real backend ID to dummy
 217                          * procs, because prepared transactions don't take part in cache
 218                          * invalidation like a real backend ID would imply, but having a
 219                          * unique ID for them is nevertheless handy. This arrangement
 220                          * allows you to allocate an array of size (MaxBackends +
 221                          * max_prepared_xacts + 1), and have a slot for every backend and
 222                          * prepared transaction. Currently multixact.c uses that
 223                          * technique.
 224                          */
 225                         gxacts[i].dummyBackendId = MaxBackends + 1 + i;
 226                 }
 227         }
 228         else
 229                 Assert(found);
 230 }
 231
 232
 233 /*
 234  * MarkAsPreparing
 235  *              Reserve the GID for the given transaction.
 236  *
 237  * Internally, this creates a gxact struct and puts it into the active array.
 238  * NOTE: this is also used when reloading a gxact after a crash; so avoid
 239  * assuming that we can use very much backend context.
 240  */
 241 GlobalTransaction
 242 MarkAsPreparing(TransactionId xid, const char *gid,
 243                                 TimestampTz prepared_at, Oid owner, Oid databaseid)
 244 {
 245         GlobalTransaction gxact;
 246         PGPROC     *proc;
 247         PGXACT     *pgxact;
 248         int                     i;
 249
 250         if (strlen(gid) >= GIDSIZE)
 251                 ereport(ERROR,
 252                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 253                                  errmsg("transaction identifier \"%s\" is too long",
 254                                                 gid)));
 255
 256         /* fail immediately if feature is disabled */
 257         if (max_prepared_xacts == 0)
 258                 ereport(ERROR,
 259                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 260                                  errmsg("prepared transactions are disabled"),
 261                           errhint("Set max_prepared_transactions to a nonzero value.")));
 262
 263         LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 264
 265         /*
 266          * First, find and recycle any gxacts that failed during prepare. We do
 267          * this partly to ensure we don't mistakenly say their GIDs are still
 268          * reserved, and partly so we don't fail on out-of-slots unnecessarily.
 269          */
 270         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 271         {
 272                 gxact = TwoPhaseState->prepXacts[i];
 273                 if (!gxact->valid && !TransactionIdIsActive(gxact->locking_xid))
 274                 {
 275                         /* It's dead Jim ... remove from the active array */
 276                         TwoPhaseState->numPrepXacts--;
 277                         TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
 278                         /* and put it back in the freelist */
 279                         gxact->next = TwoPhaseState->freeGXacts;
 280                         TwoPhaseState->freeGXacts = gxact;
 281                         /* Back up index count too, so we don't miss scanning one */
 282                         i--;
 283                 }
 284         }
 285
 286         /* Check for conflicting GID */
 287         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 288         {
 289                 gxact = TwoPhaseState->prepXacts[i];
 290                 if (strcmp(gxact->gid, gid) == 0)
 291                 {
 292                         ereport(ERROR,
 293                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
 294                                          errmsg("transaction identifier \"%s\" is already in use",
 295                                                         gid)));
 296                 }
 297         }
 298
 299         /* Get a free gxact from the freelist */
 300         if (TwoPhaseState->freeGXacts == NULL)
 301                 ereport(ERROR,
 302                                 (errcode(ERRCODE_OUT_OF_MEMORY),
 303                                  errmsg("maximum number of prepared transactions reached"),
 304                                  errhint("Increase max_prepared_transactions (currently %d).",
 305                                                  max_prepared_xacts)));
 306         gxact = TwoPhaseState->freeGXacts;
 307         TwoPhaseState->freeGXacts = gxact->next;
 308
 309         proc = &ProcGlobal->allProcs[gxact->pgprocno];
 310         pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 311
 312         /* Initialize the PGPROC entry */
 313         MemSet(proc, 0, sizeof(PGPROC));
 314         proc->pgprocno = gxact->pgprocno;
 315         SHMQueueElemInit(&(proc->links));
 316         proc->waitStatus = STATUS_OK;
 317         /* We set up the gxact's VXID as InvalidBackendId/XID */
 318         proc->lxid = (LocalTransactionId) xid;
 319         pgxact->xid = xid;
 320         pgxact->xmin = InvalidTransactionId;
 321         pgxact->inCommit = false;
 322         pgxact->vacuumFlags = 0;
 323         proc->pid = 0;
 324         proc->backendId = InvalidBackendId;
 325         proc->databaseId = databaseid;
 326         proc->roleId = owner;
 327         proc->lwWaiting = false;
 328         proc->lwWaitMode = 0;
 329         proc->lwWaitLink = NULL;
 330         proc->waitLock = NULL;
 331         proc->waitProcLock = NULL;
 332         for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
 333                 SHMQueueInit(&(proc->myProcLocks[i]));
 334         /* subxid data must be filled later by GXactLoadSubxactData */
 335         pgxact->overflowed = false;
 336         pgxact->nxids = 0;
 337
 338         gxact->prepared_at = prepared_at;
 339         /* initialize LSN to 0 (start of WAL) */
 340         gxact->prepare_lsn = 0;
 341         gxact->owner = owner;
 342         gxact->locking_xid = xid;
 343         gxact->valid = false;
 344         strcpy(gxact->gid, gid);
 345
 346         /* And insert it into the active array */
 347         Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
 348         TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
 349
 350         LWLockRelease(TwoPhaseStateLock);
 351
 352         return gxact;
 353 }
 354
 355 /*
 356  * GXactLoadSubxactData
 357  *
 358  * If the transaction being persisted had any subtransactions, this must
 359  * be called before MarkAsPrepared() to load information into the dummy
 360  * PGPROC.
 361  */
 362 static void
 363 GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
 364                                          TransactionId *children)
 365 {
 366         PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
 367         PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 368
 369         /* We need no extra lock since the GXACT isn't valid yet */
 370         if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
 371         {
 372                 pgxact->overflowed = true;
 373                 nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
 374         }
 375         if (nsubxacts > 0)
 376         {
 377                 memcpy(proc->subxids.xids, children,
 378                            nsubxacts * sizeof(TransactionId));
 379                 pgxact->nxids = nsubxacts;
 380         }
 381 }
 382
 383 /*
 384  * MarkAsPrepared
 385  *              Mark the GXACT as fully valid, and enter it into the global ProcArray.
 386  */
 387 static void
 388 MarkAsPrepared(GlobalTransaction gxact)
 389 {
 390         /* Lock here may be overkill, but I'm not convinced of that ... */
 391         LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 392         Assert(!gxact->valid);
 393         gxact->valid = true;
 394         LWLockRelease(TwoPhaseStateLock);
 395
 396         /*
 397          * Put it into the global ProcArray so TransactionIdIsInProgress considers
 398          * the XID as still running.
 399          */
 400         ProcArrayAdd(&ProcGlobal->allProcs[gxact->pgprocno]);
 401 }
 402
 403 /*
 404  * LockGXact
 405  *              Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
 406  */
 407 static GlobalTransaction
 408 LockGXact(const char *gid, Oid user)
 409 {
 410         int                     i;
 411
 412         LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 413
 414         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 415         {
 416                 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 417                 PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
 418
 419                 /* Ignore not-yet-valid GIDs */
 420                 if (!gxact->valid)
 421                         continue;
 422                 if (strcmp(gxact->gid, gid) != 0)
 423                         continue;
 424
 425                 /* Found it, but has someone else got it locked? */
 426                 if (TransactionIdIsValid(gxact->locking_xid))
 427                 {
 428                         if (TransactionIdIsActive(gxact->locking_xid))
 429                                 ereport(ERROR,
 430                                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 431                                 errmsg("prepared transaction with identifier \"%s\" is busy",
 432                                            gid)));
 433                         gxact->locking_xid = InvalidTransactionId;
 434                 }
 435
 436                 if (user != gxact->owner && !superuser_arg(user))
 437                         ereport(ERROR,
 438                                         (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 439                                   errmsg("permission denied to finish prepared transaction"),
 440                                          errhint("Must be superuser or the user that prepared the transaction.")));
 441
 442                 /*
 443                  * Note: it probably would be possible to allow committing from
 444                  * another database; but at the moment NOTIFY is known not to work and
 445                  * there may be some other issues as well.      Hence disallow until
 446                  * someone gets motivated to make it work.
 447                  */
 448                 if (MyDatabaseId != proc->databaseId)
 449                         ereport(ERROR,
 450                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 451                                   errmsg("prepared transaction belongs to another database"),
 452                                          errhint("Connect to the database where the transaction was prepared to finish it.")));
 453
 454                 /* OK for me to lock it */
 455                 gxact->locking_xid = GetTopTransactionId();
 456
 457                 LWLockRelease(TwoPhaseStateLock);
 458
 459                 return gxact;
 460         }
 461
 462         LWLockRelease(TwoPhaseStateLock);
 463
 464         ereport(ERROR,
 465                         (errcode(ERRCODE_UNDEFINED_OBJECT),
 466                  errmsg("prepared transaction with identifier \"%s\" does not exist",
 467                                 gid)));
 468
 469         /* NOTREACHED */
 470         return NULL;
 471 }
 472
 473 /*
 474  * RemoveGXact
 475  *              Remove the prepared transaction from the shared memory array.
 476  *
 477  * NB: caller should have already removed it from ProcArray
 478  */
 479 static void
 480 RemoveGXact(GlobalTransaction gxact)
 481 {
 482         int                     i;
 483
 484         LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 485
 486         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 487         {
 488                 if (gxact == TwoPhaseState->prepXacts[i])
 489                 {
 490                         /* remove from the active array */
 491                         TwoPhaseState->numPrepXacts--;
 492                         TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
 493
 494                         /* and put it back in the freelist */
 495                         gxact->next = TwoPhaseState->freeGXacts;
 496                         TwoPhaseState->freeGXacts = gxact;
 497
 498                         LWLockRelease(TwoPhaseStateLock);
 499
 500                         return;
 501                 }
 502         }
 503
 504         LWLockRelease(TwoPhaseStateLock);
 505
 506         elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
 507 }
 508
 509 /*
 510  * TransactionIdIsPrepared
 511  *              True iff transaction associated with the identifier is prepared
 512  *              for two-phase commit
 513  *
 514  * Note: only gxacts marked "valid" are considered; but notice we do not
 515  * check the locking status.
 516  *
 517  * This is not currently exported, because it is only needed internally.
 518  */
 519 static bool
 520 TransactionIdIsPrepared(TransactionId xid)
 521 {
 522         bool            result = false;
 523         int                     i;
 524
 525         LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 526
 527         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 528         {
 529                 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 530                 PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 531
 532                 if (gxact->valid && pgxact->xid == xid)
 533                 {
 534                         result = true;
 535                         break;
 536                 }
 537         }
 538
 539         LWLockRelease(TwoPhaseStateLock);
 540
 541         return result;
 542 }
 543
 544 /*
 545  * Returns an array of all prepared transactions for the user-level
 546  * function pg_prepared_xact.
 547  *
 548  * The returned array and all its elements are copies of internal data
 549  * structures, to minimize the time we need to hold the TwoPhaseStateLock.
 550  *
 551  * WARNING -- we return even those transactions that are not fully prepared
 552  * yet.  The caller should filter them out if he doesn't want them.
 553  *
 554  * The returned array is palloc'd.
 555  */
 556 static int
 557 GetPreparedTransactionList(GlobalTransaction *gxacts)
 558 {
 559         GlobalTransaction array;
 560         int                     num;
 561         int                     i;
 562
 563         LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 564
 565         if (TwoPhaseState->numPrepXacts == 0)
 566         {
 567                 LWLockRelease(TwoPhaseStateLock);
 568
 569                 *gxacts = NULL;
 570                 return 0;
 571         }
 572
 573         num = TwoPhaseState->numPrepXacts;
 574         array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
 575         *gxacts = array;
 576         for (i = 0; i < num; i++)
 577                 memcpy(array + i, TwoPhaseState->prepXacts[i],
 578                            sizeof(GlobalTransactionData));
 579
 580         LWLockRelease(TwoPhaseStateLock);
 581
 582         return num;
 583 }
 584
 585
 586 /* Working status for pg_prepared_xact */
 587 typedef struct
 588 {
 589         GlobalTransaction array;
 590         int                     ngxacts;
 591         int                     currIdx;
 592 } Working_State;
 593
 594 /*
 595  * pg_prepared_xact
 596  *              Produce a view with one row per prepared transaction.
 597  *
 598  * This function is here so we don't have to export the
 599  * GlobalTransactionData struct definition.
 600  */
 601 Datum
 602 pg_prepared_xact(PG_FUNCTION_ARGS)
 603 {
 604         FuncCallContext *funcctx;
 605         Working_State *status;
 606
 607         if (SRF_IS_FIRSTCALL())
 608         {
 609                 TupleDesc       tupdesc;
 610                 MemoryContext oldcontext;
 611
 612                 /* create a function context for cross-call persistence */
 613                 funcctx = SRF_FIRSTCALL_INIT();
 614
 615                 /*
 616                  * Switch to memory context appropriate for multiple function calls
 617                  */
 618                 oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
 619
 620                 /* build tupdesc for result tuples */
 621                 /* this had better match pg_prepared_xacts view in system_views.sql */
 622                 tupdesc = CreateTemplateTupleDesc(5, false);
 623                 TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
 624                                                    XIDOID, -1, 0);
 625                 TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
 626                                                    TEXTOID, -1, 0);
 627                 TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
 628                                                    TIMESTAMPTZOID, -1, 0);
 629                 TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
 630                                                    OIDOID, -1, 0);
 631                 TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
 632                                                    OIDOID, -1, 0);
 633
 634                 funcctx->tuple_desc = BlessTupleDesc(tupdesc);
 635
 636                 /*
 637                  * Collect all the 2PC status information that we will format and send
 638                  * out as a result set.
 639                  */
 640                 status = (Working_State *) palloc(sizeof(Working_State));
 641                 funcctx->user_fctx = (void *) status;
 642
 643                 status->ngxacts = GetPreparedTransactionList(&status->array);
 644                 status->currIdx = 0;
 645
 646                 MemoryContextSwitchTo(oldcontext);
 647         }
 648
 649         funcctx = SRF_PERCALL_SETUP();
 650         status = (Working_State *) funcctx->user_fctx;
 651
 652         while (status->array != NULL && status->currIdx < status->ngxacts)
 653         {
 654                 GlobalTransaction gxact = &status->array[status->currIdx++];
 655                 PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
 656                 PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 657                 Datum           values[5];
 658                 bool            nulls[5];
 659                 HeapTuple       tuple;
 660                 Datum           result;
 661
 662                 if (!gxact->valid)
 663                         continue;
 664
 665                 /*
 666                  * Form tuple with appropriate data.
 667                  */
 668                 MemSet(values, 0, sizeof(values));
 669                 MemSet(nulls, 0, sizeof(nulls));
 670
 671                 values[0] = TransactionIdGetDatum(pgxact->xid);
 672                 values[1] = CStringGetTextDatum(gxact->gid);
 673                 values[2] = TimestampTzGetDatum(gxact->prepared_at);
 674                 values[3] = ObjectIdGetDatum(gxact->owner);
 675                 values[4] = ObjectIdGetDatum(proc->databaseId);
 676
 677                 tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
 678                 result = HeapTupleGetDatum(tuple);
 679                 SRF_RETURN_NEXT(funcctx, result);
 680         }
 681
 682         SRF_RETURN_DONE(funcctx);
 683 }
 684
 685 /*
 686  * TwoPhaseGetGXact
 687  *              Get the GlobalTransaction struct for a prepared transaction
 688  *              specified by XID
 689  */
 690 static GlobalTransaction
 691 TwoPhaseGetGXact(TransactionId xid)
 692 {
 693         GlobalTransaction result = NULL;
 694         int                     i;
 695
 696         static TransactionId cached_xid = InvalidTransactionId;
 697         static GlobalTransaction cached_gxact = NULL;
 698
 699         /*
 700          * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
 701          * repeatedly for the same XID.  We can save work with a simple cache.
 702          */
 703         if (xid == cached_xid)
 704                 return cached_gxact;
 705
 706         LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 707
 708         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 709         {
 710                 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 711                 PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 712
 713                 if (pgxact->xid == xid)
 714                 {
 715                         result = gxact;
 716                         break;
 717                 }
 718         }
 719
 720         LWLockRelease(TwoPhaseStateLock);
 721
 722         if (result == NULL)                     /* should not happen */
 723                 elog(ERROR, "failed to find GlobalTransaction for xid %u", xid);
 724
 725         cached_xid = xid;
 726         cached_gxact = result;
 727
 728         return result;
 729 }
 730
 731 /*
 732  * TwoPhaseGetDummyProc
 733  *              Get the dummy backend ID for prepared transaction specified by XID
 734  *
 735  * Dummy backend IDs are similar to real backend IDs of real backends.
 736  * They start at MaxBackends + 1, and are unique across all currently active
 737  * real backends and prepared transactions.
 738  */
 739 BackendId
 740 TwoPhaseGetDummyBackendId(TransactionId xid)
 741 {
 742         GlobalTransaction gxact = TwoPhaseGetGXact(xid);
 743
 744         return gxact->dummyBackendId;
 745 }
 746
 747 /*
 748  * TwoPhaseGetDummyProc
 749  *              Get the PGPROC that represents a prepared transaction specified by XID
 750  */
 751 PGPROC *
 752 TwoPhaseGetDummyProc(TransactionId xid)
 753 {
 754         GlobalTransaction gxact = TwoPhaseGetGXact(xid);
 755
 756         return &ProcGlobal->allProcs[gxact->pgprocno];
 757 }
 758
 759 /************************************************************************/
 760 /* State file support                                                                                                   */
 761 /************************************************************************/
 762
 763 #define TwoPhaseFilePath(path, xid) \
 764         snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)
 765
 766 /*
 767  * 2PC state file format:
 768  *
 769  *      1. TwoPhaseFileHeader
 770  *      2. TransactionId[] (subtransactions)
 771  *      3. RelFileNode[] (files to be deleted at commit)
 772  *      4. RelFileNode[] (files to be deleted at abort)
 773  *      5. SharedInvalidationMessage[] (inval messages to be sent at commit)
 774  *      6. TwoPhaseRecordOnDisk
 775  *      7. ...
 776  *      8. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
 777  *      9. CRC32
 778  *
 779  * Each segment except the final CRC32 is MAXALIGN'd.
 780  */
 781
 782 /*
 783  * Header for a 2PC state file
 784  */
 785 #define TWOPHASE_MAGIC  0x57F94532              /* format identifier */
 786
 787 typedef struct TwoPhaseFileHeader
 788 {
 789         uint32          magic;                  /* format identifier */
 790         uint32          total_len;              /* actual file length */
 791         TransactionId xid;                      /* original transaction XID */
 792         Oid                     database;               /* OID of database it was in */
 793         TimestampTz prepared_at;        /* time of preparation */
 794         Oid                     owner;                  /* user running the transaction */
 795         int32           nsubxacts;              /* number of following subxact XIDs */
 796         int32           ncommitrels;    /* number of delete-on-commit rels */
 797         int32           nabortrels;             /* number of delete-on-abort rels */
 798         int32           ninvalmsgs;             /* number of cache invalidation messages */
 799         bool            initfileinval;  /* does relcache init file need invalidation? */
 800         char            gid[GIDSIZE];   /* GID for transaction */
 801 } TwoPhaseFileHeader;
 802
 803 /*
 804  * Header for each record in a state file
 805  *
 806  * NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
 807  * The rmgr data will be stored starting on a MAXALIGN boundary.
 808  */
 809 typedef struct TwoPhaseRecordOnDisk
 810 {
 811         uint32          len;                    /* length of rmgr data */
 812         TwoPhaseRmgrId rmid;            /* resource manager for this record */
 813         uint16          info;                   /* flag bits for use by rmgr */
 814 } TwoPhaseRecordOnDisk;
 815
 816 /*
 817  * During prepare, the state file is assembled in memory before writing it
 818  * to WAL and the actual state file.  We use a chain of XLogRecData blocks
 819  * so that we will be able to pass the state file contents directly to
 820  * XLogInsert.
 821  */
 822 static struct xllist
 823 {
 824         XLogRecData *head;                      /* first data block in the chain */
 825         XLogRecData *tail;                      /* last block in chain */
 826         uint32          bytes_free;             /* free bytes left in tail block */
 827         uint32          total_len;              /* total data bytes in chain */
 828 }       records;
 829
 830
 831 /*
 832  * Append a block of data to records data structure.
 833  *
 834  * NB: each block is padded to a MAXALIGN multiple.  This must be
 835  * accounted for when the file is later read!
 836  *
 837  * The data is copied, so the caller is free to modify it afterwards.
 838  */
 839 static void
 840 save_state_data(const void *data, uint32 len)
 841 {
 842         uint32          padlen = MAXALIGN(len);
 843
 844         if (padlen > records.bytes_free)
 845         {
 846                 records.tail->next = palloc0(sizeof(XLogRecData));
 847                 records.tail = records.tail->next;
 848                 records.tail->buffer = InvalidBuffer;
 849                 records.tail->len = 0;
 850                 records.tail->next = NULL;
 851
 852                 records.bytes_free = Max(padlen, 512);
 853                 records.tail->data = palloc(records.bytes_free);
 854         }
 855
 856         memcpy(((char *) records.tail->data) + records.tail->len, data, len);
 857         records.tail->len += padlen;
 858         records.bytes_free -= padlen;
 859         records.total_len += padlen;
 860 }
 861
 862 /*
 863  * Start preparing a state file.
 864  *
 865  * Initializes data structure and inserts the 2PC file header record.
 866  */
 867 void
 868 StartPrepare(GlobalTransaction gxact)
 869 {
 870         PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
 871         PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 872         TransactionId xid = pgxact->xid;
 873         TwoPhaseFileHeader hdr;
 874         TransactionId *children;
 875         RelFileNode *commitrels;
 876         RelFileNode *abortrels;
 877         SharedInvalidationMessage *invalmsgs;
 878
 879         /* Initialize linked list */
 880         records.head = palloc0(sizeof(XLogRecData));
 881         records.head->buffer = InvalidBuffer;
 882         records.head->len = 0;
 883         records.head->next = NULL;
 884
 885         records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
 886         records.head->data = palloc(records.bytes_free);
 887
 888         records.tail = records.head;
 889
 890         records.total_len = 0;
 891
 892         /* Create header */
 893         hdr.magic = TWOPHASE_MAGIC;
 894         hdr.total_len = 0;                      /* EndPrepare will fill this in */
 895         hdr.xid = xid;
 896         hdr.database = proc->databaseId;
 897         hdr.prepared_at = gxact->prepared_at;
 898         hdr.owner = gxact->owner;
 899         hdr.nsubxacts = xactGetCommittedChildren(&children);
 900         hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels);
 901         hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
 902         hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
 903                                                                                                                   &hdr.initfileinval);
 904         StrNCpy(hdr.gid, gxact->gid, GIDSIZE);
 905
 906         save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
 907
 908         /*
 909          * Add the additional info about subxacts, deletable files and cache
 910          * invalidation messages.
 911          */
 912         if (hdr.nsubxacts > 0)
 913         {
 914                 save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
 915                 /* While we have the child-xact data, stuff it in the gxact too */
 916                 GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
 917         }
 918         if (hdr.ncommitrels > 0)
 919         {
 920                 save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileNode));
 921                 pfree(commitrels);
 922         }
 923         if (hdr.nabortrels > 0)
 924         {
 925                 save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileNode));
 926                 pfree(abortrels);
 927         }
 928         if (hdr.ninvalmsgs > 0)
 929         {
 930                 save_state_data(invalmsgs,
 931                                                 hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
 932                 pfree(invalmsgs);
 933         }
 934 }
 935
 936 /*
 937  * Finish preparing state file.
 938  *
 939  * Calculates CRC and writes state file to WAL and in pg_twophase directory.
 940  */
 941 void
 942 EndPrepare(GlobalTransaction gxact)
 943 {
 944         PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
 945         TransactionId xid = pgxact->xid;
 946         TwoPhaseFileHeader *hdr;
 947         char            path[MAXPGPATH];
 948         XLogRecData *record;
 949         pg_crc32        statefile_crc;
 950         pg_crc32        bogus_crc;
 951         int                     fd;
 952
 953         /* Add the end sentinel to the list of 2PC records */
 954         RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
 955                                                    NULL, 0);
 956
 957         /* Go back and fill in total_len in the file header record */
 958         hdr = (TwoPhaseFileHeader *) records.head->data;
 959         Assert(hdr->magic == TWOPHASE_MAGIC);
 960         hdr->total_len = records.total_len + sizeof(pg_crc32);
 961
 962         /*
 963          * If the file size exceeds MaxAllocSize, we won't be able to read it in
 964          * ReadTwoPhaseFile. Check for that now, rather than fail at commit time.
 965          */
 966         if (hdr->total_len > MaxAllocSize)
 967                 ereport(ERROR,
 968                                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
 969                                  errmsg("two-phase state file maximum length exceeded")));
 970
 971         /*
 972          * Create the 2PC state file.
 973          *
 974          * Note: because we use BasicOpenFile(), we are responsible for ensuring
 975          * the FD gets closed in any error exit path.  Once we get into the
 976          * critical section, though, it doesn't matter since any failure causes
 977          * PANIC anyway.
 978          */
 979         TwoPhaseFilePath(path, xid);
 980
 981         fd = BasicOpenFile(path,
 982                                            O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
 983                                            S_IRUSR | S_IWUSR);
 984         if (fd < 0)
 985                 ereport(ERROR,
 986                                 (errcode_for_file_access(),
 987                                  errmsg("could not create two-phase state file \"%s\": %m",
 988                                                 path)));
 989
 990         /* Write data to file, and calculate CRC as we pass over it */
 991         INIT_CRC32(statefile_crc);
 992
 993         for (record = records.head; record != NULL; record = record->next)
 994         {
 995                 COMP_CRC32(statefile_crc, record->data, record->len);
 996                 if ((write(fd, record->data, record->len)) != record->len)
 997                 {
 998                         close(fd);
 999                         ereport(ERROR,
1000                                         (errcode_for_file_access(),
1001                                          errmsg("could not write two-phase state file: %m")));
1002                 }
1003         }
1004
1005         FIN_CRC32(statefile_crc);
1006
1007         /*
1008          * Write a deliberately bogus CRC to the state file; this is just paranoia
1009          * to catch the case where four more bytes will run us out of disk space.
1010          */
1011         bogus_crc = ~statefile_crc;
1012
1013         if ((write(fd, &bogus_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
1014         {
1015                 close(fd);
1016                 ereport(ERROR,
1017                                 (errcode_for_file_access(),
1018                                  errmsg("could not write two-phase state file: %m")));
1019         }
1020
1021         /* Back up to prepare for rewriting the CRC */
1022         if (lseek(fd, -((off_t) sizeof(pg_crc32)), SEEK_CUR) < 0)
1023         {
1024                 close(fd);
1025                 ereport(ERROR,
1026                                 (errcode_for_file_access(),
1027                                  errmsg("could not seek in two-phase state file: %m")));
1028         }
1029
1030         /*
1031          * The state file isn't valid yet, because we haven't written the correct
1032          * CRC yet.  Before we do that, insert entry in WAL and flush it to disk.
1033          *
1034          * Between the time we have written the WAL entry and the time we write
1035          * out the correct state file CRC, we have an inconsistency: the xact is
1036          * prepared according to WAL but not according to our on-disk state. We
1037          * use a critical section to force a PANIC if we are unable to complete
1038          * the write --- then, WAL replay should repair the inconsistency.      The
1039          * odds of a PANIC actually occurring should be very tiny given that we
1040          * were able to write the bogus CRC above.
1041          *
1042          * We have to set inCommit here, too; otherwise a checkpoint starting
1043          * immediately after the WAL record is inserted could complete without
1044          * fsync'ing our state file.  (This is essentially the same kind of race
1045          * condition as the COMMIT-to-clog-write case that RecordTransactionCommit
1046          * uses inCommit for; see notes there.)
1047          *
1048          * We save the PREPARE record's location in the gxact for later use by
1049          * CheckPointTwoPhase.
1050          */
1051         START_CRIT_SECTION();
1052
1053         MyPgXact->inCommit = true;
1054
1055         gxact->prepare_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE,
1056                                                                         records.head);
1057         XLogFlush(gxact->prepare_lsn);
1058
1059         /* If we crash now, we have prepared: WAL replay will fix things */
1060
1061         /* write correct CRC and close file */
1062         if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
1063         {
1064                 close(fd);
1065                 ereport(ERROR,
1066                                 (errcode_for_file_access(),
1067                                  errmsg("could not write two-phase state file: %m")));
1068         }
1069
1070         if (close(fd) != 0)
1071                 ereport(ERROR,
1072                                 (errcode_for_file_access(),
1073                                  errmsg("could not close two-phase state file: %m")));
1074
1075         /*
1076          * Mark the prepared transaction as valid.      As soon as xact.c marks
1077          * MyPgXact as not running our XID (which it will do immediately after
1078          * this function returns), others can commit/rollback the xact.
1079          *
1080          * NB: a side effect of this is to make a dummy ProcArray entry for the
1081          * prepared XID.  This must happen before we clear the XID from MyPgXact,
1082          * else there is a window where the XID is not running according to
1083          * TransactionIdIsInProgress, and onlookers would be entitled to assume
1084          * the xact crashed.  Instead we have a window where the same XID appears
1085          * twice in ProcArray, which is OK.
1086          */
1087         MarkAsPrepared(gxact);
1088
1089         /*
1090          * Now we can mark ourselves as out of the commit critical section: a
1091          * checkpoint starting after this will certainly see the gxact as a
1092          * candidate for fsyncing.
1093          */
1094         MyPgXact->inCommit = false;
1095
1096         END_CRIT_SECTION();
1097
1098         /*
1099          * Wait for synchronous replication, if required.
1100          *
1101          * Note that at this stage we have marked the prepare, but still show as
1102          * running in the procarray (twice!) and continue to hold locks.
1103          */
1104         SyncRepWaitForLSN(gxact->prepare_lsn);
1105
1106         records.tail = records.head = NULL;
1107 }
1108
1109 /*
1110  * Register a 2PC record to be written to state file.
1111  */
1112 void
1113 RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
1114                                            const void *data, uint32 len)
1115 {
1116         TwoPhaseRecordOnDisk record;
1117
1118         record.rmid = rmid;
1119         record.info = info;
1120         record.len = len;
1121         save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
1122         if (len > 0)
1123                 save_state_data(data, len);
1124 }
1125
1126
1127 /*
1128  * Read and validate the state file for xid.
1129  *
1130  * If it looks OK (has a valid magic number and CRC), return the palloc'd
1131  * contents of the file.  Otherwise return NULL.
1132  */
1133 static char *
1134 ReadTwoPhaseFile(TransactionId xid, bool give_warnings)
1135 {
1136         char            path[MAXPGPATH];
1137         char       *buf;
1138         TwoPhaseFileHeader *hdr;
1139         int                     fd;
1140         struct stat stat;
1141         uint32          crc_offset;
1142         pg_crc32        calc_crc,
1143                                 file_crc;
1144
1145         TwoPhaseFilePath(path, xid);
1146
1147         fd = BasicOpenFile(path, O_RDONLY | PG_BINARY, 0);
1148         if (fd < 0)
1149         {
1150                 if (give_warnings)
1151                         ereport(WARNING,
1152                                         (errcode_for_file_access(),
1153                                          errmsg("could not open two-phase state file \"%s\": %m",
1154                                                         path)));
1155                 return NULL;
1156         }
1157
1158         /*
1159          * Check file length.  We can determine a lower bound pretty easily. We
1160          * set an upper bound to avoid palloc() failure on a corrupt file, though
1161          * we can't guarantee that we won't get an out of memory error anyway,
1162          * even on a valid file.
1163          */
1164         if (fstat(fd, &stat))
1165         {
1166                 close(fd);
1167                 if (give_warnings)
1168                         ereport(WARNING,
1169                                         (errcode_for_file_access(),
1170                                          errmsg("could not stat two-phase state file \"%s\": %m",
1171                                                         path)));
1172                 return NULL;
1173         }
1174
1175         if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
1176                                                 MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
1177                                                 sizeof(pg_crc32)) ||
1178                 stat.st_size > MaxAllocSize)
1179         {
1180                 close(fd);
1181                 return NULL;
1182         }
1183
1184         crc_offset = stat.st_size - sizeof(pg_crc32);
1185         if (crc_offset != MAXALIGN(crc_offset))
1186         {
1187                 close(fd);
1188                 return NULL;
1189         }
1190
1191         /*
1192          * OK, slurp in the file.
1193          */
1194         buf = (char *) palloc(stat.st_size);
1195
1196         if (read(fd, buf, stat.st_size) != stat.st_size)
1197         {
1198                 close(fd);
1199                 if (give_warnings)
1200                         ereport(WARNING,
1201                                         (errcode_for_file_access(),
1202                                          errmsg("could not read two-phase state file \"%s\": %m",
1203                                                         path)));
1204                 pfree(buf);
1205                 return NULL;
1206         }
1207
1208         close(fd);
1209
1210         hdr = (TwoPhaseFileHeader *) buf;
1211         if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
1212         {
1213                 pfree(buf);
1214                 return NULL;
1215         }
1216
1217         INIT_CRC32(calc_crc);
1218         COMP_CRC32(calc_crc, buf, crc_offset);
1219         FIN_CRC32(calc_crc);
1220
1221         file_crc = *((pg_crc32 *) (buf + crc_offset));
1222
1223         if (!EQ_CRC32(calc_crc, file_crc))
1224         {
1225                 pfree(buf);
1226                 return NULL;
1227         }
1228
1229         return buf;
1230 }
1231
1232 /*
1233  * Confirms an xid is prepared, during recovery
1234  */
1235 bool
1236 StandbyTransactionIdIsPrepared(TransactionId xid)
1237 {
1238         char       *buf;
1239         TwoPhaseFileHeader *hdr;
1240         bool            result;
1241
1242         Assert(TransactionIdIsValid(xid));
1243
1244         if (max_prepared_xacts <= 0)
1245                 return false;                   /* nothing to do */
1246
1247         /* Read and validate file */
1248         buf = ReadTwoPhaseFile(xid, false);
1249         if (buf == NULL)
1250                 return false;
1251
1252         /* Check header also */
1253         hdr = (TwoPhaseFileHeader *) buf;
1254         result = TransactionIdEquals(hdr->xid, xid);
1255         pfree(buf);
1256
1257         return result;
1258 }
1259
1260 /*
1261  * FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
1262  */
1263 void
1264 FinishPreparedTransaction(const char *gid, bool isCommit)
1265 {
1266         GlobalTransaction gxact;
1267         PGPROC     *proc;
1268         PGXACT     *pgxact;
1269         TransactionId xid;
1270         char       *buf;
1271         char       *bufptr;
1272         TwoPhaseFileHeader *hdr;
1273         TransactionId latestXid;
1274         TransactionId *children;
1275         RelFileNode *commitrels;
1276         RelFileNode *abortrels;
1277         RelFileNode *delrels;
1278         int                     ndelrels;
1279         SharedInvalidationMessage *invalmsgs;
1280         int                     i;
1281
1282         /*
1283          * Validate the GID, and lock the GXACT to ensure that two backends do not
1284          * try to commit the same GID at once.
1285          */
1286         gxact = LockGXact(gid, GetUserId());
1287         proc = &ProcGlobal->allProcs[gxact->pgprocno];
1288         pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
1289         xid = pgxact->xid;
1290
1291         /*
1292          * Read and validate the state file
1293          */
1294         buf = ReadTwoPhaseFile(xid, true);
1295         if (buf == NULL)
1296                 ereport(ERROR,
1297                                 (errcode(ERRCODE_DATA_CORRUPTED),
1298                                  errmsg("two-phase state file for transaction %u is corrupt",
1299                                                 xid)));
1300
1301         /*
1302          * Disassemble the header area
1303          */
1304         hdr = (TwoPhaseFileHeader *) buf;
1305         Assert(TransactionIdEquals(hdr->xid, xid));
1306         bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
1307         children = (TransactionId *) bufptr;
1308         bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
1309         commitrels = (RelFileNode *) bufptr;
1310         bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
1311         abortrels = (RelFileNode *) bufptr;
1312         bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
1313         invalmsgs = (SharedInvalidationMessage *) bufptr;
1314         bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
1315
1316         /* compute latestXid among all children */
1317         latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
1318
1319         /*
1320          * The order of operations here is critical: make the XLOG entry for
1321          * commit or abort, then mark the transaction committed or aborted in
1322          * pg_clog, then remove its PGPROC from the global ProcArray (which means
1323          * TransactionIdIsInProgress will stop saying the prepared xact is in
1324          * progress), then run the post-commit or post-abort callbacks. The
1325          * callbacks will release the locks the transaction held.
1326          */
1327         if (isCommit)
1328                 RecordTransactionCommitPrepared(xid,
1329                                                                                 hdr->nsubxacts, children,
1330                                                                                 hdr->ncommitrels, commitrels,
1331                                                                                 hdr->ninvalmsgs, invalmsgs,
1332                                                                                 hdr->initfileinval);
1333         else
1334                 RecordTransactionAbortPrepared(xid,
1335                                                                            hdr->nsubxacts, children,
1336                                                                            hdr->nabortrels, abortrels);
1337
1338         ProcArrayRemove(proc, latestXid);
1339
1340         /*
1341          * In case we fail while running the callbacks, mark the gxact invalid so
1342          * no one else will try to commit/rollback, and so it can be recycled
1343          * properly later.      It is still locked by our XID so it won't go away yet.
1344          *
1345          * (We assume it's safe to do this without taking TwoPhaseStateLock.)
1346          */
1347         gxact->valid = false;
1348
1349         /*
1350          * We have to remove any files that were supposed to be dropped. For
1351          * consistency with the regular xact.c code paths, must do this before
1352          * releasing locks, so do it before running the callbacks.
1353          *
1354          * NB: this code knows that we couldn't be dropping any temp rels ...
1355          */
1356         if (isCommit)
1357         {
1358                 delrels = commitrels;
1359                 ndelrels = hdr->ncommitrels;
1360         }
1361         else
1362         {
1363                 delrels = abortrels;
1364                 ndelrels = hdr->nabortrels;
1365         }
1366         for (i = 0; i < ndelrels; i++)
1367         {
1368                 SMgrRelation srel = smgropen(delrels[i], InvalidBackendId);
1369
1370                 smgrdounlink(srel, false);
1371                 smgrclose(srel);
1372         }
1373
1374         /*
1375          * Handle cache invalidation messages.
1376          *
1377          * Relcache init file invalidation requires processing both before and
1378          * after we send the SI messages. See AtEOXact_Inval()
1379          */
1380         if (hdr->initfileinval)
1381                 RelationCacheInitFilePreInvalidate();
1382         SendSharedInvalidMessages(invalmsgs, hdr->ninvalmsgs);
1383         if (hdr->initfileinval)
1384                 RelationCacheInitFilePostInvalidate();
1385
1386         /* And now do the callbacks */
1387         if (isCommit)
1388                 ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
1389         else
1390                 ProcessRecords(bufptr, xid, twophase_postabort_callbacks);
1391
1392         PredicateLockTwoPhaseFinish(xid, isCommit);
1393
1394         /* Count the prepared xact as committed or aborted */
1395         AtEOXact_PgStat(isCommit);
1396
1397         /*
1398          * And now we can clean up our mess.
1399          */
1400         RemoveTwoPhaseFile(xid, true);
1401
1402         RemoveGXact(gxact);
1403
1404         pfree(buf);
1405 }
1406
1407 /*
1408  * Scan a 2PC state file (already read into memory by ReadTwoPhaseFile)
1409  * and call the indicated callbacks for each 2PC record.
1410  */
1411 static void
1412 ProcessRecords(char *bufptr, TransactionId xid,
1413                            const TwoPhaseCallback callbacks[])
1414 {
1415         for (;;)
1416         {
1417                 TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;
1418
1419                 Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
1420                 if (record->rmid == TWOPHASE_RM_END_ID)
1421                         break;
1422
1423                 bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));
1424
1425                 if (callbacks[record->rmid] != NULL)
1426                         callbacks[record->rmid] (xid, record->info,
1427                                                                          (void *) bufptr, record->len);
1428
1429                 bufptr += MAXALIGN(record->len);
1430         }
1431 }
1432
1433 /*
1434  * Remove the 2PC file for the specified XID.
1435  *
1436  * If giveWarning is false, do not complain about file-not-present;
1437  * this is an expected case during WAL replay.
1438  */
1439 void
1440 RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
1441 {
1442         char            path[MAXPGPATH];
1443
1444         TwoPhaseFilePath(path, xid);
1445         if (unlink(path))
1446                 if (errno != ENOENT || giveWarning)
1447                         ereport(WARNING,
1448                                         (errcode_for_file_access(),
1449                                    errmsg("could not remove two-phase state file \"%s\": %m",
1450                                                   path)));
1451 }
1452
1453 /*
1454  * Recreates a state file. This is used in WAL replay.
1455  *
1456  * Note: content and len don't include CRC.
1457  */
1458 void
1459 RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
1460 {
1461         char            path[MAXPGPATH];
1462         pg_crc32        statefile_crc;
1463         int                     fd;
1464
1465         /* Recompute CRC */
1466         INIT_CRC32(statefile_crc);
1467         COMP_CRC32(statefile_crc, content, len);
1468         FIN_CRC32(statefile_crc);
1469
1470         TwoPhaseFilePath(path, xid);
1471
1472         fd = BasicOpenFile(path,
1473                                            O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY,
1474                                            S_IRUSR | S_IWUSR);
1475         if (fd < 0)
1476                 ereport(ERROR,
1477                                 (errcode_for_file_access(),
1478                                  errmsg("could not recreate two-phase state file \"%s\": %m",
1479                                                 path)));
1480
1481         /* Write content and CRC */
1482         if (write(fd, content, len) != len)
1483         {
1484                 close(fd);
1485                 ereport(ERROR,
1486                                 (errcode_for_file_access(),
1487                                  errmsg("could not write two-phase state file: %m")));
1488         }
1489         if (write(fd, &statefile_crc, sizeof(pg_crc32)) != sizeof(pg_crc32))
1490         {
1491                 close(fd);
1492                 ereport(ERROR,
1493                                 (errcode_for_file_access(),
1494                                  errmsg("could not write two-phase state file: %m")));
1495         }
1496
1497         /*
1498          * We must fsync the file because the end-of-replay checkpoint will not do
1499          * so, there being no GXACT in shared memory yet to tell it to.
1500          */
1501         if (pg_fsync(fd) != 0)
1502         {
1503                 close(fd);
1504                 ereport(ERROR,
1505                                 (errcode_for_file_access(),
1506                                  errmsg("could not fsync two-phase state file: %m")));
1507         }
1508
1509         if (close(fd) != 0)
1510                 ereport(ERROR,
1511                                 (errcode_for_file_access(),
1512                                  errmsg("could not close two-phase state file: %m")));
1513 }
1514
1515 /*
1516  * CheckPointTwoPhase -- handle 2PC component of checkpointing.
1517  *
1518  * We must fsync the state file of any GXACT that is valid and has a PREPARE
1519  * LSN <= the checkpoint's redo horizon.  (If the gxact isn't valid yet or
1520  * has a later LSN, this checkpoint is not responsible for fsyncing it.)
1521  *
1522  * This is deliberately run as late as possible in the checkpoint sequence,
1523  * because GXACTs ordinarily have short lifespans, and so it is quite
1524  * possible that GXACTs that were valid at checkpoint start will no longer
1525  * exist if we wait a little bit.
1526  *
1527  * If a GXACT remains valid across multiple checkpoints, it'll be fsynced
1528  * each time.  This is considered unusual enough that we don't bother to
1529  * expend any extra code to avoid the redundant fsyncs.  (They should be
1530  * reasonably cheap anyway, since they won't cause I/O.)
1531  */
1532 void
1533 CheckPointTwoPhase(XLogRecPtr redo_horizon)
1534 {
1535         TransactionId *xids;
1536         int                     nxids;
1537         char            path[MAXPGPATH];
1538         int                     i;
1539
1540         /*
1541          * We don't want to hold the TwoPhaseStateLock while doing I/O, so we grab
1542          * it just long enough to make a list of the XIDs that require fsyncing,
1543          * and then do the I/O afterwards.
1544          *
1545          * This approach creates a race condition: someone else could delete a
1546          * GXACT between the time we release TwoPhaseStateLock and the time we try
1547          * to open its state file.      We handle this by special-casing ENOENT
1548          * failures: if we see that, we verify that the GXACT is no longer valid,
1549          * and if so ignore the failure.
1550          */
1551         if (max_prepared_xacts <= 0)
1552                 return;                                 /* nothing to do */
1553
1554         TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();
1555
1556         xids = (TransactionId *) palloc(max_prepared_xacts * sizeof(TransactionId));
1557         nxids = 0;
1558
1559         LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
1560
1561         for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
1562         {
1563                 GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
1564                 PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
1565
1566                 if (gxact->valid &&
1567                         XLByteLE(gxact->prepare_lsn, redo_horizon))
1568                         xids[nxids++] = pgxact->xid;
1569         }
1570
1571         LWLockRelease(TwoPhaseStateLock);
1572
1573         for (i = 0; i < nxids; i++)
1574         {
1575                 TransactionId xid = xids[i];
1576                 int                     fd;
1577
1578                 TwoPhaseFilePath(path, xid);
1579
1580                 fd = BasicOpenFile(path, O_RDWR | PG_BINARY, 0);
1581                 if (fd < 0)
1582                 {
1583                         if (errno == ENOENT)
1584                         {
1585                                 /* OK if gxact is no longer valid */
1586                                 if (!TransactionIdIsPrepared(xid))
1587                                         continue;
1588                                 /* Restore errno in case it was changed */
1589                                 errno = ENOENT;
1590                         }
1591                         ereport(ERROR,
1592                                         (errcode_for_file_access(),
1593                                          errmsg("could not open two-phase state file \"%s\": %m",
1594                                                         path)));
1595                 }
1596
1597                 if (pg_fsync(fd) != 0)
1598                 {
1599                         close(fd);
1600                         ereport(ERROR,
1601                                         (errcode_for_file_access(),
1602                                          errmsg("could not fsync two-phase state file \"%s\": %m",
1603                                                         path)));
1604                 }
1605
1606                 if (close(fd) != 0)
1607                         ereport(ERROR,
1608                                         (errcode_for_file_access(),
1609                                          errmsg("could not close two-phase state file \"%s\": %m",
1610                                                         path)));
1611         }
1612
1613         pfree(xids);
1614
1615         TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
1616 }
1617
1618 /*
1619  * PrescanPreparedTransactions
1620  *
1621  * Scan the pg_twophase directory and determine the range of valid XIDs
1622  * present.  This is run during database startup, after we have completed
1623  * reading WAL.  ShmemVariableCache->nextXid has been set to one more than
1624  * the highest XID for which evidence exists in WAL.
1625  *
1626  * We throw away any prepared xacts with main XID beyond nextXid --- if any
1627  * are present, it suggests that the DBA has done a PITR recovery to an
1628  * earlier point in time without cleaning out pg_twophase.      We dare not
1629  * try to recover such prepared xacts since they likely depend on database
1630  * state that doesn't exist now.
1631  *
1632  * However, we will advance nextXid beyond any subxact XIDs belonging to
1633  * valid prepared xacts.  We need to do this since subxact commit doesn't
1634  * write a WAL entry, and so there might be no evidence in WAL of those
1635  * subxact XIDs.
1636  *
1637  * Our other responsibility is to determine and return the oldest valid XID
1638  * among the prepared xacts (if none, return ShmemVariableCache->nextXid).
1639  * This is needed to synchronize pg_subtrans startup properly.
1640  *
1641  * If xids_p and nxids_p are not NULL, pointer to a palloc'd array of all
1642  * top-level xids is stored in *xids_p. The number of entries in the array
1643  * is returned in *nxids_p.
1644  */
1645 TransactionId
1646 PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
1647 {
1648         TransactionId origNextXid = ShmemVariableCache->nextXid;
1649         TransactionId result = origNextXid;
1650         DIR                *cldir;
1651         struct dirent *clde;
1652         TransactionId *xids = NULL;
1653         int                     nxids = 0;
1654         int                     allocsize = 0;
1655
1656         cldir = AllocateDir(TWOPHASE_DIR);
1657         while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
1658         {
1659                 if (strlen(clde->d_name) == 8 &&
1660                         strspn(clde->d_name, "0123456789ABCDEF") == 8)
1661                 {
1662                         TransactionId xid;
1663                         char       *buf;
1664                         TwoPhaseFileHeader *hdr;
1665                         TransactionId *subxids;
1666                         int                     i;
1667
1668                         xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
1669
1670                         /* Reject XID if too new */
1671                         if (TransactionIdFollowsOrEquals(xid, origNextXid))
1672                         {
1673                                 ereport(WARNING,
1674                                                 (errmsg("removing future two-phase state file \"%s\"",
1675                                                                 clde->d_name)));
1676                                 RemoveTwoPhaseFile(xid, true);
1677                                 continue;
1678                         }
1679
1680                         /*
1681                          * Note: we can't check if already processed because clog
1682                          * subsystem isn't up yet.
1683                          */
1684
1685                         /* Read and validate file */
1686                         buf = ReadTwoPhaseFile(xid, true);
1687                         if (buf == NULL)
1688                         {
1689                                 ereport(WARNING,
1690                                           (errmsg("removing corrupt two-phase state file \"%s\"",
1691                                                           clde->d_name)));
1692                                 RemoveTwoPhaseFile(xid, true);
1693                                 continue;
1694                         }
1695
1696                         /* Deconstruct header */
1697                         hdr = (TwoPhaseFileHeader *) buf;
1698                         if (!TransactionIdEquals(hdr->xid, xid))
1699                         {
1700                                 ereport(WARNING,
1701                                           (errmsg("removing corrupt two-phase state file \"%s\"",
1702                                                           clde->d_name)));
1703                                 RemoveTwoPhaseFile(xid, true);
1704                                 pfree(buf);
1705                                 continue;
1706                         }
1707
1708                         /*
1709                          * OK, we think this file is valid.  Incorporate xid into the
1710                          * running-minimum result.
1711                          */
1712                         if (TransactionIdPrecedes(xid, result))
1713                                 result = xid;
1714
1715                         /*
1716                          * Examine subtransaction XIDs ... they should all follow main
1717                          * XID, and they may force us to advance nextXid.
1718                          *
1719                          * We don't expect anyone else to modify nextXid, hence we don't
1720                          * need to hold a lock while examining it.      We still acquire the
1721                          * lock to modify it, though.
1722                          */
1723                         subxids = (TransactionId *)
1724                                 (buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
1725                         for (i = 0; i < hdr->nsubxacts; i++)
1726                         {
1727                                 TransactionId subxid = subxids[i];
1728
1729                                 Assert(TransactionIdFollows(subxid, xid));
1730                                 if (TransactionIdFollowsOrEquals(subxid,
1731                                                                                                  ShmemVariableCache->nextXid))
1732                                 {
1733                                         LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
1734                                         ShmemVariableCache->nextXid = subxid;
1735                                         TransactionIdAdvance(ShmemVariableCache->nextXid);
1736                                         LWLockRelease(XidGenLock);
1737                                 }
1738                         }
1739
1740
1741                         if (xids_p)
1742                         {
1743                                 if (nxids == allocsize)
1744                                 {
1745                                         if (nxids == 0)
1746                                         {
1747                                                 allocsize = 10;
1748                                                 xids = palloc(allocsize * sizeof(TransactionId));
1749                                         }
1750                                         else
1751                                         {
1752                                                 allocsize = allocsize * 2;
1753                                                 xids = repalloc(xids, allocsize * sizeof(TransactionId));
1754                                         }
1755                                 }
1756                                 xids[nxids++] = xid;
1757                         }
1758
1759                         pfree(buf);
1760                 }
1761         }
1762         FreeDir(cldir);
1763
1764         if (xids_p)
1765         {
1766                 *xids_p = xids;
1767                 *nxids_p = nxids;
1768         }
1769
1770         return result;
1771 }
1772
1773 /*
1774  * StandbyRecoverPreparedTransactions
1775  *
1776  * Scan the pg_twophase directory and setup all the required information to
1777  * allow standby queries to treat prepared transactions as still active.
1778  * This is never called at the end of recovery - we use
1779  * RecoverPreparedTransactions() at that point.
1780  *
1781  * Currently we simply call SubTransSetParent() for any subxids of prepared
1782  * transactions. If overwriteOK is true, it's OK if some XIDs have already
1783  * been marked in pg_subtrans.
1784  */
1785 void
1786 StandbyRecoverPreparedTransactions(bool overwriteOK)
1787 {
1788         DIR                *cldir;
1789         struct dirent *clde;
1790
1791         cldir = AllocateDir(TWOPHASE_DIR);
1792         while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
1793         {
1794                 if (strlen(clde->d_name) == 8 &&
1795                         strspn(clde->d_name, "0123456789ABCDEF") == 8)
1796                 {
1797                         TransactionId xid;
1798                         char       *buf;
1799                         TwoPhaseFileHeader *hdr;
1800                         TransactionId *subxids;
1801                         int                     i;
1802
1803                         xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
1804
1805                         /* Already processed? */
1806                         if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
1807                         {
1808                                 ereport(WARNING,
1809                                                 (errmsg("removing stale two-phase state file \"%s\"",
1810                                                                 clde->d_name)));
1811                                 RemoveTwoPhaseFile(xid, true);
1812                                 continue;
1813                         }
1814
1815                         /* Read and validate file */
1816                         buf = ReadTwoPhaseFile(xid, true);
1817                         if (buf == NULL)
1818                         {
1819                                 ereport(WARNING,
1820                                           (errmsg("removing corrupt two-phase state file \"%s\"",
1821                                                           clde->d_name)));
1822                                 RemoveTwoPhaseFile(xid, true);
1823                                 continue;
1824                         }
1825
1826                         /* Deconstruct header */
1827                         hdr = (TwoPhaseFileHeader *) buf;
1828                         if (!TransactionIdEquals(hdr->xid, xid))
1829                         {
1830                                 ereport(WARNING,
1831                                           (errmsg("removing corrupt two-phase state file \"%s\"",
1832                                                           clde->d_name)));
1833                                 RemoveTwoPhaseFile(xid, true);
1834                                 pfree(buf);
1835                                 continue;
1836                         }
1837
1838                         /*
1839                          * Examine subtransaction XIDs ... they should all follow main
1840                          * XID.
1841                          */
1842                         subxids = (TransactionId *)
1843                                 (buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
1844                         for (i = 0; i < hdr->nsubxacts; i++)
1845                         {
1846                                 TransactionId subxid = subxids[i];
1847
1848                                 Assert(TransactionIdFollows(subxid, xid));
1849                                 SubTransSetParent(xid, subxid, overwriteOK);
1850                         }
1851                 }
1852         }
1853         FreeDir(cldir);
1854 }
1855
1856 /*
1857  * RecoverPreparedTransactions
1858  *
1859  * Scan the pg_twophase directory and reload shared-memory state for each
1860  * prepared transaction (reacquire locks, etc).  This is run during database
1861  * startup.
1862  */
1863 void
1864 RecoverPreparedTransactions(void)
1865 {
1866         char            dir[MAXPGPATH];
1867         DIR                *cldir;
1868         struct dirent *clde;
1869         bool            overwriteOK = false;
1870
1871         snprintf(dir, MAXPGPATH, "%s", TWOPHASE_DIR);
1872
1873         cldir = AllocateDir(dir);
1874         while ((clde = ReadDir(cldir, dir)) != NULL)
1875         {
1876                 if (strlen(clde->d_name) == 8 &&
1877                         strspn(clde->d_name, "0123456789ABCDEF") == 8)
1878                 {
1879                         TransactionId xid;
1880                         char       *buf;
1881                         char       *bufptr;
1882                         TwoPhaseFileHeader *hdr;
1883                         TransactionId *subxids;
1884                         GlobalTransaction gxact;
1885                         int                     i;
1886
1887                         xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
1888
1889                         /* Already processed? */
1890                         if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
1891                         {
1892                                 ereport(WARNING,
1893                                                 (errmsg("removing stale two-phase state file \"%s\"",
1894                                                                 clde->d_name)));
1895                                 RemoveTwoPhaseFile(xid, true);
1896                                 continue;
1897                         }
1898
1899                         /* Read and validate file */
1900                         buf = ReadTwoPhaseFile(xid, true);
1901                         if (buf == NULL)
1902                         {
1903                                 ereport(WARNING,
1904                                           (errmsg("removing corrupt two-phase state file \"%s\"",
1905                                                           clde->d_name)));
1906                                 RemoveTwoPhaseFile(xid, true);
1907                                 continue;
1908                         }
1909
1910                         ereport(LOG,
1911                                         (errmsg("recovering prepared transaction %u", xid)));
1912
1913                         /* Deconstruct header */
1914                         hdr = (TwoPhaseFileHeader *) buf;
1915                         Assert(TransactionIdEquals(hdr->xid, xid));
1916                         bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
1917                         subxids = (TransactionId *) bufptr;
1918                         bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
1919                         bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
1920                         bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
1921                         bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
1922
1923                         /*
1924                          * It's possible that SubTransSetParent has been set before, if
1925                          * the prepared transaction generated xid assignment records. Test
1926                          * here must match one used in AssignTransactionId().
1927                          */
1928                         if (InHotStandby && hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS)
1929                                 overwriteOK = true;
1930
1931                         /*
1932                          * Reconstruct subtrans state for the transaction --- needed
1933                          * because pg_subtrans is not preserved over a restart.  Note that
1934                          * we are linking all the subtransactions directly to the
1935                          * top-level XID; there may originally have been a more complex
1936                          * hierarchy, but there's no need to restore that exactly.
1937                          */
1938                         for (i = 0; i < hdr->nsubxacts; i++)
1939                                 SubTransSetParent(subxids[i], xid, overwriteOK);
1940
1941                         /*
1942                          * Recreate its GXACT and dummy PGPROC
1943                          *
1944                          * Note: since we don't have the PREPARE record's WAL location at
1945                          * hand, we leave prepare_lsn zeroes.  This means the GXACT will
1946                          * be fsync'd on every future checkpoint.  We assume this
1947                          * situation is infrequent enough that the performance cost is
1948                          * negligible (especially since we know the state file has already
1949                          * been fsynced).
1950                          */
1951                         gxact = MarkAsPreparing(xid, hdr->gid,
1952                                                                         hdr->prepared_at,
1953                                                                         hdr->owner, hdr->database);
1954                         GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
1955                         MarkAsPrepared(gxact);
1956
1957                         /*
1958                          * Recover other state (notably locks) using resource managers
1959                          */
1960                         ProcessRecords(bufptr, xid, twophase_recover_callbacks);
1961
1962                         /*
1963                          * Release locks held by the standby process after we process each
1964                          * prepared transaction. As a result, we don't need too many
1965                          * additional locks at any one time.
1966                          */
1967                         if (InHotStandby)
1968                                 StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids);
1969
1970                         pfree(buf);
1971                 }
1972         }
1973         FreeDir(cldir);
1974 }
1975
1976 /*
1977  *      RecordTransactionCommitPrepared
1978  *
1979  * This is basically the same as RecordTransactionCommit: in particular,
1980  * we must set the inCommit flag to avoid a race condition.
1981  *
1982  * We know the transaction made at least one XLOG entry (its PREPARE),
1983  * so it is never possible to optimize out the commit record.
1984  */
1985 static void
1986 RecordTransactionCommitPrepared(TransactionId xid,
1987                                                                 int nchildren,
1988                                                                 TransactionId *children,
1989                                                                 int nrels,
1990                                                                 RelFileNode *rels,
1991                                                                 int ninvalmsgs,
1992                                                                 SharedInvalidationMessage *invalmsgs,
1993                                                                 bool initfileinval)
1994 {
1995         XLogRecData rdata[4];
1996         int                     lastrdata = 0;
1997         xl_xact_commit_prepared xlrec;
1998         XLogRecPtr      recptr;
1999
2000         START_CRIT_SECTION();
2001
2002         /* See notes in RecordTransactionCommit */
2003         MyPgXact->inCommit = true;
2004
2005         /* Emit the XLOG commit record */
2006         xlrec.xid = xid;
2007         xlrec.crec.xact_time = GetCurrentTimestamp();
2008         xlrec.crec.xinfo = initfileinval ? XACT_COMPLETION_UPDATE_RELCACHE_FILE : 0;
2009         xlrec.crec.nmsgs = 0;
2010         xlrec.crec.nrels = nrels;
2011         xlrec.crec.nsubxacts = nchildren;
2012         xlrec.crec.nmsgs = ninvalmsgs;
2013
2014         rdata[0].data = (char *) (&xlrec);
2015         rdata[0].len = MinSizeOfXactCommitPrepared;
2016         rdata[0].buffer = InvalidBuffer;
2017         /* dump rels to delete */
2018         if (nrels > 0)
2019         {
2020                 rdata[0].next = &(rdata[1]);
2021                 rdata[1].data = (char *) rels;
2022                 rdata[1].len = nrels * sizeof(RelFileNode);
2023                 rdata[1].buffer = InvalidBuffer;
2024                 lastrdata = 1;
2025         }
2026         /* dump committed child Xids */
2027         if (nchildren > 0)
2028         {
2029                 rdata[lastrdata].next = &(rdata[2]);
2030                 rdata[2].data = (char *) children;
2031                 rdata[2].len = nchildren * sizeof(TransactionId);
2032                 rdata[2].buffer = InvalidBuffer;
2033                 lastrdata = 2;
2034         }
2035         /* dump cache invalidation messages */
2036         if (ninvalmsgs > 0)
2037         {
2038                 rdata[lastrdata].next = &(rdata[3]);
2039                 rdata[3].data = (char *) invalmsgs;
2040                 rdata[3].len = ninvalmsgs * sizeof(SharedInvalidationMessage);
2041                 rdata[3].buffer = InvalidBuffer;
2042                 lastrdata = 3;
2043         }
2044         rdata[lastrdata].next = NULL;
2045
2046         recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT_PREPARED, rdata);
2047
2048         /*
2049          * We don't currently try to sleep before flush here ... nor is there any
2050          * support for async commit of a prepared xact (the very idea is probably
2051          * a contradiction)
2052          */
2053
2054         /* Flush XLOG to disk */
2055         XLogFlush(recptr);
2056
2057         /* Mark the transaction committed in pg_clog */
2058         TransactionIdCommitTree(xid, nchildren, children);
2059
2060         /* Checkpoint can proceed now */
2061         MyPgXact->inCommit = false;
2062
2063         END_CRIT_SECTION();
2064
2065         /*
2066          * Wait for synchronous replication, if required.
2067          *
2068          * Note that at this stage we have marked clog, but still show as running
2069          * in the procarray and continue to hold locks.
2070          */
2071         SyncRepWaitForLSN(recptr);
2072 }
2073
2074 /*
2075  *      RecordTransactionAbortPrepared
2076  *
2077  * This is basically the same as RecordTransactionAbort.
2078  *
2079  * We know the transaction made at least one XLOG entry (its PREPARE),
2080  * so it is never possible to optimize out the abort record.
2081  */
2082 static void
2083 RecordTransactionAbortPrepared(TransactionId xid,
2084                                                            int nchildren,
2085                                                            TransactionId *children,
2086                                                            int nrels,
2087                                                            RelFileNode *rels)
2088 {
2089         XLogRecData rdata[3];
2090         int                     lastrdata = 0;
2091         xl_xact_abort_prepared xlrec;
2092         XLogRecPtr      recptr;
2093
2094         /*
2095          * Catch the scenario where we aborted partway through
2096          * RecordTransactionCommitPrepared ...
2097          */
2098         if (TransactionIdDidCommit(xid))
2099                 elog(PANIC, "cannot abort transaction %u, it was already committed",
2100                          xid);
2101
2102         START_CRIT_SECTION();
2103
2104         /* Emit the XLOG abort record */
2105         xlrec.xid = xid;
2106         xlrec.arec.xact_time = GetCurrentTimestamp();
2107         xlrec.arec.nrels = nrels;
2108         xlrec.arec.nsubxacts = nchildren;
2109         rdata[0].data = (char *) (&xlrec);
2110         rdata[0].len = MinSizeOfXactAbortPrepared;
2111         rdata[0].buffer = InvalidBuffer;
2112         /* dump rels to delete */
2113         if (nrels > 0)
2114         {
2115                 rdata[0].next = &(rdata[1]);
2116                 rdata[1].data = (char *) rels;
2117                 rdata[1].len = nrels * sizeof(RelFileNode);
2118                 rdata[1].buffer = InvalidBuffer;
2119                 lastrdata = 1;
2120         }
2121         /* dump committed child Xids */
2122         if (nchildren > 0)
2123         {
2124                 rdata[lastrdata].next = &(rdata[2]);
2125                 rdata[2].data = (char *) children;
2126                 rdata[2].len = nchildren * sizeof(TransactionId);
2127                 rdata[2].buffer = InvalidBuffer;
2128                 lastrdata = 2;
2129         }
2130         rdata[lastrdata].next = NULL;
2131
2132         recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ABORT_PREPARED, rdata);
2133
2134         /* Always flush, since we're about to remove the 2PC state file */
2135         XLogFlush(recptr);
2136
2137         /*
2138          * Mark the transaction aborted in clog.  This is not absolutely necessary
2139          * but we may as well do it while we are here.
2140          */
2141         TransactionIdAbortTree(xid, nchildren, children);
2142
2143         END_CRIT_SECTION();
2144
2145         /*
2146          * Wait for synchronous replication, if required.
2147          *
2148          * Note that at this stage we have marked clog, but still show as running
2149          * in the procarray and continue to hold locks.
2150          */
2151         SyncRepWaitForLSN(recptr);
2152 }