twophase.h File Reference

#include "access/xact.h"
#include "access/xlogdefs.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"

Include dependency graph for twophase.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Typedefs
typedef struct GlobalTransactionData *	GlobalTransaction

Functions
Size	TwoPhaseShmemSize (void)
void	TwoPhaseShmemInit (void)
void	AtAbort_Twophase (void)
void	PostPrepare_Twophase (void)
TransactionId	TwoPhaseGetXidByVirtualXID (VirtualTransactionId vxid, bool *have_more)
PGPROC *	TwoPhaseGetDummyProc (FullTransactionId fxid, bool lock_held)
int	TwoPhaseGetDummyProcNumber (FullTransactionId fxid, bool lock_held)
GlobalTransaction	MarkAsPreparing (FullTransactionId fxid, const char *gid, TimestampTz prepared_at, Oid owner, Oid databaseid)
void	StartPrepare (GlobalTransaction gxact)
void	EndPrepare (GlobalTransaction gxact)
bool	StandbyTransactionIdIsPrepared (TransactionId xid)
TransactionId	PrescanPreparedTransactions (TransactionId **xids_p, int *nxids_p)
void	StandbyRecoverPreparedTransactions (void)
void	RecoverPreparedTransactions (void)
void	CheckPointTwoPhase (XLogRecPtr redo_horizon)
void	FinishPreparedTransaction (const char *gid, bool isCommit)
void	PrepareRedoAdd (FullTransactionId fxid, char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn, ReplOriginId origin_id)
void	PrepareRedoRemove (TransactionId xid, bool giveWarning)
void	restoreTwoPhaseData (void)
bool	LookupGXact (const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp)
void	TwoPhaseTransactionGid (Oid subid, TransactionId xid, char *gid_res, int szgid)
bool	LookupGXactBySubid (Oid subid)
TransactionId	TwoPhaseGetOldestXidInCommit (void)

Variables
PGDLLIMPORT int	max_prepared_xacts

Typedef Documentation

GlobalTransaction

typedef struct GlobalTransactionData* GlobalTransaction

Definition at line 26 of file twophase.h.

Function Documentation

AtAbort_Twophase()

void AtAbort_Twophase ( void  )

extern

Definition at line 307 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().

CheckPointTwoPhase()

void CheckPointTwoPhase ( XLogRecPtr  redo_horizon )

extern

Definition at line 1824 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 )

extern

Definition at line 1147 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  )

extern

Definition at line 1499 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().

LookupGXact()

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

extern

Definition at line 2690 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 )

extern

Definition at line 2799 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().

MarkAsPreparing()

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

extern

Definition at line 362 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().

PostPrepare_Twophase()

void PostPrepare_Twophase ( void  )

extern

Definition at line 347 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  )

extern

Definition at line 2509 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  )

extern

Definition at line 2666 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().

PrescanPreparedTransactions()

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

extern

Definition at line 1968 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().

RecoverPreparedTransactions()

void RecoverPreparedTransactions ( void  )

extern

Definition at line 2085 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().

restoreTwoPhaseData()

void restoreTwoPhaseData ( void  )

extern

Definition at line 1906 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().

StandbyRecoverPreparedTransactions()

void StandbyRecoverPreparedTransactions ( void  )

extern

Definition at line 2047 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 )

extern

Definition at line 1469 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 )

extern

Definition at line 1054 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().

TwoPhaseGetDummyProc()

PGPROC * TwoPhaseGetDummyProc ( FullTransactionId  fxid, bool  lock_held  )

extern

Definition at line 925 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()

int TwoPhaseGetDummyProcNumber ( FullTransactionId  fxid, bool  lock_held  )

extern

Definition at line 910 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().

TwoPhaseGetOldestXidInCommit()

TransactionId TwoPhaseGetOldestXidInCommit ( void  )

extern

Definition at line 2831 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  )

extern

Definition at line 858 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  )

extern

Definition at line 256 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  )

extern

Definition at line 240 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  )

extern

Definition at line 2749 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(), ERRCODE_PROTOCOL_VIOLATION, errmsg_internal(), ERROR, fb(), OidIsValid, snprintf, and TransactionIdIsValid.

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

Variable Documentation

max_prepared_xacts

PGDLLIMPORT int max_prepared_xacts

extern

Definition at line 117 of file twophase.c.

Referenced by CheckPointTwoPhase(), CheckRequiredParameterValues(), FastPathLockShmemSize(), GetLockConflicts(), GetSerializableTransactionSnapshotInt(), InitControlFile(), InitProcGlobal(), MarkAsPreparing(), PGProcShmemSize(), predicatelock_twophase_recover(), PredicateLockShmemInit(), PredicateLockShmemSize(), PrepareRedoAdd(), StandbyTransactionIdIsPrepared(), TwoPhaseShmemInit(), TwoPhaseShmemSize(), XactLockForVirtualXact(), and XLogReportParameters().