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