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