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