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

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