twophase.c File Reference

#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/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 "replication/walsender.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 "storage/sinvaladt.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/timestamp.h"

Include dependency graph for twophase.c:

Go to the source code of this file.

Data Structures
struct	GlobalTransactionData
struct	TwoPhaseStateData
struct	Working_State
struct	TwoPhaseRecordOnDisk
struct	StateFileChunk
struct	xllist

Macros
#define	TWOPHASE_DIR   "pg_twophase"
#define	TwoPhaseFilePath(path, xid)   snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)
#define	TWOPHASE_MAGIC   0x57F94534 /* format identifier */

Typedefs
typedef struct GlobalTransactionData	GlobalTransactionData
typedef struct TwoPhaseStateData	TwoPhaseStateData
typedef xl_xact_prepare	TwoPhaseFileHeader
typedef struct TwoPhaseRecordOnDisk	TwoPhaseRecordOnDisk
typedef struct StateFileChunk	StateFileChunk

Functions
static void	RecordTransactionCommitPrepared (TransactionId xid, int nchildren, TransactionId *children, int nrels, RelFileNode *rels, int ninvalmsgs, SharedInvalidationMessage *invalmsgs, bool initfileinval, const char *gid)
static void	RecordTransactionAbortPrepared (TransactionId xid, int nchildren, TransactionId *children, int nrels, RelFileNode *rels, const char *gid)
static void	ProcessRecords (char *bufptr, TransactionId xid, const TwoPhaseCallback callbacks[])
static void	RemoveGXact (GlobalTransaction gxact)
static void	XlogReadTwoPhaseData (XLogRecPtr lsn, char **buf, int *len)
static char *	ProcessTwoPhaseBuffer (TransactionId xid, XLogRecPtr prepare_start_lsn, bool fromdisk, bool setParent, bool setNextXid)
static void	MarkAsPreparingGuts (GlobalTransaction gxact, TransactionId xid, const char *gid, TimestampTz prepared_at, Oid owner, Oid databaseid)
static void	RemoveTwoPhaseFile (TransactionId xid, bool giveWarning)
static void	RecreateTwoPhaseFile (TransactionId xid, 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 (TransactionId xid, 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 (TransactionId xid, bool lock_held)
BackendId	TwoPhaseGetDummyBackendId (TransactionId xid, bool lock_held)
PGPROC *	TwoPhaseGetDummyProc (TransactionId xid, bool lock_held)
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 (TransactionId xid, 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 (char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn, RepOriginId origin_id)
void	PrepareRedoRemove (TransactionId xid, bool giveWarning)
bool	LookupGXact (const char *gid, XLogRecPtr prepare_end_lsn, TimestampTz origin_prepare_timestamp)

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 114 of file twophase.c.

Referenced by CheckPointTwoPhase(), and restoreTwoPhaseData().

TWOPHASE_MAGIC

#define TWOPHASE_MAGIC   0x57F94534 /* format identifier */

Definition at line 907 of file twophase.c.

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

TwoPhaseFilePath

#define TwoPhaseFilePath	(		path,
			xid
	)		snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)

Definition at line 885 of file twophase.c.

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

Typedef Documentation

GlobalTransactionData

typedef struct GlobalTransactionData GlobalTransactionData

StateFileChunk

typedef struct StateFileChunk StateFileChunk

TwoPhaseFileHeader

typedef xl_xact_prepare TwoPhaseFileHeader

Definition at line 909 of file twophase.c.

TwoPhaseRecordOnDisk

typedef struct TwoPhaseRecordOnDisk TwoPhaseRecordOnDisk

TwoPhaseStateData

typedef struct TwoPhaseStateData TwoPhaseStateData

Function Documentation

AtAbort_Twophase()

void AtAbort_Twophase

(

void

)

Definition at line 317 of file twophase.c.

References InvalidBackendId, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), RemoveGXact(), and GlobalTransactionData::valid.

Referenced by AbortTransaction(), and AtProcExit_Twophase().

{
    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 = InvalidBackendId;
    LWLockRelease(TwoPhaseStateLock);

    MyLockedGxact = NULL;
}

AtProcExit_Twophase()

static void AtProcExit_Twophase ( int  code, Datum  arg  )

static

Definition at line 307 of file twophase.c.

References AtAbort_Twophase().

Referenced by LockGXact(), and MarkAsPreparing().

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

CheckPointTwoPhase()

void CheckPointTwoPhase

(

XLogRecPtr

redo_horizon

)

Definition at line 1707 of file twophase.c.

References buf, ereport, errmsg_plural(), fsync_fname(), i, GlobalTransactionData::inredo, InvalidXLogRecPtr, LOG, log_checkpoints, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_prepared_xacts, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, pfree(), GlobalTransactionData::prepare_end_lsn, GlobalTransactionData::prepare_start_lsn, TwoPhaseStateData::prepXacts, RecreateTwoPhaseFile(), TWOPHASE_DIR, GlobalTransactionData::valid, GlobalTransactionData::xid, and XlogReadTwoPhaseData().

Referenced by CheckPointGuts().

{
    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 delayChkpt.
     */
    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->xid, 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)));
}

EndPrepare()

void EndPrepare

(

GlobalTransaction

gxact

)

Definition at line 1055 of file twophase.c.

References Assert, StateFileChunk::data, PGPROC::delayChkpt, DoNotReplicateId, END_CRIT_SECTION, ereport, errcode(), errmsg(), ERROR, xllist::head, InvalidRepOriginId, InvalidXLogRecPtr, StateFileChunk::len, xl_xact_prepare::magic, MarkAsPrepared(), MaxAllocSize, MyProc, StateFileChunk::next, xllist::num_chunks, xl_xact_prepare::origin_lsn, xl_xact_prepare::origin_timestamp, GlobalTransactionData::prepare_end_lsn, GlobalTransactionData::prepare_start_lsn, ProcLastRecPtr, records, RegisterTwoPhaseRecord(), replorigin_session_advance(), replorigin_session_origin, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, START_CRIT_SECTION, SyncRepWaitForLSN(), xllist::tail, xl_xact_prepare::total_len, xllist::total_len, TWOPHASE_MAGIC, TWOPHASE_RM_END_ID, XLOG_INCLUDE_ORIGIN, XLOG_XACT_PREPARE, XLogBeginInsert(), XLogEnsureRecordSpace(), XLogFlush(), XLogInsert(), XLogRegisterData(), and XLogSetRecordFlags().

Referenced by PrepareTransaction().

{
    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_session_origin != InvalidRepOriginId &&
                  replorigin_session_origin != DoNotReplicateId);

    if (replorigin)
    {
        Assert(replorigin_session_origin_lsn != InvalidXLogRecPtr);
        hdr->origin_lsn = replorigin_session_origin_lsn;
        hdr->origin_timestamp = replorigin_session_origin_timestamp;
    }
    else
    {
        hdr->origin_lsn = InvalidXLogRecPtr;
        hdr->origin_timestamp = 0;
    }

    /*
     * 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 delayChkpt 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 delayChkpt for; see notes there.)
     *
     * We save the PREPARE record's location in the gxact for later use by
     * CheckPointTwoPhase.
     */
    XLogEnsureRecordSpace(0, records.num_chunks);

    START_CRIT_SECTION();

    MyProc->delayChkpt = true;

    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_session_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->delayChkpt = false;

    /*
     * 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;
}

FinishPreparedTransaction()

void FinishPreparedTransaction	(	const char *	gid,
		bool	isCommit
	)

Definition at line 1409 of file twophase.c.

References PROC_HDR::allProcs, Assert, AtEOXact_PgStat(), buf, DropRelationFiles(), GetUserId(), xl_xact_prepare::gidlen, HOLD_INTERRUPTS, xl_xact_prepare::initfileinval, LockGXact(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXALIGN, xl_xact_prepare::nabortrels, xl_xact_prepare::ncommitrels, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, GlobalTransactionData::ondisk, pfree(), GlobalTransactionData::pgprocno, PredicateLockTwoPhaseFinish(), GlobalTransactionData::prepare_start_lsn, ProcArrayRemove(), ProcessRecords(), ProcGlobal, ReadTwoPhaseFile(), RecordTransactionAbortPrepared(), RecordTransactionCommitPrepared(), RelationCacheInitFilePostInvalidate(), RelationCacheInitFilePreInvalidate(), RemoveGXact(), RemoveTwoPhaseFile(), RESUME_INTERRUPTS, SendSharedInvalidMessages(), TransactionIdEquals, TransactionIdLatest(), twophase_postabort_callbacks, twophase_postcommit_callbacks, GlobalTransactionData::valid, GlobalTransactionData::xid, xl_xact_prepare::xid, and XlogReadTwoPhaseData().

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

{
    GlobalTransaction gxact;
    PGPROC     *proc;
    TransactionId xid;
    char       *buf;
    char       *bufptr;
    TwoPhaseFileHeader *hdr;
    TransactionId latestXid;
    TransactionId *children;
    RelFileNode *commitrels;
    RelFileNode *abortrels;
    RelFileNode *delrels;
    int         ndelrels;
    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 = &ProcGlobal->allProcs[gxact->pgprocno];
    xid = gxact->xid;

    /*
     * 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(xid, 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 = (RelFileNode *) bufptr;
    bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
    abortrels = (RelFileNode *) bufptr;
    bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
    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->ninvalmsgs, invalmsgs,
                                        hdr->initfileinval, gid);
    else
        RecordTransactionAbortPrepared(xid,
                                       hdr->nsubxacts, children,
                                       hdr->nabortrels, abortrels,
                                       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);

    /*
     * 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, xid, twophase_postcommit_callbacks);
    else
        ProcessRecords(bufptr, xid, twophase_postabort_callbacks);

    PredicateLockTwoPhaseFinish(xid, isCommit);

    /* 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 (gxact->ondisk)
        RemoveTwoPhaseFile(xid, true);

    MyLockedGxact = NULL;

    RESUME_INTERRUPTS();

    pfree(buf);
}

GetPreparedTransactionList()

static int GetPreparedTransactionList ( GlobalTransaction *  gxacts )

static

Definition at line 670 of file twophase.c.

References i, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, palloc(), and TwoPhaseStateData::prepXacts.

Referenced by pg_prepared_xact().

{
    GlobalTransaction array;
    int         num;
    int         i;

    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

    if (TwoPhaseState->numPrepXacts == 0)
    {
        LWLockRelease(TwoPhaseStateLock);

        *gxacts = NULL;
        return 0;
    }

    num = TwoPhaseState->numPrepXacts;
    array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
    *gxacts = array;
    for (i = 0; i < num; i++)
        memcpy(array + i, TwoPhaseState->prepXacts[i],
               sizeof(GlobalTransactionData));

    LWLockRelease(TwoPhaseStateLock);

    return num;
}

GXactLoadSubxactData()

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

static

Definition at line 508 of file twophase.c.

References PROC_HDR::allProcs, XidCacheStatus::count, XidCacheStatus::overflowed, PGPROC_MAX_CACHED_SUBXIDS, GlobalTransactionData::pgprocno, ProcGlobal, PGPROC::subxids, PGPROC::subxidStatus, and XidCache::xids.

Referenced by RecoverPreparedTransactions(), and StartPrepare().

{
    PGPROC     *proc = &ProcGlobal->allProcs[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;
    }
}

LockGXact()

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

static

Definition at line 556 of file twophase.c.

References PROC_HDR::allProcs, AtProcExit_Twophase(), before_shmem_exit(), PGPROC::databaseId, ereport, errcode(), errhint(), errmsg(), ERROR, GlobalTransactionData::gid, i, InvalidBackendId, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyBackendId, MyDatabaseId, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::owner, GlobalTransactionData::pgprocno, TwoPhaseStateData::prepXacts, ProcGlobal, superuser_arg(), twophaseExitRegistered, and GlobalTransactionData::valid.

Referenced by FinishPreparedTransaction().

{
    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 = &ProcGlobal->allProcs[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 != InvalidBackendId)
            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 = MyBackendId;
        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;
}

LookupGXact()

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

Definition at line 2482 of file twophase.c.

References Assert, buf, GlobalTransactionData::gid, i, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, xl_xact_prepare::origin_lsn, xl_xact_prepare::origin_timestamp, pfree(), GlobalTransactionData::prepare_start_lsn, TwoPhaseStateData::prepXacts, ReadTwoPhaseFile(), GlobalTransactionData::valid, GlobalTransactionData::xid, and XlogReadTwoPhaseData().

Referenced by apply_handle_rollback_prepared().

{
    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->xid, 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;
}

MarkAsPrepared()

static void MarkAsPrepared ( GlobalTransaction  gxact, bool  lock_held  )

static

Definition at line 534 of file twophase.c.

References PROC_HDR::allProcs, Assert, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), GlobalTransactionData::pgprocno, ProcArrayAdd(), ProcGlobal, and GlobalTransactionData::valid.

Referenced by EndPrepare(), and RecoverPreparedTransactions().

{
    /* 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(&ProcGlobal->allProcs[gxact->pgprocno]);
}

MarkAsPreparing()

GlobalTransaction MarkAsPreparing	(	TransactionId	xid,
		const char *	gid,
		TimestampTz	prepared_at,
		Oid	owner,
		Oid	databaseid
	)

Definition at line 372 of file twophase.c.

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

Referenced by PrepareTransaction().

{
    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, xid, 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;
}

MarkAsPreparingGuts()

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

static

Definition at line 446 of file twophase.c.

References PROC_HDR::allProcs, Assert, PGPROC::backendId, XidCacheStatus::count, PGPROC::databaseId, PGPROC::delayChkpt, GlobalTransactionData::gid, i, GlobalTransactionData::inredo, InvalidBackendId, InvalidOid, InvalidTransactionId, PGPROC::isBackgroundWorker, PGPROC::links, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockHeldByMeInMode(), PGPROC::lwWaiting, PGPROC::lwWaitMode, PGPROC::lxid, MemSet, MyBackendId, PGPROC::myProcLocks, NUM_LOCK_PARTITIONS, XidCacheStatus::overflowed, GlobalTransactionData::owner, pg_atomic_init_u64(), PGPROC::pgprocno, GlobalTransactionData::pgprocno, PGPROC::pid, GlobalTransactionData::prepared_at, PROC_WAIT_STATUS_OK, ProcGlobal, PGPROC::roleId, SHMQueueElemInit(), SHMQueueInit(), PGPROC::statusFlags, PGPROC::subxidStatus, PGPROC::tempNamespaceId, GlobalTransactionData::valid, PGPROC::waitLock, PGPROC::waitProcLock, PGPROC::waitStart, PGPROC::waitStatus, PGPROC::xid, GlobalTransactionData::xid, and PGPROC::xmin.

Referenced by MarkAsPreparing(), and RecoverPreparedTransactions().

{
    PGPROC     *proc;
    int         i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

    Assert(gxact != NULL);
    proc = &ProcGlobal->allProcs[gxact->pgprocno];

    /* Initialize the PGPROC entry */
    MemSet(proc, 0, sizeof(PGPROC));
    proc->pgprocno = gxact->pgprocno;
    SHMQueueElemInit(&(proc->links));
    proc->waitStatus = PROC_WAIT_STATUS_OK;
    /* We set up the gxact's VXID as InvalidBackendId/XID */
    proc->lxid = (LocalTransactionId) xid;
    proc->xid = xid;
    Assert(proc->xmin == InvalidTransactionId);
    proc->delayChkpt = false;
    proc->statusFlags = 0;
    proc->pid = 0;
    proc->backendId = InvalidBackendId;
    proc->databaseId = databaseid;
    proc->roleId = owner;
    proc->tempNamespaceId = InvalidOid;
    proc->isBackgroundWorker = false;
    proc->lwWaiting = false;
    proc->lwWaitMode = 0;
    proc->waitLock = NULL;
    proc->waitProcLock = NULL;
    pg_atomic_init_u64(&proc->waitStart, 0);
    for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
        SHMQueueInit(&(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->xid = xid;
    gxact->owner = owner;
    gxact->locking_backend = MyBackendId;
    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;
}

pg_prepared_xact()

Datum pg_prepared_xact

(

PG_FUNCTION_ARGS

)

Definition at line 715 of file twophase.c.

References PROC_HDR::allProcs, Working_State::array, BlessTupleDesc(), CreateTemplateTupleDesc(), CStringGetTextDatum, Working_State::currIdx, PGPROC::databaseId, GetPreparedTransactionList(), GlobalTransactionData::gid, heap_form_tuple(), HeapTupleGetDatum, MemoryContextSwitchTo(), MemSet, FuncCallContext::multi_call_memory_ctx, Working_State::ngxacts, ObjectIdGetDatum, GlobalTransactionData::owner, palloc(), GlobalTransactionData::pgprocno, GlobalTransactionData::prepared_at, ProcGlobal, SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, status(), TimestampTzGetDatum, TransactionIdGetDatum, FuncCallContext::tuple_desc, TupleDescInitEntry(), FuncCallContext::user_fctx, GlobalTransactionData::valid, values, and PGPROC::xid.

{
    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 = (Working_State *) palloc(sizeof(Working_State));
        funcctx->user_fctx = (void *) 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 = &ProcGlobal->allProcs[gxact->pgprocno];
        Datum       values[5];
        bool        nulls[5];
        HeapTuple   tuple;
        Datum       result;

        if (!gxact->valid)
            continue;

        /*
         * Form tuple with appropriate data.
         */
        MemSet(values, 0, sizeof(values));
        MemSet(nulls, 0, sizeof(nulls));

        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);
}

PostPrepare_Twophase()

void PostPrepare_Twophase

(

void

)

Definition at line 357 of file twophase.c.

References InvalidBackendId, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockAcquire(), and LWLockRelease().

Referenced by PrepareTransaction(), and RecoverPreparedTransactions().

{
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    MyLockedGxact->locking_backend = InvalidBackendId;
    LWLockRelease(TwoPhaseStateLock);

    MyLockedGxact = NULL;
}

PrepareRedoAdd()

void PrepareRedoAdd	(	char *	buf,
		XLogRecPtr	start_lsn,
		XLogRecPtr	end_lsn,
		RepOriginId	origin_id
	)

Definition at line 2360 of file twophase.c.

References Assert, DEBUG2, elog, ereport, errcode(), errhint(), errmsg(), ERROR, TwoPhaseStateData::freeGXacts, GlobalTransactionData::gid, GlobalTransactionData::inredo, InvalidBackendId, InvalidRepOriginId, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockHeldByMeInMode(), max_prepared_xacts, MAXALIGN, GlobalTransactionData::next, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, xl_xact_prepare::origin_lsn, GlobalTransactionData::owner, xl_xact_prepare::owner, GlobalTransactionData::prepare_end_lsn, GlobalTransactionData::prepare_start_lsn, GlobalTransactionData::prepared_at, xl_xact_prepare::prepared_at, TwoPhaseStateData::prepXacts, RecoveryInProgress(), replorigin_advance(), GlobalTransactionData::valid, GlobalTransactionData::xid, xl_xact_prepare::xid, and XLogRecPtrIsInvalid.

Referenced by restoreTwoPhaseData(), and xact_redo().

{
    TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf;
    char       *bufptr;
    const char *gid;
    GlobalTransaction gxact;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
    Assert(RecoveryInProgress());

    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
     */

    /* 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->xid = hdr->xid;
    gxact->owner = hdr->owner;
    gxact->locking_backend = InvalidBackendId;
    gxact->valid = false;
    gxact->ondisk = XLogRecPtrIsInvalid(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 != InvalidRepOriginId)
    {
        /* 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", gxact->xid);
}

PrepareRedoRemove()

void PrepareRedoRemove	(	TransactionId	xid,
		bool	giveWarning
	)

Definition at line 2430 of file twophase.c.

References Assert, DEBUG2, elog, i, GlobalTransactionData::inredo, LW_EXCLUSIVE, LWLockHeldByMeInMode(), TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, TwoPhaseStateData::prepXacts, RecoveryInProgress(), RemoveGXact(), RemoveTwoPhaseFile(), and GlobalTransactionData::xid.

Referenced by ProcessTwoPhaseBuffer(), and xact_redo().

{
    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 (gxact->xid == xid)
        {
            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", xid);
    if (gxact->ondisk)
        RemoveTwoPhaseFile(xid, giveWarning);
    RemoveGXact(gxact);
}

PrescanPreparedTransactions()

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

Definition at line 1851 of file twophase.c.

References Assert, buf, i, GlobalTransactionData::inredo, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::nextXid, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, palloc(), pfree(), GlobalTransactionData::prepare_start_lsn, TwoPhaseStateData::prepXacts, ProcessTwoPhaseBuffer(), repalloc(), ShmemVariableCache, TransactionIdPrecedes(), GlobalTransactionData::xid, and XidFromFullTransactionId.

Referenced by StartupXLOG(), and xlog_redo().

{
    FullTransactionId nextXid = ShmemVariableCache->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);

        xid = gxact->xid;

        buf = ProcessTwoPhaseBuffer(xid,
                                    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.
         */
        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;
}

ProcessRecords()

static void ProcessRecords ( char *  bufptr, TransactionId  xid, const TwoPhaseCallback  callbacks[]  )

static

Definition at line 1580 of file twophase.c.

References Assert, TwoPhaseRecordOnDisk::info, TwoPhaseRecordOnDisk::len, MAXALIGN, TwoPhaseRecordOnDisk::rmid, TWOPHASE_RM_END_ID, TWOPHASE_RM_MAX_ID, and GlobalTransactionData::xid.

Referenced by FinishPreparedTransaction(), and RecoverPreparedTransactions().

{
    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] (xid, record->info,
                                     (void *) bufptr, record->len);

        bufptr += MAXALIGN(record->len);
    }
}

ProcessTwoPhaseBuffer()

static char * ProcessTwoPhaseBuffer ( TransactionId  xid, XLogRecPtr  prepare_start_lsn, bool  fromdisk, bool  setParent, bool  setNextXid  )

static

Definition at line 2074 of file twophase.c.

References AdvanceNextFullTransactionIdPastXid(), Assert, buf, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errmsg(), ERROR, xl_xact_prepare::gidlen, i, InvalidXLogRecPtr, LW_EXCLUSIVE, LWLockHeldByMeInMode(), MAXALIGN, VariableCacheData::nextXid, xl_xact_prepare::nsubxacts, PrepareRedoRemove(), ReadTwoPhaseFile(), RemoveTwoPhaseFile(), ShmemVariableCache, SubTransSetParent(), TransactionIdDidAbort(), TransactionIdDidCommit(), TransactionIdEquals, TransactionIdFollows(), TransactionIdFollowsOrEquals(), WARNING, xl_xact_prepare::xid, XidFromFullTransactionId, and XlogReadTwoPhaseData().

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

{
    FullTransactionId nextXid = ShmemVariableCache->nextXid;
    TransactionId origNextXid = XidFromFullTransactionId(nextXid);
    TransactionId *subxids;
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int         i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

    if (!fromdisk)
        Assert(prepare_start_lsn != InvalidXLogRecPtr);

    /* Already processed? */
    if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state file for transaction %u",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing stale two-phase state from memory for transaction %u",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        return NULL;
    }

    /* Reject XID if too new */
    if (TransactionIdFollowsOrEquals(xid, origNextXid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state file for transaction %u",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing future two-phase state from memory for transaction %u",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        return NULL;
    }

    if (fromdisk)
    {
        /* Read and validate file */
        buf = ReadTwoPhaseFile(xid, false);
    }
    else
    {
        /* Read xlog data */
        XlogReadTwoPhaseData(prepare_start_lsn, &buf, NULL);
    }

    /* Deconstruct header */
    hdr = (TwoPhaseFileHeader *) buf;
    if (!TransactionIdEquals(hdr->xid, xid))
    {
        if (fromdisk)
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("corrupted two-phase state file for transaction %u",
                            xid)));
        else
            ereport(ERROR,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("corrupted two-phase state in memory for transaction %u",
                            xid)));
    }

    /*
     * 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, xid));

        /* update nextXid if needed */
        if (setNextXid)
            AdvanceNextFullTransactionIdPastXid(subxid);

        if (setParent)
            SubTransSetParent(subxid, xid);
    }

    return buf;
}

ReadTwoPhaseFile()

static char* ReadTwoPhaseFile ( TransactionId  xid, bool  missing_ok  )

static

Definition at line 1205 of file twophase.c.

References buf, CloseTransientFile(), COMP_CRC32C, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errmsg(), errmsg_plural(), ERROR, 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, TwoPhaseFilePath, and WAIT_EVENT_TWOPHASE_FILE_READ.

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

{
    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, xid);

    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;
}

RecordTransactionAbortPrepared()

static void RecordTransactionAbortPrepared ( TransactionId  xid, int  nchildren, TransactionId *  children, int  nrels, RelFileNode *  rels, const char *  gid  )

static

Definition at line 2288 of file twophase.c.

References DoNotReplicateId, elog, END_CRIT_SECTION, GetCurrentTimestamp(), InvalidRepOriginId, MyXactFlags, PANIC, replorigin_session_advance(), replorigin_session_origin, replorigin_session_origin_lsn, START_CRIT_SECTION, SyncRepWaitForLSN(), TransactionIdAbortTree(), TransactionIdDidCommit(), XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK, XactLastRecEnd, XactLogAbortRecord(), and XLogFlush().

Referenced by FinishPreparedTransaction().

{
    XLogRecPtr  recptr;
    bool        replorigin;

    /*
     * Are we using the replication origins feature?  Or, in other words, are
     * we replaying remote actions?
     */
    replorigin = (replorigin_session_origin != InvalidRepOriginId &&
                  replorigin_session_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,
                                MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                xid, gid);

    if (replorigin)
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_session_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);
}

RecordTransactionCommitPrepared()

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

static

Definition at line 2194 of file twophase.c.

References PGPROC::delayChkpt, DoNotReplicateId, END_CRIT_SECTION, GetCurrentTimestamp(), InvalidRepOriginId, MyProc, MyXactFlags, replorigin_session_advance(), replorigin_session_origin, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, START_CRIT_SECTION, SyncRepWaitForLSN(), TransactionIdCommitTree(), TransactionTreeSetCommitTsData(), XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK, XactLastRecEnd, XactLogCommitRecord(), and XLogFlush().

Referenced by FinishPreparedTransaction().

{
    XLogRecPtr  recptr;
    TimestampTz committs = GetCurrentTimestamp();
    bool        replorigin;

    /*
     * Are we using the replication origins feature?  Or, in other words, are
     * we replaying remote actions?
     */
    replorigin = (replorigin_session_origin != InvalidRepOriginId &&
                  replorigin_session_origin != DoNotReplicateId);

    START_CRIT_SECTION();

    /* See notes in RecordTransactionCommit */
    MyProc->delayChkpt = true;

    /*
     * 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,
                                 ninvalmsgs, invalmsgs,
                                 initfileinval,
                                 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                 xid, gid);


    if (replorigin)
        /* Move LSNs forward for this replication origin */
        replorigin_session_advance(replorigin_session_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_session_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_session_origin_timestamp == 0)
        replorigin_session_origin_timestamp = committs;

    TransactionTreeSetCommitTsData(xid, nchildren, children,
                                   replorigin_session_origin_timestamp,
                                   replorigin_session_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->delayChkpt = false;

    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);
}

RecoverPreparedTransactions()

void RecoverPreparedTransactions

(

void

)

Definition at line 1973 of file twophase.c.

References Assert, buf, xl_xact_prepare::database, ereport, errmsg(), GlobalTransactionData::gid, xl_xact_prepare::gidlen, GXactLoadSubxactData(), i, InHotStandby, GlobalTransactionData::inredo, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MarkAsPrepared(), MarkAsPreparingGuts(), MAXALIGN, xl_xact_prepare::nabortrels, xl_xact_prepare::ncommitrels, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, xl_xact_prepare::owner, pfree(), PostPrepare_Twophase(), GlobalTransactionData::prepare_start_lsn, xl_xact_prepare::prepared_at, TwoPhaseStateData::prepXacts, ProcessRecords(), ProcessTwoPhaseBuffer(), StandbyReleaseLockTree(), TransactionIdEquals, twophase_recover_callbacks, GlobalTransactionData::xid, and xl_xact_prepare::xid.

Referenced by StartupXLOG().

{
    int         i;

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        TransactionId xid;
        char       *buf;
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        char       *bufptr;
        TwoPhaseFileHeader *hdr;
        TransactionId *subxids;
        const char *gid;

        xid = gxact->xid;

        /*
         * 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(xid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, true, false);
        if (buf == NULL)
            continue;

        ereport(LOG,
                (errmsg("recovering prepared transaction %u from shared memory", xid)));

        hdr = (TwoPhaseFileHeader *) buf;
        Assert(TransactionIdEquals(hdr->xid, xid));
        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(RelFileNode));
        bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
        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, xid, 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, xid, 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(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);
}

RecreateTwoPhaseFile()

static void RecreateTwoPhaseFile ( TransactionId  xid, void *  content, int  len  )

static

Definition at line 1627 of file twophase.c.

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

Referenced by CheckPointTwoPhase().

{
    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, xid);

    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)));
}

RegisterTwoPhaseRecord()

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

Definition at line 1182 of file twophase.c.

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

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

{
    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);
}

RemoveGXact()

static void RemoveGXact ( GlobalTransaction  gxact )

static

Definition at line 632 of file twophase.c.

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

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

{
    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);
}

RemoveTwoPhaseFile()

static void RemoveTwoPhaseFile ( TransactionId  xid, bool  giveWarning  )

static

Definition at line 1608 of file twophase.c.

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

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

{
    char        path[MAXPGPATH];

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

restoreTwoPhaseData()

void restoreTwoPhaseData

(

void

)

Definition at line 1789 of file twophase.c.

References AllocateDir(), buf, dirent::d_name, FreeDir(), InvalidRepOriginId, InvalidXLogRecPtr, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), PrepareRedoAdd(), ProcessTwoPhaseBuffer(), ReadDir(), TWOPHASE_DIR, and GlobalTransactionData::xid.

Referenced by StartupXLOG().

{
    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) == 8 &&
            strspn(clde->d_name, "0123456789ABCDEF") == 8)
        {
            TransactionId xid;
            char       *buf;

            xid = (TransactionId) strtoul(clde->d_name, NULL, 16);

            buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr,
                                        true, false, false);
            if (buf == NULL)
                continue;

            PrepareRedoAdd(buf, InvalidXLogRecPtr,
                           InvalidXLogRecPtr, InvalidRepOriginId);
        }
    }
    LWLockRelease(TwoPhaseStateLock);
    FreeDir(cldir);
}

save_state_data()

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

static

Definition at line 955 of file twophase.c.

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

Referenced by RegisterTwoPhaseRecord(), and StartPrepare().

{
    uint32      padlen = MAXALIGN(len);

    if (padlen > records.bytes_free)
    {
        records.tail->next = palloc0(sizeof(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(((char *) records.tail->data) + records.tail->len, data, len);
    records.tail->len += padlen;
    records.bytes_free -= padlen;
    records.total_len += padlen;
}

StandbyRecoverPreparedTransactions()

void StandbyRecoverPreparedTransactions

(

void

)

Definition at line 1932 of file twophase.c.

References Assert, buf, i, GlobalTransactionData::inredo, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, pfree(), GlobalTransactionData::prepare_start_lsn, TwoPhaseStateData::prepXacts, ProcessTwoPhaseBuffer(), and GlobalTransactionData::xid.

Referenced by StartupXLOG(), and xlog_redo().

{
    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);

        xid = gxact->xid;

        buf = ProcessTwoPhaseBuffer(xid,
                                    gxact->prepare_start_lsn,
                                    gxact->ondisk, false, false);
        if (buf != NULL)
            pfree(buf);
    }
    LWLockRelease(TwoPhaseStateLock);
}

StandbyTransactionIdIsPrepared()

bool StandbyTransactionIdIsPrepared

(

TransactionId

xid

)

Definition at line 1381 of file twophase.c.

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

Referenced by KnownAssignedXidsRemovePreceding(), and StandbyReleaseOldLocks().

{
    char       *buf;
    TwoPhaseFileHeader *hdr;
    bool        result;

    Assert(TransactionIdIsValid(xid));

    if (max_prepared_xacts <= 0)
        return false;           /* nothing to do */

    /* Read and validate file */
    buf = ReadTwoPhaseFile(xid, true);
    if (buf == NULL)
        return false;

    /* Check header also */
    hdr = (TwoPhaseFileHeader *) buf;
    result = TransactionIdEquals(hdr->xid, xid);
    pfree(buf);

    return result;
}

StartPrepare()

void StartPrepare

(

GlobalTransaction

gxact

)

Definition at line 983 of file twophase.c.

References PROC_HDR::allProcs, xllist::bytes_free, StateFileChunk::data, xl_xact_prepare::database, PGPROC::databaseId, GlobalTransactionData::gid, xl_xact_prepare::gidlen, GXactLoadSubxactData(), xllist::head, xl_xact_prepare::initfileinval, StateFileChunk::len, xl_xact_prepare::magic, Max, xl_xact_prepare::nabortrels, xl_xact_prepare::ncommitrels, StateFileChunk::next, xl_xact_prepare::ninvalmsgs, xl_xact_prepare::nsubxacts, xllist::num_chunks, GlobalTransactionData::owner, xl_xact_prepare::owner, palloc(), palloc0(), pfree(), GlobalTransactionData::pgprocno, GlobalTransactionData::prepared_at, xl_xact_prepare::prepared_at, ProcGlobal, records, save_state_data(), smgrGetPendingDeletes(), xllist::tail, xl_xact_prepare::total_len, xllist::total_len, TWOPHASE_MAGIC, xactGetCommittedChildren(), xactGetCommittedInvalidationMessages(), GlobalTransactionData::xid, and xl_xact_prepare::xid.

Referenced by PrepareTransaction().

{
    PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
    TransactionId xid = gxact->xid;
    TwoPhaseFileHeader hdr;
    TransactionId *children;
    RelFileNode *commitrels;
    RelFileNode *abortrels;
    SharedInvalidationMessage *invalmsgs;

    /* Initialize linked list */
    records.head = palloc0(sizeof(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.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
                                                          &hdr.initfileinval);
    hdr.gidlen = strlen(gxact->gid) + 1;    /* Include '\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(RelFileNode));
        pfree(commitrels);
    }
    if (hdr.nabortrels > 0)
    {
        save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileNode));
        pfree(abortrels);
    }
    if (hdr.ninvalmsgs > 0)
    {
        save_state_data(invalmsgs,
                        hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
        pfree(invalmsgs);
    }
}

TwoPhaseGetDummyBackendId()

BackendId TwoPhaseGetDummyBackendId	(	TransactionId	xid,
		bool	lock_held
	)

Definition at line 859 of file twophase.c.

References GlobalTransactionData::dummyBackendId, and TwoPhaseGetGXact().

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

{
    GlobalTransaction gxact = TwoPhaseGetGXact(xid, lock_held);

    return gxact->dummyBackendId;
}

TwoPhaseGetDummyProc()

PGPROC* TwoPhaseGetDummyProc	(	TransactionId	xid,
		bool	lock_held
	)

Definition at line 874 of file twophase.c.

References PROC_HDR::allProcs, GlobalTransactionData::pgprocno, ProcGlobal, and TwoPhaseGetGXact().

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

{
    GlobalTransaction gxact = TwoPhaseGetGXact(xid, lock_held);

    return &ProcGlobal->allProcs[gxact->pgprocno];
}

TwoPhaseGetGXact()

static GlobalTransaction TwoPhaseGetGXact ( TransactionId  xid, bool  lock_held  )

static

Definition at line 806 of file twophase.c.

References Assert, elog, ERROR, i, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, TwoPhaseStateData::prepXacts, and GlobalTransactionData::xid.

Referenced by TwoPhaseGetDummyBackendId(), and TwoPhaseGetDummyProc().

{
    GlobalTransaction result = NULL;
    int         i;

    static TransactionId cached_xid = 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 (xid == cached_xid)
        return cached_gxact;

    if (!lock_held)
        LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

        if (gxact->xid == xid)
        {
            result = gxact;
            break;
        }
    }

    if (!lock_held)
        LWLockRelease(TwoPhaseStateLock);

    if (result == NULL)         /* should not happen */
        elog(ERROR, "failed to find GlobalTransaction for xid %u", xid);

    cached_xid = xid;
    cached_gxact = result;

    return result;
}

TwoPhaseShmemInit()

void TwoPhaseShmemInit

(

void

)

Definition at line 252 of file twophase.c.

References Assert, GlobalTransactionData::dummyBackendId, TwoPhaseStateData::freeGXacts, i, IsUnderPostmaster, max_prepared_xacts, MAXALIGN, MaxBackends, GlobalTransactionData::next, TwoPhaseStateData::numPrepXacts, offsetof, PGPROC::pgprocno, GlobalTransactionData::pgprocno, PreparedXactProcs, ShmemInitStruct(), and TwoPhaseShmemSize().

Referenced by CreateSharedMemoryAndSemaphores().

{
    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 = PreparedXactProcs[i].pgprocno;

            /*
             * Assign a unique ID for each dummy proc, so that the range of
             * dummy backend IDs immediately follows the range of normal
             * backend IDs. We don't dare to assign a real backend ID to dummy
             * procs, because prepared transactions don't take part in cache
             * invalidation like a real backend ID would imply, but having a
             * unique ID for them is nevertheless handy. This arrangement
             * allows you to allocate an array of size (MaxBackends +
             * max_prepared_xacts + 1), and have a slot for every backend and
             * prepared transaction. Currently multixact.c uses that
             * technique.
             */
            gxacts[i].dummyBackendId = MaxBackends + 1 + i;
        }
    }
    else
        Assert(found);
}

TwoPhaseShmemSize()

Size TwoPhaseShmemSize

(

void

)

Definition at line 236 of file twophase.c.

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

Referenced by CalculateShmemSize(), and TwoPhaseShmemInit().

{
    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;
}

XlogReadTwoPhaseData()

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

static

Definition at line 1326 of file twophase.c.

References ereport, errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, LSN_FORMAT_ARGS, palloc(), read_local_xlog_page(), ThisTimeLineID, 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().

{
    XLogRecord *record;
    XLogReaderState *xlogreader;
    char       *errormsg;
    TimeLineID  save_currtli = ThisTimeLineID;

    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);

    /*
     * Restore immediately the timeline where it was previously, as
     * read_local_xlog_page() could have changed it if the record was read
     * while recovery was finishing or if the timeline has jumped in-between.
     */
    ThisTimeLineID = save_currtli;

    if (record == NULL)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not read two-phase state from WAL at %X/%X",
                        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/%X",
                        LSN_FORMAT_ARGS(lsn))));

    if (len != NULL)
        *len = XLogRecGetDataLen(xlogreader);

    *buf = palloc(sizeof(char) * XLogRecGetDataLen(xlogreader));
    memcpy(*buf, XLogRecGetData(xlogreader), sizeof(char) * XLogRecGetDataLen(xlogreader));

    XLogReaderFree(xlogreader);
}

Variable Documentation

max_prepared_xacts

int max_prepared_xacts = 0

Definition at line 117 of file twophase.c.

Referenced by CheckPointTwoPhase(), CheckRequiredParameterValues(), GetLockConflicts(), GetSerializableTransactionSnapshotInt(), InitControlFile(), InitPredicateLocks(), InitProcGlobal(), MarkAsPreparing(), predicatelock_twophase_recover(), PredicateLockShmemSize(), PrepareRedoAdd(), ProcGlobalShmemSize(), StandbyTransactionIdIsPrepared(), TwoPhaseShmemInit(), TwoPhaseShmemSize(), and XLogReportParameters().

MyLockedGxact

GlobalTransaction MyLockedGxact = NULL

static

Definition at line 199 of file twophase.c.

records

struct xllist records

static

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

twophaseExitRegistered

bool twophaseExitRegistered = false

static

Definition at line 201 of file twophase.c.

Referenced by LockGXact(), and MarkAsPreparing().

TwoPhaseState

TwoPhaseStateData* TwoPhaseState

static

Definition at line 191 of file twophase.c.