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