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

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