LCOV - code coverage report
Current view: top level - src/backend/access/transam - twophase.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13devel Lines: 600 685 87.6 %
Date: 2019-11-13 21:06:57 Functions: 36 37 97.3 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.13