#include "postgres.h"
#include <fcntl.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include "access/commit_ts.h"
#include "access/htup_details.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
#include "access/twophase_rmgr.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "access/xlogreader.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "catalog/storage.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
#include "replication/origin.h"
#include "replication/syncrep.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/md.h"
#include "storage/predicate.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
#include "utils/injection_point.h"
#include "utils/memutils.h"
#include "utils/timestamp.h"

Include dependency graph for twophase.c:

Data Structures
struct	GlobalTransactionData

struct	TwoPhaseStateData

struct	Working_State

struct	TwoPhaseRecordOnDisk

struct	StateFileChunk

struct	xllist

Macros
#define	TWOPHASE_DIR "pg_twophase"

#define	TWOPHASE_MAGIC 0x57F94534 /* format identifier */

Functions
static void	PrepareRedoRemoveFull (FullTransactionId fxid, bool giveWarning)

static void	RecordTransactionCommitPrepared (TransactionId xid, int nchildren, TransactionId children, int nrels, RelFileLocator rels, int nstats, xl_xact_stats_item stats, int ninvalmsgs, SharedInvalidationMessage invalmsgs, bool initfileinval, const char *gid)

static void	RecordTransactionAbortPrepared (TransactionId xid, int nchildren, TransactionId children, int nrels, RelFileLocator rels, int nstats, xl_xact_stats_item stats, const char gid)

static void	ProcessRecords (char *bufptr, FullTransactionId fxid, const TwoPhaseCallback callbacks[])

static void	RemoveGXact (GlobalTransaction gxact)

static void	XlogReadTwoPhaseData (XLogRecPtr lsn, char *buf, int len)

static char *	ProcessTwoPhaseBuffer (FullTransactionId fxid, XLogRecPtr prepare_start_lsn, bool fromdisk, bool setParent, bool setNextXid)

static void	MarkAsPreparingGuts (GlobalTransaction gxact, FullTransactionId fxid, const char *gid, TimestampTz prepared_at, Oid owner, Oid databaseid)

static void	RemoveTwoPhaseFile (FullTransactionId fxid, bool giveWarning)

static void	RecreateTwoPhaseFile (FullTransactionId fxid, void *content, int len)

Size	TwoPhaseShmemSize (void)

void	TwoPhaseShmemInit (void)

static void	AtProcExit_Twophase (int code, Datum arg)

void	AtAbort_Twophase (void)

void	PostPrepare_Twophase (void)

GlobalTransaction	MarkAsPreparing (FullTransactionId fxid, const char *gid, TimestampTz prepared_at, Oid owner, Oid databaseid)

static void	GXactLoadSubxactData (GlobalTransaction gxact, int nsubxacts, TransactionId *children)

static void	MarkAsPrepared (GlobalTransaction gxact, bool lock_held)

static GlobalTransaction	LockGXact (const char *gid, Oid user)

static int	GetPreparedTransactionList (GlobalTransaction *gxacts)

Datum	pg_prepared_xact (PG_FUNCTION_ARGS)

static GlobalTransaction	TwoPhaseGetGXact (FullTransactionId fxid, bool lock_held)

TransactionId	TwoPhaseGetXidByVirtualXID (VirtualTransactionId vxid, bool *have_more)

ProcNumber	TwoPhaseGetDummyProcNumber (FullTransactionId fxid, bool lock_held)

PGPROC *	TwoPhaseGetDummyProc (FullTransactionId fxid, bool lock_held)

static FullTransactionId	AdjustToFullTransactionId (TransactionId xid)

static int	TwoPhaseFilePath (char *path, FullTransactionId fxid)

static void	save_state_data (const void *data, uint32 len)

void	StartPrepare (GlobalTransaction gxact)

void	EndPrepare (GlobalTransaction gxact)

void	RegisterTwoPhaseRecord (TwoPhaseRmgrId rmid, uint16 info, const void *data, uint32 len)

static char *	ReadTwoPhaseFile (FullTransactionId fxid, bool missing_ok)

bool	StandbyTransactionIdIsPrepared (TransactionId xid)

void	FinishPreparedTransaction (const char *gid, bool isCommit)

void	CheckPointTwoPhase (XLogRecPtr redo_horizon)

void	restoreTwoPhaseData (void)

TransactionId	PrescanPreparedTransactions (TransactionId *xids_p, int nxids_p)

void	StandbyRecoverPreparedTransactions (void)

void	RecoverPreparedTransactions (void)

void	PrepareRedoAdd (FullTransactionId fxid, char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn, ReplOriginId origin_id)

void	PrepareRedoRemove (TransactionId xid, bool giveWarning)

bool	LookupGXact (const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp)

void	TwoPhaseTransactionGid (Oid subid, TransactionId xid, char *gid_res, int szgid)

static bool	IsTwoPhaseTransactionGidForSubid (Oid subid, char *gid)

bool	LookupGXactBySubid (Oid subid)

TransactionId	TwoPhaseGetOldestXidInCommit (void)

Variables
int	max_prepared_xacts = 0

static TwoPhaseStateData *	TwoPhaseState

static GlobalTransaction	MyLockedGxact = NULL

static bool	twophaseExitRegistered = false

static struct xllist	records

Macro Definition Documentation

◆ TWOPHASE_DIR

#define TWOPHASE_DIR "pg_twophase"

Definition at line 113 of file twophase.c.

◆ TWOPHASE_MAGIC

#define TWOPHASE_MAGIC 0x57F94534 /* format identifier */

Definition at line 976 of file twophase.c.

Typedef Documentation

◆ GlobalTransactionData

typedef struct GlobalTransactionData GlobalTransactionData

◆ StateFileChunk

typedef struct StateFileChunk StateFileChunk

◆ TwoPhaseFileHeader

typedef xl_xact_prepare TwoPhaseFileHeader

Definition at line 978 of file twophase.c.

◆ TwoPhaseRecordOnDisk

typedef struct TwoPhaseRecordOnDisk TwoPhaseRecordOnDisk

◆ TwoPhaseStateData

typedef struct TwoPhaseStateData TwoPhaseStateData

Function Documentation

◆ AdjustToFullTransactionId()

static FullTransactionId AdjustToFullTransactionId ( TransactionId xid )

inlinestatic

Definition at line 943 of file twophase.c.

{
    Assert(TransactionIdIsValid(xid));
    return FullTransactionIdFromAllowableAt(ReadNextFullTransactionId(), xid);
}

References Assert, FullTransactionIdFromAllowableAt(), ReadNextFullTransactionId(), and TransactionIdIsValid.

Referenced by StandbyTransactionIdIsPrepared().

◆ AtAbort_Twophase()

void AtAbort_Twophase ( void )

Definition at line 306 of file twophase.c.

{
    if (MyLockedGxact == NULL)
        return;
 
    /*
     * What to do with the locked global transaction entry?  If we were in the
     * process of preparing the transaction, but haven't written the WAL
     * record and state file yet, the transaction must not be considered as
     * prepared.  Likewise, if we are in the process of finishing an
     * already-prepared transaction, and fail after having already written the
     * 2nd phase commit or rollback record to the WAL, the transaction should
     * not be considered as prepared anymore.  In those cases, just remove the
     * entry from shared memory.
     *
     * Otherwise, the entry must be left in place so that the transaction can
     * be finished later, so just unlock it.
     *
     * If we abort during prepare, after having written the WAL record, we
     * might not have transferred all locks and other state to the prepared
     * transaction yet.  Likewise, if we abort during commit or rollback,
     * after having written the WAL record, we might not have released all the
     * resources held by the transaction yet.  In those cases, the in-memory
     * state can be wrong, but it's too late to back out.
     */
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    if (!MyLockedGxact->valid)
        RemoveGXact(MyLockedGxact);
    else
        MyLockedGxact->locking_backend = INVALID_PROC_NUMBER;
    LWLockRelease(TwoPhaseStateLock);
 
    MyLockedGxact = NULL;
}

References fb(), INVALID_PROC_NUMBER, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyLockedGxact, RemoveGXact(), and GlobalTransactionData::valid.

Referenced by AbortTransaction(), and AtProcExit_Twophase().

◆ AtProcExit_Twophase()

static void AtProcExit_Twophase	(	int	code,
		Datum	arg
	)

static

Definition at line 296 of file twophase.c.

{
    /* same logic as abort */
    AtAbort_Twophase();
}

References AtAbort_Twophase().

Referenced by LockGXact(), and MarkAsPreparing().

◆ CheckPointTwoPhase()

void CheckPointTwoPhase ( XLogRecPtr redo_horizon )

Definition at line 1822 of file twophase.c.

{
    int         i;
    int         serialized_xacts = 0;
 
    if (max_prepared_xacts <= 0)
        return;                 /* nothing to do */
 
    TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();
 
    /*
     * We are expecting there to be zero GXACTs that need to be copied to
     * disk, so we perform all I/O while holding TwoPhaseStateLock for
     * simplicity. This prevents any new xacts from preparing while this
     * occurs, which shouldn't be a problem since the presence of long-lived
     * prepared xacts indicates the transaction manager isn't active.
     *
     * It's also possible to move I/O out of the lock, but on every error we
     * should check whether somebody committed our transaction in different
     * backend. Let's leave this optimization for future, if somebody will
     * spot that this place cause bottleneck.
     *
     * Note that it isn't possible for there to be a GXACT with a
     * prepare_end_lsn set prior to the last checkpoint yet is marked invalid,
     * because of the efforts with delayChkptFlags.
     */
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        /*
         * Note that we are using gxact not PGPROC so this works in recovery
         * also
         */
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        if ((gxact->valid || gxact->inredo) &&
            !gxact->ondisk &&
            gxact->prepare_end_lsn <= redo_horizon)
        {
            char       *buf;
            int         len;
 
            XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len);
            RecreateTwoPhaseFile(gxact->fxid, buf, len);
            gxact->ondisk = true;
            gxact->prepare_start_lsn = InvalidXLogRecPtr;
            gxact->prepare_end_lsn = InvalidXLogRecPtr;
            pfree(buf);
            serialized_xacts++;
        }
    }
    LWLockRelease(TwoPhaseStateLock);
 
    /*
     * Flush unconditionally the parent directory to make any information
     * durable on disk.  Two-phase files could have been removed and those
     * removals need to be made persistent as well as any files newly created
     * previously since the last checkpoint.
     */
    fsync_fname(TWOPHASE_DIR, true);
 
    TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
 
    if (log_checkpoints && serialized_xacts > 0)
        ereport(LOG,
                (errmsg_plural("%u two-phase state file was written "
                               "for a long-running prepared transaction",
                               "%u two-phase state files were written "
                               "for long-running prepared transactions",
                               serialized_xacts,
                               serialized_xacts)));
}

References buf, ereport, errmsg_plural(), fb(), fsync_fname(), i, InvalidXLogRecPtr, len, LOG, log_checkpoints, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_prepared_xacts, TwoPhaseStateData::numPrepXacts, pfree(), TwoPhaseStateData::prepXacts, RecreateTwoPhaseFile(), TWOPHASE_DIR, TwoPhaseState, and XlogReadTwoPhaseData().

Referenced by CheckPointGuts().

◆ EndPrepare()

void EndPrepare ( GlobalTransaction gxact )

Definition at line 1145 of file twophase.c.

{
    TwoPhaseFileHeader *hdr;
    StateFileChunk *record;
    bool        replorigin;
 
    /* Add the end sentinel to the list of 2PC records */
    RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
                           NULL, 0);
 
    /* Go back and fill in total_len in the file header record */
    hdr = (TwoPhaseFileHeader *) records.head->data;
    Assert(hdr->magic == TWOPHASE_MAGIC);
    hdr->total_len = records.total_len + sizeof(pg_crc32c);
 
    replorigin = (replorigin_xact_state.origin != InvalidReplOriginId &&
                  replorigin_xact_state.origin != DoNotReplicateId);
 
    if (replorigin)
    {
        hdr->origin_lsn = replorigin_xact_state.origin_lsn;
        hdr->origin_timestamp = replorigin_xact_state.origin_timestamp;
    }
 
    /*
     * If the data size exceeds MaxAllocSize, we won't be able to read it in
     * ReadTwoPhaseFile. Check for that now, rather than fail in the case
     * where we write data to file and then re-read at commit time.
     */
    if (hdr->total_len > MaxAllocSize)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("two-phase state file maximum length exceeded")));
 
    /*
     * Now writing 2PC state data to WAL. We let the WAL's CRC protection
     * cover us, so no need to calculate a separate CRC.
     *
     * We have to set DELAY_CHKPT_START here, too; otherwise a checkpoint
     * starting immediately after the WAL record is inserted could complete
     * without fsync'ing our state file.  (This is essentially the same kind
     * of race condition as the COMMIT-to-clog-write case that
     * RecordTransactionCommit uses DELAY_CHKPT_IN_COMMIT for; see notes
     * there.) Note that DELAY_CHKPT_IN_COMMIT is used to find transactions in
     * the critical commit section. We need to know about such transactions
     * for conflict detection in logical replication. See
     * GetOldestActiveTransactionId(true, false) and its use.
     *
     * We save the PREPARE record's location in the gxact for later use by
     * CheckPointTwoPhase.
     */
    XLogEnsureRecordSpace(0, records.num_chunks);
 
    START_CRIT_SECTION();
 
    Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
    MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
    XLogBeginInsert();
    for (record = records.head; record != NULL; record = record->next)
        XLogRegisterData(record->data, record->len);
 
    XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
 
    gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
 
    if (replorigin)
    {
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_xact_state.origin_lsn,
                                   gxact->prepare_end_lsn);
    }
 
    XLogFlush(gxact->prepare_end_lsn);
 
    /* If we crash now, we have prepared: WAL replay will fix things */
 
    /* Store record's start location to read that later on Commit */
    gxact->prepare_start_lsn = ProcLastRecPtr;
 
    /*
     * Mark the prepared transaction as valid.  As soon as xact.c marks MyProc
     * as not running our XID (which it will do immediately after this
     * function returns), others can commit/rollback the xact.
     *
     * NB: a side effect of this is to make a dummy ProcArray entry for the
     * prepared XID.  This must happen before we clear the XID from MyProc /
     * ProcGlobal->xids[], else there is a window where the XID is not running
     * according to TransactionIdIsInProgress, and onlookers would be entitled
     * to assume the xact crashed.  Instead we have a window where the same
     * XID appears twice in ProcArray, which is OK.
     */
    MarkAsPrepared(gxact, false);
 
    /*
     * Now we can mark ourselves as out of the commit critical section: a
     * checkpoint starting after this will certainly see the gxact as a
     * candidate for fsyncing.
     */
    MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
 
    /*
     * Remember that we have this GlobalTransaction entry locked for us.  If
     * we crash after this point, it's too late to abort, but we must unlock
     * it so that the prepared transaction can be committed or rolled back.
     */
    MyLockedGxact = gxact;
 
    END_CRIT_SECTION();
 
    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked the prepare, but still show as
     * running in the procarray (twice!) and continue to hold locks.
     */
    SyncRepWaitForLSN(gxact->prepare_end_lsn, false);
 
    records.tail = records.head = NULL;
    records.num_chunks = 0;
}

References Assert, StateFileChunk::data, DELAY_CHKPT_START, PGPROC::delayChkptFlags, DoNotReplicateId, END_CRIT_SECTION, ereport, errcode(), errmsg(), ERROR, fb(), xllist::head, InvalidReplOriginId, StateFileChunk::len, xl_xact_prepare::magic, MarkAsPrepared(), MaxAllocSize, MyLockedGxact, MyProc, StateFileChunk::next, xllist::num_chunks, ReplOriginXactState::origin, xl_xact_prepare::origin_lsn, ReplOriginXactState::origin_lsn, xl_xact_prepare::origin_timestamp, ReplOriginXactState::origin_timestamp, ProcLastRecPtr, records, RegisterTwoPhaseRecord(), replorigin_session_advance(), replorigin_xact_state, START_CRIT_SECTION, SyncRepWaitForLSN(), xllist::tail, xllist::total_len, xl_xact_prepare::total_len, TWOPHASE_MAGIC, TWOPHASE_RM_END_ID, XLOG_INCLUDE_ORIGIN, XLOG_XACT_PREPARE, XLogBeginInsert(), XLogEnsureRecordSpace(), XLogFlush(), XLogInsert(), XLogRegisterData(), and XLogSetRecordFlags().

Referenced by PrepareTransaction().

◆ FinishPreparedTransaction()

void FinishPreparedTransaction	(	const char *	gid,
		bool	isCommit
	)

Definition at line 1497 of file twophase.c.

{
    GlobalTransaction gxact;
    PGPROC     *proc;
    FullTransactionId fxid;
    TransactionId xid;
    bool        ondisk;
    char       *buf;
    char       *bufptr;
    TwoPhaseFileHeader *hdr;
    TransactionId latestXid;
    TransactionId *children;
    RelFileLocator *commitrels;
    RelFileLocator *abortrels;
    RelFileLocator *delrels;
    int         ndelrels;
    xl_xact_stats_item *commitstats;
    xl_xact_stats_item *abortstats;
    SharedInvalidationMessage *invalmsgs;
 
    /*
     * Validate the GID, and lock the GXACT to ensure that two backends do not
     * try to commit the same GID at once.
     */
    gxact = LockGXact(gid, GetUserId());
    proc = GetPGProcByNumber(gxact->pgprocno);
    fxid = gxact->fxid;
    xid = XidFromFullTransactionId(fxid);
 
    /*
     * Read and validate 2PC state data. State data will typically be stored
     * in WAL files if the LSN is after the last checkpoint record, or moved
     * to disk if for some reason they have lived for a long time.
     */
    if (gxact->ondisk)
        buf = ReadTwoPhaseFile(fxid, false);
    else
        XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
 
 
    /*
     * Disassemble the header area
     */
    hdr = (TwoPhaseFileHeader *) buf;
    Assert(TransactionIdEquals(hdr->xid, xid));
    bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
    bufptr += MAXALIGN(hdr->gidlen);
    children = (TransactionId *) bufptr;
    bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
    commitrels = (RelFileLocator *) bufptr;
    bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileLocator));
    abortrels = (RelFileLocator *) bufptr;
    bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileLocator));
    commitstats = (xl_xact_stats_item *) bufptr;
    bufptr += MAXALIGN(hdr->ncommitstats * sizeof(xl_xact_stats_item));
    abortstats = (xl_xact_stats_item *) bufptr;
    bufptr += MAXALIGN(hdr->nabortstats * sizeof(xl_xact_stats_item));
    invalmsgs = (SharedInvalidationMessage *) bufptr;
    bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
 
    /* compute latestXid among all children */
    latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
 
    /* Prevent cancel/die interrupt while cleaning up */
    HOLD_INTERRUPTS();
 
    /*
     * The order of operations here is critical: make the XLOG entry for
     * commit or abort, then mark the transaction committed or aborted in
     * pg_xact, then remove its PGPROC from the global ProcArray (which means
     * TransactionIdIsInProgress will stop saying the prepared xact is in
     * progress), then run the post-commit or post-abort callbacks. The
     * callbacks will release the locks the transaction held.
     */
    if (isCommit)
        RecordTransactionCommitPrepared(xid,
                                        hdr->nsubxacts, children,
                                        hdr->ncommitrels, commitrels,
                                        hdr->ncommitstats,
                                        commitstats,
                                        hdr->ninvalmsgs, invalmsgs,
                                        hdr->initfileinval, gid);
    else
        RecordTransactionAbortPrepared(xid,
                                       hdr->nsubxacts, children,
                                       hdr->nabortrels, abortrels,
                                       hdr->nabortstats,
                                       abortstats,
                                       gid);
 
    ProcArrayRemove(proc, latestXid);
 
    /*
     * In case we fail while running the callbacks, mark the gxact invalid so
     * no one else will try to commit/rollback, and so it will be recycled if
     * we fail after this point.  It is still locked by our backend so it
     * won't go away yet.
     *
     * (We assume it's safe to do this without taking TwoPhaseStateLock.)
     */
    gxact->valid = false;
 
    /*
     * We have to remove any files that were supposed to be dropped. For
     * consistency with the regular xact.c code paths, must do this before
     * releasing locks, so do it before running the callbacks.
     *
     * NB: this code knows that we couldn't be dropping any temp rels ...
     */
    if (isCommit)
    {
        delrels = commitrels;
        ndelrels = hdr->ncommitrels;
    }
    else
    {
        delrels = abortrels;
        ndelrels = hdr->nabortrels;
    }
 
    /* Make sure files supposed to be dropped are dropped */
    DropRelationFiles(delrels, ndelrels, false);
 
    if (isCommit)
        pgstat_execute_transactional_drops(hdr->ncommitstats, commitstats, false);
    else
        pgstat_execute_transactional_drops(hdr->nabortstats, abortstats, false);
 
    /*
     * Handle cache invalidation messages.
     *
     * Relcache init file invalidation requires processing both before and
     * after we send the SI messages, only when committing.  See
     * AtEOXact_Inval().
     */
    if (isCommit)
    {
        if (hdr->initfileinval)
            RelationCacheInitFilePreInvalidate();
        SendSharedInvalidMessages(invalmsgs, hdr->ninvalmsgs);
        if (hdr->initfileinval)
            RelationCacheInitFilePostInvalidate();
    }
 
    /*
     * Acquire the two-phase lock.  We want to work on the two-phase callbacks
     * while holding it to avoid potential conflicts with other transactions
     * attempting to use the same GID, so the lock is released once the shared
     * memory state is cleared.
     */
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 
    /* And now do the callbacks */
    if (isCommit)
        ProcessRecords(bufptr, fxid, twophase_postcommit_callbacks);
    else
        ProcessRecords(bufptr, fxid, twophase_postabort_callbacks);
 
    PredicateLockTwoPhaseFinish(fxid, isCommit);
 
    /*
     * Read this value while holding the two-phase lock, as the on-disk 2PC
     * file is physically removed after the lock is released.
     */
    ondisk = gxact->ondisk;
 
    /* Clear shared memory state */
    RemoveGXact(gxact);
 
    /*
     * Release the lock as all callbacks are called and shared memory cleanup
     * is done.
     */
    LWLockRelease(TwoPhaseStateLock);
 
    /* Count the prepared xact as committed or aborted */
    AtEOXact_PgStat(isCommit, false);
 
    /*
     * And now we can clean up any files we may have left.
     */
    if (ondisk)
        RemoveTwoPhaseFile(fxid, true);
 
    MyLockedGxact = NULL;
 
    RESUME_INTERRUPTS();
 
    pfree(buf);
}

References Assert, AtEOXact_PgStat(), buf, DropRelationFiles(), fb(), GetPGProcByNumber, GetUserId(), xl_xact_prepare::gidlen, HOLD_INTERRUPTS, xl_xact_prepare::initfileinval, LockGXact(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXALIGN, MyLockedGxact, xl_xact_prepare::nabortrels, xl_xact_prepare::nabortstats, xl_xact_prepare::ncommitrels, xl_xact_prepare::ncommitstats, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, pfree(), pgstat_execute_transactional_drops(), PredicateLockTwoPhaseFinish(), ProcArrayRemove(), ProcessRecords(), ReadTwoPhaseFile(), RecordTransactionAbortPrepared(), RecordTransactionCommitPrepared(), RelationCacheInitFilePostInvalidate(), RelationCacheInitFilePreInvalidate(), RemoveGXact(), RemoveTwoPhaseFile(), RESUME_INTERRUPTS, SendSharedInvalidMessages(), TransactionIdEquals, TransactionIdLatest(), twophase_postabort_callbacks, twophase_postcommit_callbacks, xl_xact_prepare::xid, XidFromFullTransactionId, and XlogReadTwoPhaseData().

Referenced by apply_handle_commit_prepared(), apply_handle_rollback_prepared(), and standard_ProcessUtility().

◆ GetPreparedTransactionList()

static int GetPreparedTransactionList ( GlobalTransaction * gxacts )

static

Definition at line 670 of file twophase.c.

{
    GlobalTransaction array;
    int         num;
    int         i;
 
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 
    if (TwoPhaseState->numPrepXacts == 0)
    {
        LWLockRelease(TwoPhaseStateLock);
 
        *gxacts = NULL;
        return 0;
    }
 
    num = TwoPhaseState->numPrepXacts;
    array = palloc_array(GlobalTransactionData, num);
    *gxacts = array;
    for (i = 0; i < num; i++)
        memcpy(array + i, TwoPhaseState->prepXacts[i],
               sizeof(GlobalTransactionData));
 
    LWLockRelease(TwoPhaseStateLock);
 
    return num;
}

References fb(), i, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, palloc_array, TwoPhaseStateData::prepXacts, and TwoPhaseState.

Referenced by pg_prepared_xact().

◆ GXactLoadSubxactData()

static void GXactLoadSubxactData	(	GlobalTransaction	gxact,
		int	nsubxacts,
		TransactionId *	children
	)

static

Definition at line 508 of file twophase.c.

{
    PGPROC     *proc = GetPGProcByNumber(gxact->pgprocno);
 
    /* We need no extra lock since the GXACT isn't valid yet */
    if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
    {
        proc->subxidStatus.overflowed = true;
        nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
    }
    if (nsubxacts > 0)
    {
        memcpy(proc->subxids.xids, children,
               nsubxacts * sizeof(TransactionId));
        proc->subxidStatus.count = nsubxacts;
    }
}

References XidCacheStatus::count, fb(), GetPGProcByNumber, XidCacheStatus::overflowed, PGPROC_MAX_CACHED_SUBXIDS, PGPROC::subxids, PGPROC::subxidStatus, and XidCache::xids.

Referenced by RecoverPreparedTransactions(), and StartPrepare().

◆ IsTwoPhaseTransactionGidForSubid()

static bool IsTwoPhaseTransactionGidForSubid	(	Oid	subid,
		char *	gid
	)

static

Definition at line 2765 of file twophase.c.

{
    int         ret;
    Oid         subid_from_gid;
    TransactionId xid_from_gid;
    char        gid_tmp[GIDSIZE];
 
    /* Extract the subid and xid from the given GID */
    ret = sscanf(gid, "pg_gid_%u_%u", &subid_from_gid, &xid_from_gid);
 
    /*
     * Check that the given GID has expected format, and at least the subid
     * matches.
     */
    if (ret != 2 || subid != subid_from_gid)
        return false;
 
    /*
     * Reconstruct a temporary GID based on the subid and xid extracted from
     * the given GID and check whether the temporary GID and the given GID
     * match.
     */
    TwoPhaseTransactionGid(subid, xid_from_gid, gid_tmp, sizeof(gid_tmp));
 
    return strcmp(gid, gid_tmp) == 0;
}

References fb(), GIDSIZE, and TwoPhaseTransactionGid().

Referenced by LookupGXactBySubid().

◆ LockGXact()

static GlobalTransaction LockGXact	(	const char *	gid,
		Oid	user
	)

static

Definition at line 556 of file twophase.c.

{
    int         i;
 
    /* on first call, register the exit hook */
    if (!twophaseExitRegistered)
    {
        before_shmem_exit(AtProcExit_Twophase, 0);
        twophaseExitRegistered = true;
    }
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGPROC     *proc = GetPGProcByNumber(gxact->pgprocno);
 
        /* Ignore not-yet-valid GIDs */
        if (!gxact->valid)
            continue;
        if (strcmp(gxact->gid, gid) != 0)
            continue;
 
        /* Found it, but has someone else got it locked? */
        if (gxact->locking_backend != INVALID_PROC_NUMBER)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("prepared transaction with identifier \"%s\" is busy",
                            gid)));
 
        if (user != gxact->owner && !superuser_arg(user))
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("permission denied to finish prepared transaction"),
                     errhint("Must be superuser or the user that prepared the transaction.")));
 
        /*
         * Note: it probably would be possible to allow committing from
         * another database; but at the moment NOTIFY is known not to work and
         * there may be some other issues as well.  Hence disallow until
         * someone gets motivated to make it work.
         */
        if (MyDatabaseId != proc->databaseId)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("prepared transaction belongs to another database"),
                     errhint("Connect to the database where the transaction was prepared to finish it.")));
 
        /* OK for me to lock it */
        gxact->locking_backend = MyProcNumber;
        MyLockedGxact = gxact;
 
        LWLockRelease(TwoPhaseStateLock);
 
        return gxact;
    }
 
    LWLockRelease(TwoPhaseStateLock);
 
    ereport(ERROR,
            (errcode(ERRCODE_UNDEFINED_OBJECT),
             errmsg("prepared transaction with identifier \"%s\" does not exist",
                    gid)));
 
    /* NOTREACHED */
    return NULL;
}

References AtProcExit_Twophase(), before_shmem_exit(), PGPROC::databaseId, ereport, errcode(), errhint(), errmsg(), ERROR, fb(), GetPGProcByNumber, i, INVALID_PROC_NUMBER, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyDatabaseId, MyLockedGxact, MyProcNumber, TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, superuser_arg(), twophaseExitRegistered, TwoPhaseState, and user.

Referenced by FinishPreparedTransaction().

◆ LookupGXact()

bool LookupGXact	(	const char *	gid,
		XLogRecPtr	prepare_end_lsn,
		TimestampTz	origin_prepare_timestamp
	)

Definition at line 2688 of file twophase.c.

{
    int         i;
    bool        found = false;
 
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        /* Ignore not-yet-valid GIDs. */
        if (gxact->valid && strcmp(gxact->gid, gid) == 0)
        {
            char       *buf;
            TwoPhaseFileHeader *hdr;
 
            /*
             * We are not expecting collisions of GXACTs (same gid) between
             * publisher and subscribers, so we perform all I/O while holding
             * TwoPhaseStateLock for simplicity.
             *
             * To move the I/O out of the lock, we need to ensure that no
             * other backend commits the prepared xact in the meantime. We can
             * do this optimization if we encounter many collisions in GID
             * between publisher and subscriber.
             */
            if (gxact->ondisk)
                buf = ReadTwoPhaseFile(gxact->fxid, false);
            else
            {
                Assert(gxact->prepare_start_lsn);
                XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
            }
 
            hdr = (TwoPhaseFileHeader *) buf;
 
            if (hdr->origin_lsn == prepare_end_lsn &&
                hdr->origin_timestamp == origin_prepare_timestamp)
            {
                found = true;
                pfree(buf);
                break;
            }
 
            pfree(buf);
        }
    }
    LWLockRelease(TwoPhaseStateLock);
    return found;
}

References Assert, buf, fb(), i, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, xl_xact_prepare::origin_lsn, xl_xact_prepare::origin_timestamp, pfree(), TwoPhaseStateData::prepXacts, ReadTwoPhaseFile(), TwoPhaseState, and XlogReadTwoPhaseData().

Referenced by apply_handle_rollback_prepared().

◆ LookupGXactBySubid()

bool LookupGXactBySubid ( Oid subid )

Definition at line 2797 of file twophase.c.

{
    bool        found = false;
 
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
    for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        /* Ignore not-yet-valid GIDs. */
        if (gxact->valid &&
            IsTwoPhaseTransactionGidForSubid(subid, gxact->gid))
        {
            found = true;
            break;
        }
    }
    LWLockRelease(TwoPhaseStateLock);
 
    return found;
}

References fb(), i, IsTwoPhaseTransactionGidForSubid(), LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, and TwoPhaseState.

Referenced by AlterSubscription().

◆ MarkAsPrepared()

static void MarkAsPrepared	(	GlobalTransaction	gxact,
		bool	lock_held
	)

static

Definition at line 534 of file twophase.c.

{
    /* Lock here may be overkill, but I'm not convinced of that ... */
    if (!lock_held)
        LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    Assert(!gxact->valid);
    gxact->valid = true;
    if (!lock_held)
        LWLockRelease(TwoPhaseStateLock);
 
    /*
     * Put it into the global ProcArray so TransactionIdIsInProgress considers
     * the XID as still running.
     */
    ProcArrayAdd(GetPGProcByNumber(gxact->pgprocno));
}

References Assert, fb(), GetPGProcByNumber, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), and ProcArrayAdd().

Referenced by EndPrepare(), and RecoverPreparedTransactions().

◆ MarkAsPreparing()

GlobalTransaction MarkAsPreparing	(	FullTransactionId	fxid,
		const char *	gid,
		TimestampTz	prepared_at,
		Oid	owner,
		Oid	databaseid
	)

Definition at line 361 of file twophase.c.

{
    GlobalTransaction gxact;
    int         i;
 
    if (strlen(gid) >= GIDSIZE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("transaction identifier \"%s\" is too long",
                        gid)));
 
    /* fail immediately if feature is disabled */
    if (max_prepared_xacts == 0)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("prepared transactions are disabled"),
                 errhint("Set \"max_prepared_transactions\" to a nonzero value.")));
 
    /* on first call, register the exit hook */
    if (!twophaseExitRegistered)
    {
        before_shmem_exit(AtProcExit_Twophase, 0);
        twophaseExitRegistered = true;
    }
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 
    /* Check for conflicting GID */
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        gxact = TwoPhaseState->prepXacts[i];
        if (strcmp(gxact->gid, gid) == 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_OBJECT),
                     errmsg("transaction identifier \"%s\" is already in use",
                            gid)));
        }
    }
 
    /* Get a free gxact from the freelist */
    if (TwoPhaseState->freeGXacts == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("maximum number of prepared transactions reached"),
                 errhint("Increase \"max_prepared_transactions\" (currently %d).",
                         max_prepared_xacts)));
    gxact = TwoPhaseState->freeGXacts;
    TwoPhaseState->freeGXacts = gxact->next;
 
    MarkAsPreparingGuts(gxact, fxid, gid, prepared_at, owner, databaseid);
 
    gxact->ondisk = false;
 
    /* And insert it into the active array */
    Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
    TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
 
    LWLockRelease(TwoPhaseStateLock);
 
    return gxact;
}

References Assert, AtProcExit_Twophase(), before_shmem_exit(), ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errhint(), errmsg(), ERROR, fb(), TwoPhaseStateData::freeGXacts, GIDSIZE, i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MarkAsPreparingGuts(), max_prepared_xacts, GlobalTransactionData::next, TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, twophaseExitRegistered, and TwoPhaseState.

Referenced by PrepareTransaction().

◆ MarkAsPreparingGuts()

static void MarkAsPreparingGuts	(	GlobalTransaction	gxact,
		FullTransactionId	fxid,
		const char *	gid,
		TimestampTz	prepared_at,
		Oid	owner,
		Oid	databaseid
	)

static

Definition at line 435 of file twophase.c.

{
    PGPROC     *proc;
    int         i;
    TransactionId xid = XidFromFullTransactionId(fxid);
 
    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 
    Assert(gxact != NULL);
    proc = GetPGProcByNumber(gxact->pgprocno);
 
    /* Initialize the PGPROC entry */
    MemSet(proc, 0, sizeof(PGPROC));
    dlist_node_init(&proc->links);
    proc->waitStatus = PROC_WAIT_STATUS_OK;
    if (LocalTransactionIdIsValid(MyProc->vxid.lxid))
    {
        /* clone VXID, for TwoPhaseGetXidByVirtualXID() to find */
        proc->vxid.lxid = MyProc->vxid.lxid;
        proc->vxid.procNumber = MyProcNumber;
    }
    else
    {
        Assert(AmStartupProcess() || !IsPostmasterEnvironment);
        /* GetLockConflicts() uses this to specify a wait on the XID */
        proc->vxid.lxid = xid;
        proc->vxid.procNumber = INVALID_PROC_NUMBER;
    }
    proc->xid = xid;
    Assert(proc->xmin == InvalidTransactionId);
    proc->delayChkptFlags = 0;
    proc->statusFlags = 0;
    proc->pid = 0;
    proc->databaseId = databaseid;
    proc->roleId = owner;
    proc->tempNamespaceId = InvalidOid;
    proc->isRegularBackend = false;
    proc->lwWaiting = LW_WS_NOT_WAITING;
    proc->lwWaitMode = 0;
    proc->waitLock = NULL;
    proc->waitProcLock = NULL;
    pg_atomic_init_u64(&proc->waitStart, 0);
    for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
        dlist_init(&proc->myProcLocks[i]);
    /* subxid data must be filled later by GXactLoadSubxactData */
    proc->subxidStatus.overflowed = false;
    proc->subxidStatus.count = 0;
 
    gxact->prepared_at = prepared_at;
    gxact->fxid = fxid;
    gxact->owner = owner;
    gxact->locking_backend = MyProcNumber;
    gxact->valid = false;
    gxact->inredo = false;
    strcpy(gxact->gid, gid);
 
    /*
     * Remember that we have this GlobalTransaction entry locked for us. If we
     * abort after this, we must release it.
     */
    MyLockedGxact = gxact;
}

References AmStartupProcess, Assert, XidCacheStatus::count, PGPROC::databaseId, PGPROC::delayChkptFlags, dlist_init(), dlist_node_init(), fb(), GetPGProcByNumber, i, INVALID_PROC_NUMBER, InvalidOid, InvalidTransactionId, IsPostmasterEnvironment, PGPROC::isRegularBackend, PGPROC::links, LocalTransactionIdIsValid, LW_EXCLUSIVE, LW_WS_NOT_WAITING, LWLockHeldByMeInMode(), PGPROC::lwWaiting, PGPROC::lwWaitMode, PGPROC::lxid, MemSet, MyLockedGxact, MyProc, PGPROC::myProcLocks, MyProcNumber, NUM_LOCK_PARTITIONS, XidCacheStatus::overflowed, pg_atomic_init_u64(), PGPROC::pid, PROC_WAIT_STATUS_OK, PGPROC::procNumber, PGPROC::roleId, PGPROC::statusFlags, PGPROC::subxidStatus, PGPROC::tempNamespaceId, PGPROC::vxid, PGPROC::waitLock, PGPROC::waitProcLock, PGPROC::waitStart, PGPROC::waitStatus, PGPROC::xid, XidFromFullTransactionId, and PGPROC::xmin.

Referenced by MarkAsPreparing(), and RecoverPreparedTransactions().

◆ pg_prepared_xact()

Datum pg_prepared_xact ( PG_FUNCTION_ARGS )

Definition at line 715 of file twophase.c.

{
    FuncCallContext *funcctx;
    Working_State *status;
 
    if (SRF_IS_FIRSTCALL())
    {
        TupleDesc   tupdesc;
        MemoryContext oldcontext;
 
        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();
 
        /*
         * Switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
 
        /* build tupdesc for result tuples */
        /* this had better match pg_prepared_xacts view in system_views.sql */
        tupdesc = CreateTemplateTupleDesc(5);
        TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
                           XIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
                           TEXTOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
                           TIMESTAMPTZOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
                           OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
                           OIDOID, -1, 0);
 
        funcctx->tuple_desc = BlessTupleDesc(tupdesc);
 
        /*
         * Collect all the 2PC status information that we will format and send
         * out as a result set.
         */
        status = palloc_object(Working_State);
        funcctx->user_fctx = status;
 
        status->ngxacts = GetPreparedTransactionList(&status->array);
        status->currIdx = 0;
 
        MemoryContextSwitchTo(oldcontext);
    }
 
    funcctx = SRF_PERCALL_SETUP();
    status = (Working_State *) funcctx->user_fctx;
 
    while (status->array != NULL && status->currIdx < status->ngxacts)
    {
        GlobalTransaction gxact = &status->array[status->currIdx++];
        PGPROC     *proc = GetPGProcByNumber(gxact->pgprocno);
        Datum       values[5] = {0};
        bool        nulls[5] = {0};
        HeapTuple   tuple;
        Datum       result;
 
        if (!gxact->valid)
            continue;
 
        /*
         * Form tuple with appropriate data.
         */
 
        values[0] = TransactionIdGetDatum(proc->xid);
        values[1] = CStringGetTextDatum(gxact->gid);
        values[2] = TimestampTzGetDatum(gxact->prepared_at);
        values[3] = ObjectIdGetDatum(gxact->owner);
        values[4] = ObjectIdGetDatum(proc->databaseId);
 
        tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
        result = HeapTupleGetDatum(tuple);
        SRF_RETURN_NEXT(funcctx, result);
    }
 
    SRF_RETURN_DONE(funcctx);
}

References Working_State::array, BlessTupleDesc(), CreateTemplateTupleDesc(), CStringGetTextDatum, Working_State::currIdx, PGPROC::databaseId, fb(), GetPGProcByNumber, GetPreparedTransactionList(), heap_form_tuple(), HeapTupleGetDatum(), MemoryContextSwitchTo(), Working_State::ngxacts, ObjectIdGetDatum(), palloc_object, SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, TimestampTzGetDatum(), TransactionIdGetDatum(), TupleDescInitEntry(), values, and PGPROC::xid.

◆ PostPrepare_Twophase()

void PostPrepare_Twophase ( void )

Definition at line 346 of file twophase.c.

{
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    MyLockedGxact->locking_backend = INVALID_PROC_NUMBER;
    LWLockRelease(TwoPhaseStateLock);
 
    MyLockedGxact = NULL;
}

References fb(), INVALID_PROC_NUMBER, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), and MyLockedGxact.

Referenced by PrepareTransaction(), and RecoverPreparedTransactions().

◆ PrepareRedoAdd()

void PrepareRedoAdd	(	FullTransactionId	fxid,
		char *	buf,
		XLogRecPtr	start_lsn,
		XLogRecPtr	end_lsn,
		ReplOriginId	origin_id
	)

Definition at line 2507 of file twophase.c.

{
    TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf;
    char       *bufptr;
    const char *gid;
    GlobalTransaction gxact;
 
    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
    Assert(RecoveryInProgress());
 
    if (!FullTransactionIdIsValid(fxid))
    {
        Assert(InRecovery);
        fxid = FullTransactionIdFromAllowableAt(TransamVariables->nextXid,
                                                hdr->xid);
    }
 
    bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
    gid = (const char *) bufptr;
 
    /*
     * Reserve the GID for the given transaction in the redo code path.
     *
     * This creates a gxact struct and puts it into the active array.
     *
     * In redo, this struct is mainly used to track PREPARE/COMMIT entries in
     * shared memory. Hence, we only fill up the bare minimum contents here.
     * The gxact also gets marked with gxact->inredo set to true to indicate
     * that it got added in the redo phase
     */
 
    /*
     * In the event of a crash while a checkpoint was running, it may be
     * possible that some two-phase data found its way to disk while its
     * corresponding record needs to be replayed in the follow-up recovery. As
     * the 2PC data was on disk, it has already been restored at the beginning
     * of recovery with restoreTwoPhaseData(), so skip this record to avoid
     * duplicates in TwoPhaseState.  If a consistent state has been reached,
     * the record is added to TwoPhaseState and it should have no
     * corresponding file in pg_twophase.
     */
    if (XLogRecPtrIsValid(start_lsn))
    {
        char        path[MAXPGPATH];
 
        Assert(InRecovery);
        TwoPhaseFilePath(path, fxid);
 
        if (access(path, F_OK) == 0)
        {
            ereport(reachedConsistency ? ERROR : WARNING,
                    (errmsg("could not recover two-phase state file for transaction %u",
                            hdr->xid),
                     errdetail("Two-phase state file has been found in WAL record %X/%08X, but this transaction has already been restored from disk.",
                               LSN_FORMAT_ARGS(start_lsn))));
            return;
        }
 
        if (errno != ENOENT)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not access file \"%s\": %m", path)));
    }
 
    /* Get a free gxact from the freelist */
    if (TwoPhaseState->freeGXacts == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("maximum number of prepared transactions reached"),
                 errhint("Increase \"max_prepared_transactions\" (currently %d).",
                         max_prepared_xacts)));
    gxact = TwoPhaseState->freeGXacts;
    TwoPhaseState->freeGXacts = gxact->next;
 
    gxact->prepared_at = hdr->prepared_at;
    gxact->prepare_start_lsn = start_lsn;
    gxact->prepare_end_lsn = end_lsn;
    gxact->fxid = fxid;
    gxact->owner = hdr->owner;
    gxact->locking_backend = INVALID_PROC_NUMBER;
    gxact->valid = false;
    gxact->ondisk = !XLogRecPtrIsValid(start_lsn);
    gxact->inredo = true;       /* yes, added in redo */
    strcpy(gxact->gid, gid);
 
    /* And insert it into the active array */
    Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
    TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
 
    if (origin_id != InvalidReplOriginId)
    {
        /* recover apply progress */
        replorigin_advance(origin_id, hdr->origin_lsn, end_lsn,
                           false /* backward */ , false /* WAL */ );
    }
 
    elog(DEBUG2, "added 2PC data in shared memory for transaction %u of epoch %u",
         XidFromFullTransactionId(gxact->fxid),
         EpochFromFullTransactionId(gxact->fxid));
}

References Assert, buf, DEBUG2, elog, EpochFromFullTransactionId, ereport, errcode(), errcode_for_file_access(), errdetail(), errhint(), errmsg(), ERROR, fb(), TwoPhaseStateData::freeGXacts, FullTransactionIdFromAllowableAt(), FullTransactionIdIsValid, InRecovery, INVALID_PROC_NUMBER, InvalidReplOriginId, LSN_FORMAT_ARGS, LW_EXCLUSIVE, LWLockHeldByMeInMode(), max_prepared_xacts, MAXALIGN, MAXPGPATH, GlobalTransactionData::next, TransamVariablesData::nextXid, TwoPhaseStateData::numPrepXacts, xl_xact_prepare::origin_lsn, xl_xact_prepare::owner, GlobalTransactionData::prepared_at, xl_xact_prepare::prepared_at, TwoPhaseStateData::prepXacts, reachedConsistency, RecoveryInProgress(), replorigin_advance(), TransamVariables, TwoPhaseFilePath(), TwoPhaseState, WARNING, xl_xact_prepare::xid, XidFromFullTransactionId, and XLogRecPtrIsValid.

Referenced by restoreTwoPhaseData(), and xact_redo().

◆ PrepareRedoRemove()

void PrepareRedoRemove	(	TransactionId	xid,
		bool	giveWarning
	)

Definition at line 2664 of file twophase.c.

{
    FullTransactionId fxid =
        FullTransactionIdFromAllowableAt(TransamVariables->nextXid, xid);
 
    PrepareRedoRemoveFull(fxid, giveWarning);
}

References fb(), FullTransactionIdFromAllowableAt(), TransamVariablesData::nextXid, PrepareRedoRemoveFull(), and TransamVariables.

Referenced by xact_redo().

◆ PrepareRedoRemoveFull()

static void PrepareRedoRemoveFull	(	FullTransactionId	fxid,
		bool	giveWarning
	)

static

Definition at line 2620 of file twophase.c.

{
    GlobalTransaction gxact = NULL;
    int         i;
    bool        found = false;
 
    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
    Assert(RecoveryInProgress());
 
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        gxact = TwoPhaseState->prepXacts[i];
 
        if (FullTransactionIdEquals(gxact->fxid, fxid))
        {
            Assert(gxact->inredo);
            found = true;
            break;
        }
    }
 
    /*
     * Just leave if there is nothing, this is expected during WAL replay.
     */
    if (!found)
        return;
 
    /*
     * And now we can clean up any files we may have left.
     */
    elog(DEBUG2, "removing 2PC data for transaction %u of epoch %u ",
         XidFromFullTransactionId(fxid),
         EpochFromFullTransactionId(fxid));
 
    if (gxact->ondisk)
        RemoveTwoPhaseFile(fxid, giveWarning);
 
    RemoveGXact(gxact);
}

References Assert, DEBUG2, elog, EpochFromFullTransactionId, fb(), FullTransactionIdEquals, i, LW_EXCLUSIVE, LWLockHeldByMeInMode(), TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, RecoveryInProgress(), RemoveGXact(), RemoveTwoPhaseFile(), TwoPhaseState, and XidFromFullTransactionId.

Referenced by PrepareRedoRemove(), and ProcessTwoPhaseBuffer().

◆ PrescanPreparedTransactions()

TransactionId PrescanPreparedTransactions	(	TransactionId **	xids_p,
		int *	nxids_p
	)

Definition at line 1966 of file twophase.c.

{
    FullTransactionId nextXid = TransamVariables->nextXid;
    TransactionId origNextXid = XidFromFullTransactionId(nextXid);
    TransactionId result = origNextXid;
    TransactionId *xids = NULL;
    int         nxids = 0;
    int         allocsize = 0;
    int         i;
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        TransactionId xid;
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        Assert(gxact->inredo);
 
        buf = ProcessTwoPhaseBuffer(gxact->fxid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, false, true);
 
        if (buf == NULL)
            continue;
 
        /*
         * OK, we think this file is valid.  Incorporate xid into the
         * running-minimum result.
         */
        xid = XidFromFullTransactionId(gxact->fxid);
        if (TransactionIdPrecedes(xid, result))
            result = xid;
 
        if (xids_p)
        {
            if (nxids == allocsize)
            {
                if (nxids == 0)
                {
                    allocsize = 10;
                    xids = palloc(allocsize * sizeof(TransactionId));
                }
                else
                {
                    allocsize = allocsize * 2;
                    xids = repalloc(xids, allocsize * sizeof(TransactionId));
                }
            }
            xids[nxids++] = xid;
        }
 
        pfree(buf);
    }
    LWLockRelease(TwoPhaseStateLock);
 
    if (xids_p)
    {
        *xids_p = xids;
        *nxids_p = nxids;
    }
 
    return result;
}

References Assert, buf, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::nextXid, TwoPhaseStateData::numPrepXacts, palloc(), pfree(), TwoPhaseStateData::prepXacts, ProcessTwoPhaseBuffer(), repalloc(), TransactionIdPrecedes(), TransamVariables, TwoPhaseState, and XidFromFullTransactionId.

Referenced by StartupXLOG(), and xlog_redo().

◆ ProcessRecords()

static void ProcessRecords	(	char *	bufptr,
		FullTransactionId	fxid,
		const TwoPhaseCallback	callbacks[]
	)

static

Definition at line 1692 of file twophase.c.

{
    for (;;)
    {
        TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;
 
        Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
        if (record->rmid == TWOPHASE_RM_END_ID)
            break;
 
        bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));
 
        if (callbacks[record->rmid] != NULL)
            callbacks[record->rmid] (fxid, record->info, bufptr, record->len);
 
        bufptr += MAXALIGN(record->len);
    }
}

References Assert, fb(), TwoPhaseRecordOnDisk::info, TwoPhaseRecordOnDisk::len, MAXALIGN, TwoPhaseRecordOnDisk::rmid, TWOPHASE_RM_END_ID, and TWOPHASE_RM_MAX_ID.

Referenced by FinishPreparedTransaction(), and RecoverPreparedTransactions().

◆ ProcessTwoPhaseBuffer()

static char * ProcessTwoPhaseBuffer	(	FullTransactionId	fxid,
		XLogRecPtr	prepare_start_lsn,
		bool	fromdisk,
		bool	setParent,
		bool	setNextXid
	)

static

Definition at line 2187 of file twophase.c.

{
    FullTransactionId nextXid = TransamVariables->nextXid;
    TransactionId *subxids;
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int         i;
 
    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 
    if (!fromdisk)
        Assert(XLogRecPtrIsValid(prepare_start_lsn));
 
    /* Already processed? */
    if (TransactionIdDidCommit(XidFromFullTransactionId(fxid)) ||
        TransactionIdDidAbort(XidFromFullTransactionId(fxid)))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state file for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
            RemoveTwoPhaseFile(fxid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state from memory for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
            PrepareRedoRemoveFull(fxid, true);
        }
        return NULL;
    }
 
    /* Reject XID if too new */
    if (FullTransactionIdFollowsOrEquals(fxid, nextXid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state file for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
            RemoveTwoPhaseFile(fxid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state from memory for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
            PrepareRedoRemoveFull(fxid, true);
        }
        return NULL;
    }
 
    if (fromdisk)
    {
        /* Read and validate file */
        buf = ReadTwoPhaseFile(fxid, false);
    }
    else
    {
        /* Read xlog data */
        XlogReadTwoPhaseData(prepare_start_lsn, &buf, NULL);
    }
 
    /* Deconstruct header */
    hdr = (TwoPhaseFileHeader *) buf;
    if (!TransactionIdEquals(hdr->xid, XidFromFullTransactionId(fxid)))
    {
        if (fromdisk)
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("corrupted two-phase state file for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
        else
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("corrupted two-phase state in memory for transaction %u of epoch %u",
                            XidFromFullTransactionId(fxid),
                            EpochFromFullTransactionId(fxid))));
    }
 
    /*
     * Examine subtransaction XIDs ... they should all follow main XID, and
     * they may force us to advance nextXid.
     */
    subxids = (TransactionId *) (buf +
                                 MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                                 MAXALIGN(hdr->gidlen));
    for (i = 0; i < hdr->nsubxacts; i++)
    {
        TransactionId subxid = subxids[i];
 
        Assert(TransactionIdFollows(subxid, XidFromFullTransactionId(fxid)));
 
        /* update nextXid if needed */
        if (setNextXid)
            AdvanceNextFullTransactionIdPastXid(subxid);
 
        if (setParent)
            SubTransSetParent(subxid, XidFromFullTransactionId(fxid));
    }
 
    return buf;
}

References AdvanceNextFullTransactionIdPastXid(), Assert, buf, EpochFromFullTransactionId, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errmsg(), ERROR, fb(), FullTransactionIdFollowsOrEquals, xl_xact_prepare::gidlen, i, LW_EXCLUSIVE, LWLockHeldByMeInMode(), MAXALIGN, TransamVariablesData::nextXid, xl_xact_prepare::nsubxacts, PrepareRedoRemoveFull(), ReadTwoPhaseFile(), RemoveTwoPhaseFile(), SubTransSetParent(), TransactionIdDidAbort(), TransactionIdDidCommit(), TransactionIdEquals, TransactionIdFollows(), TransamVariables, WARNING, xl_xact_prepare::xid, XidFromFullTransactionId, XlogReadTwoPhaseData(), and XLogRecPtrIsValid.

Referenced by PrescanPreparedTransactions(), RecoverPreparedTransactions(), restoreTwoPhaseData(), and StandbyRecoverPreparedTransactions().

◆ ReadTwoPhaseFile()

static char * ReadTwoPhaseFile	(	FullTransactionId	fxid,
		bool	missing_ok
	)

static

Definition at line 1295 of file twophase.c.

{
    char        path[MAXPGPATH];
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int         fd;
    struct stat stat;
    uint32      crc_offset;
    pg_crc32c   calc_crc,
                file_crc;
    int         r;
 
    TwoPhaseFilePath(path, fxid);
 
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
    if (fd < 0)
    {
        if (missing_ok && errno == ENOENT)
            return NULL;
 
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
    }
 
    /*
     * Check file length.  We can determine a lower bound pretty easily. We
     * set an upper bound to avoid palloc() failure on a corrupt file, though
     * we can't guarantee that we won't get an out of memory error anyway,
     * even on a valid file.
     */
    if (fstat(fd, &stat))
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not stat file \"%s\": %m", path)));
 
    if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                        MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
                        sizeof(pg_crc32c)) ||
        stat.st_size > MaxAllocSize)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg_plural("incorrect size of file \"%s\": %lld byte",
                               "incorrect size of file \"%s\": %lld bytes",
                               (long long int) stat.st_size, path,
                               (long long int) stat.st_size)));
 
    crc_offset = stat.st_size - sizeof(pg_crc32c);
    if (crc_offset != MAXALIGN(crc_offset))
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("incorrect alignment of CRC offset for file \"%s\"",
                        path)));
 
    /*
     * OK, slurp in the file.
     */
    buf = (char *) palloc(stat.st_size);
 
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_READ);
    r = read(fd, buf, stat.st_size);
    if (r != stat.st_size)
    {
        if (r < 0)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        else
            ereport(ERROR,
                    (errmsg("could not read file \"%s\": read %d of %lld",
                            path, r, (long long int) stat.st_size)));
    }
 
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
 
    hdr = (TwoPhaseFileHeader *) buf;
    if (hdr->magic != TWOPHASE_MAGIC)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("invalid magic number stored in file \"%s\"",
                        path)));
 
    if (hdr->total_len != stat.st_size)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("invalid size stored in file \"%s\"",
                        path)));
 
    INIT_CRC32C(calc_crc);
    COMP_CRC32C(calc_crc, buf, crc_offset);
    FIN_CRC32C(calc_crc);
 
    file_crc = *((pg_crc32c *) (buf + crc_offset));
 
    if (!EQ_CRC32C(calc_crc, file_crc))
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("calculated CRC checksum does not match value stored in file \"%s\"",
                        path)));
 
    return buf;
}

References buf, CloseTransientFile(), COMP_CRC32C, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg(), errmsg_plural(), ERROR, fb(), fd(), FIN_CRC32C, fstat, INIT_CRC32C, xl_xact_prepare::magic, MAXALIGN, MaxAllocSize, MAXPGPATH, OpenTransientFile(), palloc(), PG_BINARY, pgstat_report_wait_end(), pgstat_report_wait_start(), read, stat::st_size, xl_xact_prepare::total_len, TWOPHASE_MAGIC, and TwoPhaseFilePath().

Referenced by FinishPreparedTransaction(), LookupGXact(), ProcessTwoPhaseBuffer(), and StandbyTransactionIdIsPrepared().

◆ RecordTransactionAbortPrepared()

static void RecordTransactionAbortPrepared	(	TransactionId	xid,
		int	nchildren,
		TransactionId *	children,
		int	nrels,
		RelFileLocator *	rels,
		int	nstats,
		xl_xact_stats_item *	stats,
		const char *	gid
	)

static

Definition at line 2432 of file twophase.c.

{
    XLogRecPtr  recptr;
    bool        replorigin;
 
    /*
     * Are we using the replication origins feature?  Or, in other words, are
     * we replaying remote actions?
     */
    replorigin = (replorigin_xact_state.origin != InvalidReplOriginId &&
                  replorigin_xact_state.origin != DoNotReplicateId);
 
    /*
     * Catch the scenario where we aborted partway through
     * RecordTransactionCommitPrepared ...
     */
    if (TransactionIdDidCommit(xid))
        elog(PANIC, "cannot abort transaction %u, it was already committed",
             xid);
 
    START_CRIT_SECTION();
 
    /*
     * Emit the XLOG commit record. Note that we mark 2PC aborts as
     * potentially having AccessExclusiveLocks since we don't know whether or
     * not they do.
     */
    recptr = XactLogAbortRecord(GetCurrentTimestamp(),
                                nchildren, children,
                                nrels, rels,
                                nstats, stats,
                                MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                xid, gid);
 
    if (replorigin)
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_xact_state.origin_lsn,
                                   XactLastRecEnd);
 
    /* Always flush, since we're about to remove the 2PC state file */
    XLogFlush(recptr);
 
    /*
     * Mark the transaction aborted in clog.  This is not absolutely necessary
     * but we may as well do it while we are here.
     */
    TransactionIdAbortTree(xid, nchildren, children);
 
    END_CRIT_SECTION();
 
    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked clog, but still show as running
     * in the procarray and continue to hold locks.
     */
    SyncRepWaitForLSN(recptr, false);
}

References DoNotReplicateId, elog, END_CRIT_SECTION, fb(), GetCurrentTimestamp(), InvalidReplOriginId, MyXactFlags, ReplOriginXactState::origin, ReplOriginXactState::origin_lsn, PANIC, replorigin_session_advance(), replorigin_xact_state, START_CRIT_SECTION, SyncRepWaitForLSN(), TransactionIdAbortTree(), TransactionIdDidCommit(), XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK, XactLastRecEnd, XactLogAbortRecord(), and XLogFlush().

Referenced by FinishPreparedTransaction().

◆ RecordTransactionCommitPrepared()

static void RecordTransactionCommitPrepared	(	TransactionId	xid,
		int	nchildren,
		TransactionId *	children,
		int	nrels,
		RelFileLocator *	rels,
		int	nstats,
		xl_xact_stats_item *	stats,
		int	ninvalmsgs,
		SharedInvalidationMessage *	invalmsgs,
		bool	initfileinval,
		const char *	gid
	)

static

Definition at line 2313 of file twophase.c.

{
    XLogRecPtr  recptr;
    TimestampTz committs;
    bool        replorigin;
 
    /*
     * Are we using the replication origins feature?  Or, in other words, are
     * we replaying remote actions?
     */
    replorigin = (replorigin_xact_state.origin != InvalidReplOriginId &&
                  replorigin_xact_state.origin != DoNotReplicateId);
 
    /* Load the injection point before entering the critical section */
    INJECTION_POINT_LOAD("commit-after-delay-checkpoint");
 
    START_CRIT_SECTION();
 
    /* See notes in RecordTransactionCommit */
    Assert((MyProc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) == 0);
    MyProc->delayChkptFlags |= DELAY_CHKPT_IN_COMMIT;
 
    INJECTION_POINT_CACHED("commit-after-delay-checkpoint", NULL);
 
    /*
     * Ensures the DELAY_CHKPT_IN_COMMIT flag write is globally visible before
     * commit time is written.
     */
    pg_write_barrier();
 
    /*
     * Note it is important to set committs value after marking ourselves as
     * in the commit critical section (DELAY_CHKPT_IN_COMMIT). This is because
     * we want to ensure all transactions that have acquired commit timestamp
     * are finished before we allow the logical replication client to advance
     * its xid which is used to hold back dead rows for conflict detection.
     * See comments atop worker.c.
     */
    committs = GetCurrentTimestamp();
 
    /*
     * Emit the XLOG commit record. Note that we mark 2PC commits as
     * potentially having AccessExclusiveLocks since we don't know whether or
     * not they do.
     */
    recptr = XactLogCommitRecord(committs,
                                 nchildren, children, nrels, rels,
                                 nstats, stats,
                                 ninvalmsgs, invalmsgs,
                                 initfileinval,
                                 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                 xid, gid);
 
 
    if (replorigin)
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_xact_state.origin_lsn,
                                   XactLastRecEnd);
 
    /*
     * Record commit timestamp.  The value comes from plain commit timestamp
     * if replorigin is not enabled, or replorigin already set a value for us
     * in replorigin_xact_state.origin_timestamp otherwise.
     *
     * We don't need to WAL-log anything here, as the commit record written
     * above already contains the data.
     */
    if (!replorigin || replorigin_xact_state.origin_timestamp == 0)
        replorigin_xact_state.origin_timestamp = committs;
 
    TransactionTreeSetCommitTsData(xid, nchildren, children,
                                   replorigin_xact_state.origin_timestamp,
                                   replorigin_xact_state.origin);
 
    /*
     * We don't currently try to sleep before flush here ... nor is there any
     * support for async commit of a prepared xact (the very idea is probably
     * a contradiction)
     */
 
    /* Flush XLOG to disk */
    XLogFlush(recptr);
 
    /* Mark the transaction committed in pg_xact */
    TransactionIdCommitTree(xid, nchildren, children);
 
    /* Checkpoint can proceed now */
    MyProc->delayChkptFlags &= ~DELAY_CHKPT_IN_COMMIT;
 
    END_CRIT_SECTION();
 
    /*
     * Wait for synchronous replication, if required.
     *
     * Note that at this stage we have marked clog, but still show as running
     * in the procarray and continue to hold locks.
     */
    SyncRepWaitForLSN(recptr, true);
}

References Assert, DELAY_CHKPT_IN_COMMIT, PGPROC::delayChkptFlags, DoNotReplicateId, END_CRIT_SECTION, fb(), GetCurrentTimestamp(), INJECTION_POINT_CACHED, INJECTION_POINT_LOAD, InvalidReplOriginId, MyProc, MyXactFlags, ReplOriginXactState::origin, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, pg_write_barrier, replorigin_session_advance(), replorigin_xact_state, START_CRIT_SECTION, SyncRepWaitForLSN(), TransactionIdCommitTree(), TransactionTreeSetCommitTsData(), XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK, XactLastRecEnd, XactLogCommitRecord(), and XLogFlush().

Referenced by FinishPreparedTransaction().

◆ RecoverPreparedTransactions()

void RecoverPreparedTransactions ( void )

Definition at line 2083 of file twophase.c.

{
    int         i;
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        FullTransactionId fxid = gxact->fxid;
        char       *bufptr;
        TwoPhaseFileHeader *hdr;
        TransactionId *subxids;
        const char *gid;
 
        /*
         * Reconstruct subtrans state for the transaction --- needed because
         * pg_subtrans is not preserved over a restart.  Note that we are
         * linking all the subtransactions directly to the top-level XID;
         * there may originally have been a more complex hierarchy, but
         * there's no need to restore that exactly. It's possible that
         * SubTransSetParent has been set before, if the prepared transaction
         * generated xid assignment records.
         */
        buf = ProcessTwoPhaseBuffer(gxact->fxid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, true, false);
        if (buf == NULL)
            continue;
 
        ereport(LOG,
                (errmsg("recovering prepared transaction %u of epoch %u from shared memory",
                        XidFromFullTransactionId(gxact->fxid),
                        EpochFromFullTransactionId(gxact->fxid))));
 
        hdr = (TwoPhaseFileHeader *) buf;
        Assert(TransactionIdEquals(hdr->xid,
                                   XidFromFullTransactionId(gxact->fxid)));
        bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
        gid = (const char *) bufptr;
        bufptr += MAXALIGN(hdr->gidlen);
        subxids = (TransactionId *) bufptr;
        bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
        bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileLocator));
        bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileLocator));
        bufptr += MAXALIGN(hdr->ncommitstats * sizeof(xl_xact_stats_item));
        bufptr += MAXALIGN(hdr->nabortstats * sizeof(xl_xact_stats_item));
        bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
 
        /*
         * Recreate its GXACT and dummy PGPROC. But, check whether it was
         * added in redo and already has a shmem entry for it.
         */
        MarkAsPreparingGuts(gxact, gxact->fxid, gid,
                            hdr->prepared_at,
                            hdr->owner, hdr->database);
 
        /* recovered, so reset the flag for entries generated by redo */
        gxact->inredo = false;
 
        GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
        MarkAsPrepared(gxact, true);
 
        LWLockRelease(TwoPhaseStateLock);
 
        /*
         * Recover other state (notably locks) using resource managers.
         */
        ProcessRecords(bufptr, fxid, twophase_recover_callbacks);
 
        /*
         * Release locks held by the standby process after we process each
         * prepared transaction. As a result, we don't need too many
         * additional locks at any one time.
         */
        if (InHotStandby)
            StandbyReleaseLockTree(hdr->xid, hdr->nsubxacts, subxids);
 
        /*
         * We're done with recovering this transaction. Clear MyLockedGxact,
         * like we do in PrepareTransaction() during normal operation.
         */
        PostPrepare_Twophase();
 
        pfree(buf);
 
        LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    }
 
    LWLockRelease(TwoPhaseStateLock);
}

References Assert, buf, xl_xact_prepare::database, EpochFromFullTransactionId, ereport, errmsg(), fb(), xl_xact_prepare::gidlen, GXactLoadSubxactData(), i, InHotStandby, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MarkAsPrepared(), MarkAsPreparingGuts(), MAXALIGN, xl_xact_prepare::nabortrels, xl_xact_prepare::nabortstats, xl_xact_prepare::ncommitrels, xl_xact_prepare::ncommitstats, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, TwoPhaseStateData::numPrepXacts, xl_xact_prepare::owner, pfree(), PostPrepare_Twophase(), xl_xact_prepare::prepared_at, TwoPhaseStateData::prepXacts, ProcessRecords(), ProcessTwoPhaseBuffer(), StandbyReleaseLockTree(), TransactionIdEquals, twophase_recover_callbacks, TwoPhaseState, xl_xact_prepare::xid, and XidFromFullTransactionId.

Referenced by StartupXLOG().

◆ RecreateTwoPhaseFile()

static void RecreateTwoPhaseFile	(	FullTransactionId	fxid,
		void *	content,
		int	len
	)

static

Definition at line 1742 of file twophase.c.

{
    char        path[MAXPGPATH];
    pg_crc32c   statefile_crc;
    int         fd;
 
    /* Recompute CRC */
    INIT_CRC32C(statefile_crc);
    COMP_CRC32C(statefile_crc, content, len);
    FIN_CRC32C(statefile_crc);
 
    TwoPhaseFilePath(path, fxid);
 
    fd = OpenTransientFile(path,
                           O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not recreate file \"%s\": %m", path)));
 
    /* Write content and CRC */
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_WRITE);
    if (write(fd, content, len) != len)
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write file \"%s\": %m", path)));
    }
    if (write(fd, &statefile_crc, sizeof(pg_crc32c)) != sizeof(pg_crc32c))
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write file \"%s\": %m", path)));
    }
    pgstat_report_wait_end();
 
    /*
     * We must fsync the file because the end-of-replay checkpoint will not do
     * so, there being no GXACT in shared memory yet to tell it to.
     */
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_SYNC);
    if (pg_fsync(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m", path)));
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
}

References CloseTransientFile(), COMP_CRC32C, ereport, errcode_for_file_access(), errmsg(), ERROR, fb(), fd(), FIN_CRC32C, INIT_CRC32C, len, MAXPGPATH, OpenTransientFile(), PG_BINARY, pg_fsync(), pgstat_report_wait_end(), pgstat_report_wait_start(), TwoPhaseFilePath(), and write.

Referenced by CheckPointTwoPhase().

◆ RegisterTwoPhaseRecord()

void RegisterTwoPhaseRecord	(	TwoPhaseRmgrId	rmid,
		uint16	info,
		const void *	data,
		uint32	len
	)

Definition at line 1271 of file twophase.c.

{
    TwoPhaseRecordOnDisk record;
 
    record.rmid = rmid;
    record.info = info;
    record.len = len;
    save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
    if (len > 0)
        save_state_data(data, len);
}

References data, TwoPhaseRecordOnDisk::info, TwoPhaseRecordOnDisk::len, len, TwoPhaseRecordOnDisk::rmid, and save_state_data().

Referenced by AtPrepare_Locks(), AtPrepare_MultiXact(), AtPrepare_PgStat_Relations(), AtPrepare_PredicateLocks(), and EndPrepare().

◆ RemoveGXact()

static void RemoveGXact ( GlobalTransaction gxact )

static

Definition at line 632 of file twophase.c.

{
    int         i;
 
    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        if (gxact == TwoPhaseState->prepXacts[i])
        {
            /* remove from the active array */
            TwoPhaseState->numPrepXacts--;
            TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
 
            /* and put it back in the freelist */
            gxact->next = TwoPhaseState->freeGXacts;
            TwoPhaseState->freeGXacts = gxact;
 
            return;
        }
    }
 
    elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
}

References Assert, elog, ERROR, fb(), TwoPhaseStateData::freeGXacts, i, LW_EXCLUSIVE, LWLockHeldByMeInMode(), GlobalTransactionData::next, TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, and TwoPhaseState.

Referenced by AtAbort_Twophase(), FinishPreparedTransaction(), and PrepareRedoRemoveFull().

◆ RemoveTwoPhaseFile()

static void RemoveTwoPhaseFile	(	FullTransactionId	fxid,
		bool	giveWarning
	)

static

Definition at line 1723 of file twophase.c.

{
    char        path[MAXPGPATH];
 
    TwoPhaseFilePath(path, fxid);
    if (unlink(path))
        if (errno != ENOENT || giveWarning)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not remove file \"%s\": %m", path)));
}

References ereport, errcode_for_file_access(), errmsg(), fb(), MAXPGPATH, TwoPhaseFilePath(), and WARNING.

Referenced by FinishPreparedTransaction(), PrepareRedoRemoveFull(), and ProcessTwoPhaseBuffer().

◆ restoreTwoPhaseData()

void restoreTwoPhaseData ( void )

Definition at line 1904 of file twophase.c.

{
    DIR        *cldir;
    struct dirent *clde;
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    cldir = AllocateDir(TWOPHASE_DIR);
    while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
    {
        if (strlen(clde->d_name) == 16 &&
            strspn(clde->d_name, "0123456789ABCDEF") == 16)
        {
            FullTransactionId fxid;
            char       *buf;
 
            fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16));
 
            buf = ProcessTwoPhaseBuffer(fxid, InvalidXLogRecPtr,
                                        true, false, false);
            if (buf == NULL)
                continue;
 
            PrepareRedoAdd(fxid, buf, InvalidXLogRecPtr,
                           InvalidXLogRecPtr, InvalidReplOriginId);
        }
    }
    LWLockRelease(TwoPhaseStateLock);
    FreeDir(cldir);
}

References AllocateDir(), buf, fb(), FreeDir(), FullTransactionIdFromU64(), InvalidReplOriginId, InvalidXLogRecPtr, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), PrepareRedoAdd(), ProcessTwoPhaseBuffer(), ReadDir(), and TWOPHASE_DIR.

Referenced by StartupXLOG().

◆ save_state_data()

static void save_state_data	(	const void *	data,
		uint32	len
	)

static

Definition at line 1024 of file twophase.c.

{
    uint32      padlen = MAXALIGN(len);
 
    if (padlen > records.bytes_free)
    {
        records.tail->next = palloc0_object(StateFileChunk);
        records.tail = records.tail->next;
        records.tail->len = 0;
        records.tail->next = NULL;
        records.num_chunks++;
 
        records.bytes_free = Max(padlen, 512);
        records.tail->data = palloc(records.bytes_free);
    }
 
    memcpy(records.tail->data + records.tail->len, data, len);
    records.tail->len += padlen;
    records.bytes_free -= padlen;
    records.total_len += padlen;
}

References xllist::bytes_free, StateFileChunk::data, data, fb(), StateFileChunk::len, len, Max, MAXALIGN, StateFileChunk::next, xllist::num_chunks, palloc(), palloc0_object, records, xllist::tail, and xllist::total_len.

Referenced by RegisterTwoPhaseRecord(), and StartPrepare().

◆ StandbyRecoverPreparedTransactions()

void StandbyRecoverPreparedTransactions ( void )

Definition at line 2045 of file twophase.c.

{
    int         i;
 
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        Assert(gxact->inredo);
 
        buf = ProcessTwoPhaseBuffer(gxact->fxid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, true, false);
        if (buf != NULL)
            pfree(buf);
    }
    LWLockRelease(TwoPhaseStateLock);
}

References Assert, buf, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, pfree(), TwoPhaseStateData::prepXacts, ProcessTwoPhaseBuffer(), and TwoPhaseState.

Referenced by StartupXLOG(), and xlog_redo().

◆ StandbyTransactionIdIsPrepared()

bool StandbyTransactionIdIsPrepared ( TransactionId xid )

Definition at line 1467 of file twophase.c.

{
    char       *buf;
    TwoPhaseFileHeader *hdr;
    bool        result;
    FullTransactionId fxid;
 
    Assert(TransactionIdIsValid(xid));
 
    if (max_prepared_xacts <= 0)
        return false;           /* nothing to do */
 
    /* Read and validate file */
    fxid = AdjustToFullTransactionId(xid);
    buf = ReadTwoPhaseFile(fxid, true);
    if (buf == NULL)
        return false;
 
    /* Check header also */
    hdr = (TwoPhaseFileHeader *) buf;
    result = TransactionIdEquals(hdr->xid, xid);
    pfree(buf);
 
    return result;
}

References AdjustToFullTransactionId(), Assert, buf, fb(), max_prepared_xacts, pfree(), ReadTwoPhaseFile(), TransactionIdEquals, TransactionIdIsValid, and xl_xact_prepare::xid.

Referenced by KnownAssignedXidsRemovePreceding(), and StandbyReleaseOldLocks().

◆ StartPrepare()

void StartPrepare ( GlobalTransaction gxact )

Definition at line 1052 of file twophase.c.

{
    PGPROC     *proc = GetPGProcByNumber(gxact->pgprocno);
    TransactionId xid = XidFromFullTransactionId(gxact->fxid);
    TwoPhaseFileHeader hdr;
    TransactionId *children;
    RelFileLocator *commitrels;
    RelFileLocator *abortrels;
    xl_xact_stats_item *abortstats = NULL;
    xl_xact_stats_item *commitstats = NULL;
    SharedInvalidationMessage *invalmsgs;
 
    /* Initialize linked list */
    records.head = palloc0_object(StateFileChunk);
    records.head->len = 0;
    records.head->next = NULL;
 
    records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
    records.head->data = palloc(records.bytes_free);
 
    records.tail = records.head;
    records.num_chunks = 1;
 
    records.total_len = 0;
 
    /* Create header */
    hdr.magic = TWOPHASE_MAGIC;
    hdr.total_len = 0;          /* EndPrepare will fill this in */
    hdr.xid = xid;
    hdr.database = proc->databaseId;
    hdr.prepared_at = gxact->prepared_at;
    hdr.owner = gxact->owner;
    hdr.nsubxacts = xactGetCommittedChildren(&children);
    hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels);
    hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
    hdr.ncommitstats =
        pgstat_get_transactional_drops(true, &commitstats);
    hdr.nabortstats =
        pgstat_get_transactional_drops(false, &abortstats);
    hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
                                                          &hdr.initfileinval);
    hdr.gidlen = strlen(gxact->gid) + 1;    /* Include '\0' */
    /* EndPrepare will fill the origin data, if necessary */
    hdr.origin_lsn = InvalidXLogRecPtr;
    hdr.origin_timestamp = 0;
 
    save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
    save_state_data(gxact->gid, hdr.gidlen);
 
    /*
     * Add the additional info about subxacts, deletable files and cache
     * invalidation messages.
     */
    if (hdr.nsubxacts > 0)
    {
        save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
        /* While we have the child-xact data, stuff it in the gxact too */
        GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
    }
    if (hdr.ncommitrels > 0)
    {
        save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileLocator));
        pfree(commitrels);
    }
    if (hdr.nabortrels > 0)
    {
        save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileLocator));
        pfree(abortrels);
    }
    if (hdr.ncommitstats > 0)
    {
        save_state_data(commitstats,
                        hdr.ncommitstats * sizeof(xl_xact_stats_item));
        pfree(commitstats);
    }
    if (hdr.nabortstats > 0)
    {
        save_state_data(abortstats,
                        hdr.nabortstats * sizeof(xl_xact_stats_item));
        pfree(abortstats);
    }
    if (hdr.ninvalmsgs > 0)
    {
        save_state_data(invalmsgs,
                        hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
        pfree(invalmsgs);
    }
}

References xllist::bytes_free, StateFileChunk::data, xl_xact_prepare::database, PGPROC::databaseId, fb(), GetPGProcByNumber, xl_xact_prepare::gidlen, GXactLoadSubxactData(), xllist::head, xl_xact_prepare::initfileinval, InvalidXLogRecPtr, StateFileChunk::len, xl_xact_prepare::magic, Max, xl_xact_prepare::nabortrels, xl_xact_prepare::nabortstats, xl_xact_prepare::ncommitrels, xl_xact_prepare::ncommitstats, StateFileChunk::next, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, xllist::num_chunks, xl_xact_prepare::origin_lsn, xl_xact_prepare::origin_timestamp, xl_xact_prepare::owner, palloc(), palloc0_object, pfree(), pgstat_get_transactional_drops(), xl_xact_prepare::prepared_at, records, save_state_data(), smgrGetPendingDeletes(), xllist::tail, xllist::total_len, xl_xact_prepare::total_len, TWOPHASE_MAGIC, xactGetCommittedChildren(), xactGetCommittedInvalidationMessages(), xl_xact_prepare::xid, and XidFromFullTransactionId.

Referenced by PrepareTransaction().

◆ TwoPhaseFilePath()

static int TwoPhaseFilePath	(	char *	path,
		FullTransactionId	fxid
	)

inlinestatic

Definition at line 950 of file twophase.c.

{
    return snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X%08X",
                    EpochFromFullTransactionId(fxid),
                    XidFromFullTransactionId(fxid));
}

References EpochFromFullTransactionId, MAXPGPATH, snprintf, TWOPHASE_DIR, and XidFromFullTransactionId.

Referenced by PrepareRedoAdd(), ReadTwoPhaseFile(), RecreateTwoPhaseFile(), and RemoveTwoPhaseFile().

◆ TwoPhaseGetDummyProc()

PGPROC * TwoPhaseGetDummyProc	(	FullTransactionId	fxid,
		bool	lock_held
	)

Definition at line 923 of file twophase.c.

{
    GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
 
    return GetPGProcByNumber(gxact->pgprocno);
}

References fb(), GetPGProcByNumber, and TwoPhaseGetGXact().

Referenced by lock_twophase_postcommit(), lock_twophase_recover(), and PostPrepare_Locks().

◆ TwoPhaseGetDummyProcNumber()

ProcNumber TwoPhaseGetDummyProcNumber	(	FullTransactionId	fxid,
		bool	lock_held
	)

Definition at line 908 of file twophase.c.

{
    GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
 
    return gxact->pgprocno;
}

References fb(), and TwoPhaseGetGXact().

Referenced by multixact_twophase_postcommit(), multixact_twophase_recover(), and PostPrepare_MultiXact().

◆ TwoPhaseGetGXact()

static GlobalTransaction TwoPhaseGetGXact	(	FullTransactionId	fxid,
		bool	lock_held
	)

static

Definition at line 804 of file twophase.c.

{
    GlobalTransaction result = NULL;
    int         i;
 
    static FullTransactionId cached_fxid = {InvalidTransactionId};
    static GlobalTransaction cached_gxact = NULL;
 
    Assert(!lock_held || LWLockHeldByMe(TwoPhaseStateLock));
 
    /*
     * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
     * repeatedly for the same XID.  We can save work with a simple cache.
     */
    if (FullTransactionIdEquals(fxid, cached_fxid))
        return cached_gxact;
 
    if (!lock_held)
        LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
 
        if (FullTransactionIdEquals(gxact->fxid, fxid))
        {
            result = gxact;
            break;
        }
    }
 
    if (!lock_held)
        LWLockRelease(TwoPhaseStateLock);
 
    if (result == NULL)         /* should not happen */
        elog(ERROR, "failed to find GlobalTransaction for xid %u",
             XidFromFullTransactionId(fxid));
 
    cached_fxid = fxid;
    cached_gxact = result;
 
    return result;
}

References Assert, elog, ERROR, fb(), FullTransactionIdEquals, i, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, TwoPhaseState, and XidFromFullTransactionId.

Referenced by TwoPhaseGetDummyProc(), and TwoPhaseGetDummyProcNumber().

◆ TwoPhaseGetOldestXidInCommit()

TransactionId TwoPhaseGetOldestXidInCommit ( void )

Definition at line 2829 of file twophase.c.

{
    TransactionId oldestRunningXid = InvalidTransactionId;
 
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 
    for (int i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGPROC     *commitproc;
        TransactionId xid;
 
        if (!gxact->valid)
            continue;
 
        if (gxact->locking_backend == INVALID_PROC_NUMBER)
            continue;
 
        /*
         * Get the backend that is handling the transaction. It's safe to
         * access this backend while holding TwoPhaseStateLock, as the backend
         * can only be destroyed after either removing or unlocking the
         * current global transaction, both of which require an exclusive
         * TwoPhaseStateLock.
         */
        commitproc = GetPGProcByNumber(gxact->locking_backend);
 
        if (MyDatabaseId != commitproc->databaseId)
            continue;
 
        if ((commitproc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) == 0)
            continue;
 
        xid = XidFromFullTransactionId(gxact->fxid);
 
        if (!TransactionIdIsValid(oldestRunningXid) ||
            TransactionIdPrecedes(xid, oldestRunningXid))
            oldestRunningXid = xid;
    }
 
    LWLockRelease(TwoPhaseStateLock);
 
    return oldestRunningXid;
}

References DELAY_CHKPT_IN_COMMIT, fb(), GetPGProcByNumber, i, INVALID_PROC_NUMBER, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyDatabaseId, TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, TransactionIdIsValid, TransactionIdPrecedes(), TwoPhaseState, and XidFromFullTransactionId.

Referenced by ProcessStandbyPSRequestMessage().

◆ TwoPhaseGetXidByVirtualXID()

TransactionId TwoPhaseGetXidByVirtualXID	(	VirtualTransactionId	vxid,
		bool *	have_more
	)

Definition at line 857 of file twophase.c.

{
    int         i;
    TransactionId result = InvalidTransactionId;
 
    Assert(VirtualTransactionIdIsValid(vxid));
    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGPROC     *proc;
        VirtualTransactionId proc_vxid;
 
        if (!gxact->valid)
            continue;
        proc = GetPGProcByNumber(gxact->pgprocno);
        GET_VXID_FROM_PGPROC(proc_vxid, *proc);
        if (VirtualTransactionIdEquals(vxid, proc_vxid))
        {
            /*
             * Startup process sets proc->vxid.procNumber to
             * INVALID_PROC_NUMBER.
             */
            Assert(!gxact->inredo);
 
            if (result != InvalidTransactionId)
            {
                *have_more = true;
                break;
            }
            result = XidFromFullTransactionId(gxact->fxid);
        }
    }
 
    LWLockRelease(TwoPhaseStateLock);
 
    return result;
}

References Assert, fb(), GET_VXID_FROM_PGPROC, GetPGProcByNumber, i, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, TwoPhaseState, VirtualTransactionIdEquals, VirtualTransactionIdIsValid, and XidFromFullTransactionId.

Referenced by XactLockForVirtualXact().

◆ TwoPhaseShmemInit()

void TwoPhaseShmemInit ( void )

Definition at line 255 of file twophase.c.

{
    bool        found;
 
    TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
                                    TwoPhaseShmemSize(),
                                    &found);
    if (!IsUnderPostmaster)
    {
        GlobalTransaction gxacts;
        int         i;
 
        Assert(!found);
        TwoPhaseState->freeGXacts = NULL;
        TwoPhaseState->numPrepXacts = 0;
 
        /*
         * Initialize the linked list of free GlobalTransactionData structs
         */
        gxacts = (GlobalTransaction)
            ((char *) TwoPhaseState +
             MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
                      sizeof(GlobalTransaction) * max_prepared_xacts));
        for (i = 0; i < max_prepared_xacts; i++)
        {
            /* insert into linked list */
            gxacts[i].next = TwoPhaseState->freeGXacts;
            TwoPhaseState->freeGXacts = &gxacts[i];
 
            /* associate it with a PGPROC assigned by InitProcGlobal */
            gxacts[i].pgprocno = GetNumberFromPGProc(&PreparedXactProcs[i]);
        }
    }
    else
        Assert(found);
}

References Assert, fb(), TwoPhaseStateData::freeGXacts, GetNumberFromPGProc, i, IsUnderPostmaster, max_prepared_xacts, MAXALIGN, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::pgprocno, PreparedXactProcs, ShmemInitStruct(), TwoPhaseShmemSize(), and TwoPhaseState.

Referenced by CreateOrAttachShmemStructs().

◆ TwoPhaseShmemSize()

Size TwoPhaseShmemSize ( void )

Definition at line 239 of file twophase.c.

{
    Size        size;
 
    /* Need the fixed struct, the array of pointers, and the GTD structs */
    size = offsetof(TwoPhaseStateData, prepXacts);
    size = add_size(size, mul_size(max_prepared_xacts,
                                   sizeof(GlobalTransaction)));
    size = MAXALIGN(size);
    size = add_size(size, mul_size(max_prepared_xacts,
                                   sizeof(GlobalTransactionData)));
 
    return size;
}

References add_size(), fb(), max_prepared_xacts, MAXALIGN, and mul_size().

Referenced by CalculateShmemSize(), and TwoPhaseShmemInit().

◆ TwoPhaseTransactionGid()

void TwoPhaseTransactionGid	(	Oid	subid,
		TransactionId	xid,
		char *	gid_res,
		int	szgid
	)

Definition at line 2747 of file twophase.c.

{
    Assert(OidIsValid(subid));
 
    if (!TransactionIdIsValid(xid))
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg_internal("invalid two-phase transaction ID")));
 
    snprintf(gid_res, szgid, "pg_gid_%u_%u", subid, xid);
}

References Assert, ereport, errcode(), errmsg_internal(), ERROR, fb(), OidIsValid, snprintf, and TransactionIdIsValid.

Referenced by apply_handle_commit_prepared(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), and IsTwoPhaseTransactionGidForSubid().

◆ XlogReadTwoPhaseData()

static void XlogReadTwoPhaseData	(	XLogRecPtr	lsn,
		char **	buf,
		int *	len
	)

static

Definition at line 1412 of file twophase.c.

{
    XLogRecord *record;
    XLogReaderState *xlogreader;
    char       *errormsg;
 
    xlogreader = XLogReaderAllocate(wal_segment_size, NULL,
                                    XL_ROUTINE(.page_read = &read_local_xlog_page,
                                               .segment_open = &wal_segment_open,
                                               .segment_close = &wal_segment_close),
                                    NULL);
    if (!xlogreader)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("out of memory"),
                 errdetail("Failed while allocating a WAL reading processor.")));
 
    XLogBeginRead(xlogreader, lsn);
    record = XLogReadRecord(xlogreader, &errormsg);
 
    if (record == NULL)
    {
        if (errormsg)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read two-phase state from WAL at %X/%08X: %s",
                            LSN_FORMAT_ARGS(lsn), errormsg)));
        else
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("could not read two-phase state from WAL at %X/%08X",
                            LSN_FORMAT_ARGS(lsn))));
    }
 
    if (XLogRecGetRmid(xlogreader) != RM_XACT_ID ||
        (XLogRecGetInfo(xlogreader) & XLOG_XACT_OPMASK) != XLOG_XACT_PREPARE)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("expected two-phase state data is not present in WAL at %X/%08X",
                        LSN_FORMAT_ARGS(lsn))));
 
    if (len != NULL)
        *len = XLogRecGetDataLen(xlogreader);
 
    *buf = palloc_array(char, XLogRecGetDataLen(xlogreader));
    memcpy(*buf, XLogRecGetData(xlogreader), sizeof(char) * XLogRecGetDataLen(xlogreader));
 
    XLogReaderFree(xlogreader);
}

References buf, ereport, errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, fb(), len, LSN_FORMAT_ARGS, palloc_array, read_local_xlog_page(), wal_segment_close(), wal_segment_open(), wal_segment_size, XL_ROUTINE, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogBeginRead(), xlogreader, XLogReaderAllocate(), XLogReaderFree(), XLogReadRecord(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetInfo, and XLogRecGetRmid.

Referenced by CheckPointTwoPhase(), FinishPreparedTransaction(), LookupGXact(), and ProcessTwoPhaseBuffer().

Variable Documentation

◆ max_prepared_xacts

int max_prepared_xacts = 0

◆ MyLockedGxact

GlobalTransaction MyLockedGxact = NULL

static

Definition at line 197 of file twophase.c.

Referenced by AtAbort_Twophase(), EndPrepare(), FinishPreparedTransaction(), LockGXact(), MarkAsPreparingGuts(), and PostPrepare_Twophase().

◆ records

struct xllist records

static

Referenced by EndPrepare(), save_state_data(), and StartPrepare().

◆ twophaseExitRegistered

bool twophaseExitRegistered = false

static

Definition at line 199 of file twophase.c.

Referenced by LockGXact(), and MarkAsPreparing().

◆ TwoPhaseState

TwoPhaseStateData* TwoPhaseState

static

Definition at line 189 of file twophase.c.

Referenced by CheckPointTwoPhase(), GetPreparedTransactionList(), LockGXact(), LookupGXact(), LookupGXactBySubid(), MarkAsPreparing(), PrepareRedoAdd(), PrepareRedoRemoveFull(), PrescanPreparedTransactions(), RecoverPreparedTransactions(), RemoveGXact(), StandbyRecoverPreparedTransactions(), TwoPhaseGetGXact(), TwoPhaseGetOldestXidInCommit(), TwoPhaseGetXidByVirtualXID(), and TwoPhaseShmemInit().

Typedefs
typedef struct GlobalTransactionData	GlobalTransactionData

typedef struct TwoPhaseStateData	TwoPhaseStateData

typedef xl_xact_prepare	TwoPhaseFileHeader

typedef struct TwoPhaseRecordOnDisk	TwoPhaseRecordOnDisk

typedef struct StateFileChunk	StateFileChunk

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ TWOPHASE_DIR

◆ TWOPHASE_MAGIC

Typedef Documentation

◆ GlobalTransactionData

◆ StateFileChunk

◆ TwoPhaseFileHeader

◆ TwoPhaseRecordOnDisk

◆ TwoPhaseStateData

Function Documentation

◆ AdjustToFullTransactionId()

◆ AtAbort_Twophase()

◆ AtProcExit_Twophase()

◆ CheckPointTwoPhase()

◆ EndPrepare()

◆ FinishPreparedTransaction()

◆ GetPreparedTransactionList()

◆ GXactLoadSubxactData()

◆ IsTwoPhaseTransactionGidForSubid()

◆ LockGXact()

◆ LookupGXact()

◆ LookupGXactBySubid()

◆ MarkAsPrepared()

◆ MarkAsPreparing()

◆ MarkAsPreparingGuts()

◆ pg_prepared_xact()

◆ PostPrepare_Twophase()

◆ PrepareRedoAdd()

◆ PrepareRedoRemove()

◆ PrepareRedoRemoveFull()

◆ PrescanPreparedTransactions()

◆ ProcessRecords()

◆ ProcessTwoPhaseBuffer()

◆ ReadTwoPhaseFile()

◆ RecordTransactionAbortPrepared()

◆ RecordTransactionCommitPrepared()

◆ RecoverPreparedTransactions()

◆ RecreateTwoPhaseFile()

◆ RegisterTwoPhaseRecord()

◆ RemoveGXact()

◆ RemoveTwoPhaseFile()

◆ restoreTwoPhaseData()

◆ save_state_data()

◆ StandbyRecoverPreparedTransactions()

◆ StandbyTransactionIdIsPrepared()

◆ StartPrepare()

◆ TwoPhaseFilePath()

◆ TwoPhaseGetDummyProc()

◆ TwoPhaseGetDummyProcNumber()

◆ TwoPhaseGetGXact()

◆ TwoPhaseGetOldestXidInCommit()

◆ TwoPhaseGetXidByVirtualXID()

◆ TwoPhaseShmemInit()

◆ TwoPhaseShmemSize()

◆ TwoPhaseTransactionGid()

◆ XlogReadTwoPhaseData()

Variable Documentation

◆ max_prepared_xacts

◆ MyLockedGxact

◆ records

◆ twophaseExitRegistered

◆ TwoPhaseState