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