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/xlogutils.h"
#include "access/xlogreader.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/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	TwoPhaseFileHeader
struct	TwoPhaseRecordOnDisk
struct	StateFileChunk
struct	xllist

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

Typedefs
typedef struct GlobalTransactionData	GlobalTransactionData
typedef struct TwoPhaseStateData	TwoPhaseStateData
typedef struct TwoPhaseFileHeader	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)
static void	RecordTransactionAbortPrepared (TransactionId xid, int nchildren, TransactionId *children, int nrels, RelFileNode *rels)
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)
BackendId	TwoPhaseGetDummyBackendId (TransactionId xid)
PGPROC *	TwoPhaseGetDummyProc (TransactionId xid)
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 give_warnings)
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)
void	PrepareRedoRemove (TransactionId xid, bool giveWarning)

Variables
int	max_prepared_xacts = 0
static TwoPhaseStateData *	TwoPhaseState
static GlobalTransaction	MyLockedGxact = NULL
static bool	twophaseExitRegistered = false
static struct xllist	records

Macro Definition Documentation

#define GIDSIZE   200

Definition at line 151 of file twophase.c.

Referenced by MarkAsPreparing().

#define TWOPHASE_DIR   "pg_twophase"

Definition at line 114 of file twophase.c.

Referenced by CheckPointTwoPhase(), and restoreTwoPhaseData().

#define TWOPHASE_MAGIC   0x57F94533 /* format identifier */

Definition at line 901 of file twophase.c.

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

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

Definition at line 879 of file twophase.c.

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

Typedef Documentation

typedef struct GlobalTransactionData GlobalTransactionData

typedef struct StateFileChunk StateFileChunk

typedef struct TwoPhaseFileHeader TwoPhaseFileHeader

typedef struct TwoPhaseRecordOnDisk TwoPhaseRecordOnDisk

typedef struct TwoPhaseStateData TwoPhaseStateData

Function Documentation

void AtAbort_Twophase

(

void

)

Definition at line 319 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;
}

static void AtProcExit_Twophase ( int  code, Datum  arg  )

static

Definition at line 309 of file twophase.c.

References AtAbort_Twophase().

Referenced by LockGXact(), and MarkAsPreparing().

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

void CheckPointTwoPhase

(

XLogRecPtr

redo_horizon

)

Definition at line 1652 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 pgxact 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)));
}

void EndPrepare

(

GlobalTransaction

gxact

)

Definition at line 1064 of file twophase.c.

References Assert, StateFileChunk::data, PGXACT::delayChkpt, END_CRIT_SECTION, ereport, errcode(), errmsg(), ERROR, xllist::head, StateFileChunk::len, TwoPhaseFileHeader::magic, MarkAsPrepared(), MaxAllocSize, MyPgXact, StateFileChunk::next, xllist::num_chunks, GlobalTransactionData::prepare_end_lsn, GlobalTransactionData::prepare_start_lsn, ProcLastRecPtr, records, RegisterTwoPhaseRecord(), START_CRIT_SECTION, SyncRepWaitForLSN(), xllist::tail, TwoPhaseFileHeader::total_len, xllist::total_len, TWOPHASE_MAGIC, TWOPHASE_RM_END_ID, XLOG_XACT_PREPARE, XLogBeginInsert(), XLogEnsureRecordSpace(), XLogFlush(), XLogInsert(), and XLogRegisterData().

Referenced by PrepareTransaction().

{
    TwoPhaseFileHeader *hdr;
    StateFileChunk *record;

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

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

    MyPgXact->delayChkpt = true;

    XLogBeginInsert();
    for (record = records.head; record != NULL; record = record->next)
        XLogRegisterData(record->data, record->len);
    gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
    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
     * MyPgXact 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 MyPgXact,
     * 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.
     */
    MyPgXact->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;
}

void FinishPreparedTransaction	(	const char *	gid,
		bool	isCommit
	)

Definition at line 1368 of file twophase.c.

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

Referenced by standard_ProcessUtility().

{
    GlobalTransaction gxact;
    PGPROC     *proc;
    PGXACT     *pgxact;
    TransactionId xid;
    char       *buf;
    char       *bufptr;
    TwoPhaseFileHeader *hdr;
    TransactionId latestXid;
    TransactionId *children;
    RelFileNode *commitrels;
    RelFileNode *abortrels;
    RelFileNode *delrels;
    int         ndelrels;
    SharedInvalidationMessage *invalmsgs;
    int         i;

    /*
     * 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];
    pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
    xid = pgxact->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, true);
    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);

    /*
     * 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);
    else
        RecordTransactionAbortPrepared(xid,
                                       hdr->nsubxacts, children,
                                       hdr->nabortrels, abortrels);

    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;
    }
    for (i = 0; i < ndelrels; i++)
    {
        SMgrRelation srel = smgropen(delrels[i], InvalidBackendId);

        smgrdounlink(srel, false);
        smgrclose(srel);
    }

    /*
     * Handle cache invalidation messages.
     *
     * Relcache init file invalidation requires processing both before and
     * after we send the SI messages. See AtEOXact_Inval()
     */
    if (hdr->initfileinval)
        RelationCacheInitFilePreInvalidate();
    SendSharedInvalidMessages(invalmsgs, hdr->ninvalmsgs);
    if (hdr->initfileinval)
        RelationCacheInitFilePostInvalidate();

    /* And now do the callbacks */
    if (isCommit)
        ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
    else
        ProcessRecords(bufptr, xid, twophase_postabort_callbacks);

    PredicateLockTwoPhaseFinish(xid, isCommit);

    /* Count the prepared xact as committed or aborted */
    AtEOXact_PgStat(isCommit);

    /*
     * And now we can clean up any files we may have left.
     */
    if (gxact->ondisk)
        RemoveTwoPhaseFile(xid, true);

    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    RemoveGXact(gxact);
    LWLockRelease(TwoPhaseStateLock);
    MyLockedGxact = NULL;

    pfree(buf);
}

static int GetPreparedTransactionList ( GlobalTransaction *  gxacts )

static

Definition at line 673 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;
}

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

static

Definition at line 510 of file twophase.c.

References PROC_HDR::allPgXact, PROC_HDR::allProcs, PGXACT::nxids, PGXACT::overflowed, PGPROC_MAX_CACHED_SUBXIDS, GlobalTransactionData::pgprocno, ProcGlobal, PGPROC::subxids, and XidCache::xids.

Referenced by RecoverPreparedTransactions(), and StartPrepare().

{
    PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
    PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];

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

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

static

Definition at line 559 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;
}

static void MarkAsPrepared ( GlobalTransaction  gxact, bool  lock_held  )

static

Definition at line 537 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]);
}

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

Definition at line 374 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;
}

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

static

Definition at line 448 of file twophase.c.

References PROC_HDR::allPgXact, PROC_HDR::allProcs, Assert, PGPROC::backendId, PGPROC::databaseId, PGXACT::delayChkpt, GlobalTransactionData::gid, i, GlobalTransactionData::inredo, InvalidBackendId, InvalidTransactionId, PGPROC::isBackgroundWorker, PGPROC::links, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockHeldByMeInMode(), PGPROC::lwWaiting, PGPROC::lwWaitMode, PGPROC::lxid, MemSet, MyBackendId, PGPROC::myProcLocks, NUM_LOCK_PARTITIONS, PGXACT::nxids, PGXACT::overflowed, GlobalTransactionData::owner, PGPROC::pgprocno, GlobalTransactionData::pgprocno, PGPROC::pid, GlobalTransactionData::prepared_at, ProcGlobal, PGPROC::roleId, SHMQueueElemInit(), SHMQueueInit(), STATUS_OK, PGXACT::vacuumFlags, GlobalTransactionData::valid, PGPROC::waitLock, PGPROC::waitProcLock, PGPROC::waitStatus, GlobalTransactionData::xid, PGXACT::xid, and PGXACT::xmin.

Referenced by MarkAsPreparing(), and RecoverPreparedTransactions().

{
    PGPROC     *proc;
    PGXACT     *pgxact;
    int         i;

    Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));

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

    /* Initialize the PGPROC entry */
    MemSet(proc, 0, sizeof(PGPROC));
    proc->pgprocno = gxact->pgprocno;
    SHMQueueElemInit(&(proc->links));
    proc->waitStatus = STATUS_OK;
    /* We set up the gxact's VXID as InvalidBackendId/XID */
    proc->lxid = (LocalTransactionId) xid;
    pgxact->xid = xid;
    pgxact->xmin = InvalidTransactionId;
    pgxact->delayChkpt = false;
    pgxact->vacuumFlags = 0;
    proc->pid = 0;
    proc->backendId = InvalidBackendId;
    proc->databaseId = databaseid;
    proc->roleId = owner;
    proc->isBackgroundWorker = false;
    proc->lwWaiting = false;
    proc->lwWaitMode = 0;
    proc->waitLock = NULL;
    proc->waitProcLock = NULL;
    for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
        SHMQueueInit(&(proc->myProcLocks[i]));
    /* subxid data must be filled later by GXactLoadSubxactData */
    pgxact->overflowed = false;
    pgxact->nxids = 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;
}

Datum pg_prepared_xact

(

PG_FUNCTION_ARGS

)

Definition at line 718 of file twophase.c.

References PROC_HDR::allPgXact, 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, OIDOID, GlobalTransactionData::owner, palloc(), GlobalTransactionData::pgprocno, GlobalTransactionData::prepared_at, ProcGlobal, result, SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, status(), TEXTOID, TimestampTzGetDatum, TIMESTAMPTZOID, TransactionIdGetDatum, FuncCallContext::tuple_desc, TupleDescInitEntry(), FuncCallContext::user_fctx, GlobalTransactionData::valid, values, PGXACT::xid, and XIDOID.

{
    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, false);
        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];
        PGXACT     *pgxact = &ProcGlobal->allPgXact[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(pgxact->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);
}

void PostPrepare_Twophase

(

void

)

Definition at line 359 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;
}

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

Definition at line 2313 of file twophase.c.

References Assert, DEBUG2, elog, ereport, errcode(), errhint(), errmsg(), ERROR, TwoPhaseStateData::freeGXacts, GlobalTransactionData::gid, GlobalTransactionData::inredo, InvalidBackendId, GlobalTransactionData::locking_backend, LW_EXCLUSIVE, LWLockHeldByMeInMode(), max_prepared_xacts, MAXALIGN, GlobalTransactionData::next, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, GlobalTransactionData::owner, TwoPhaseFileHeader::owner, GlobalTransactionData::prepare_end_lsn, GlobalTransactionData::prepare_start_lsn, GlobalTransactionData::prepared_at, TwoPhaseFileHeader::prepared_at, TwoPhaseStateData::prepXacts, RecoveryInProgress(), GlobalTransactionData::valid, GlobalTransactionData::xid, TwoPhaseFileHeader::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;

    elog(DEBUG2, "added 2PC data in shared memory for transaction %u", gxact->xid);
}

void PrepareRedoRemove	(	TransactionId	xid,
		bool	giveWarning
	)

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

    return;
}

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

Definition at line 1791 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(), result, ShmemVariableCache, TransactionIdPrecedes(), and GlobalTransactionData::xid.

Referenced by StartupXLOG(), and xlog_redo().

{
    TransactionId origNextXid = ShmemVariableCache->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;
}

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

static

Definition at line 1522 of file twophase.c.

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

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

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

static

Definition at line 2013 of file twophase.c.

References Assert, buf, ereport, errmsg(), TwoPhaseFileHeader::gidlen, i, InvalidXLogRecPtr, LW_EXCLUSIVE, LWLockAcquire(), LWLockHeldByMeInMode(), LWLockRelease(), MAXALIGN, VariableCacheData::nextXid, TwoPhaseFileHeader::nsubxacts, pfree(), PrepareRedoRemove(), ReadTwoPhaseFile(), RemoveTwoPhaseFile(), ShmemVariableCache, SubTransSetParent(), TransactionIdAdvance, TransactionIdDidAbort(), TransactionIdDidCommit(), TransactionIdEquals, TransactionIdFollows(), TransactionIdFollowsOrEquals(), WARNING, TwoPhaseFileHeader::xid, and XlogReadTwoPhaseData().

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

{
    TransactionId origNextXid = ShmemVariableCache->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, true);
        if (buf == NULL)
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state file for transaction %u",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
            return NULL;
        }
    }
    else
    {
        /* Read xlog data */
        XlogReadTwoPhaseData(prepare_start_lsn, &buf, NULL);
    }

    /* Deconstruct header */
    hdr = (TwoPhaseFileHeader *) buf;
    if (!TransactionIdEquals(hdr->xid, xid))
    {
        if (fromdisk)
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state file for transaction %u",
                            xid)));
            RemoveTwoPhaseFile(xid, true);
        }
        else
        {
            ereport(WARNING,
                    (errmsg("removing corrupt two-phase state from memory for transaction %u",
                            xid)));
            PrepareRedoRemove(xid, true);
        }
        pfree(buf);
        return NULL;
    }

    /*
     * 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 &&
            TransactionIdFollowsOrEquals(subxid,
                                         ShmemVariableCache->nextXid))
        {
            /*
             * We don't expect anyone else to modify nextXid, hence we don't
             * need to hold a lock while examining it.  We still acquire the
             * lock to modify it, though, so we recheck.
             */
            LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
            if (TransactionIdFollowsOrEquals(subxid,
                                             ShmemVariableCache->nextXid))
            {
                ShmemVariableCache->nextXid = subxid;
                TransactionIdAdvance(ShmemVariableCache->nextXid);
            }
            LWLockRelease(XidGenLock);
        }

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

    return buf;
}

static char* ReadTwoPhaseFile ( TransactionId  xid, bool  give_warnings  )

static

Definition at line 1185 of file twophase.c.

References buf, CloseTransientFile(), COMP_CRC32C, EQ_CRC32C, ereport, errcode_for_file_access(), errmsg(), fd(), FIN_CRC32C, INIT_CRC32C, TwoPhaseFileHeader::magic, MAXALIGN, MaxAllocSize, MAXPGPATH, OpenTransientFile(), palloc(), pfree(), PG_BINARY, pgstat_report_wait_end(), pgstat_report_wait_start(), read, TwoPhaseFileHeader::total_len, TWOPHASE_MAGIC, TwoPhaseFilePath, WAIT_EVENT_TWOPHASE_FILE_READ, and WARNING.

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

{
    char        path[MAXPGPATH];
    char       *buf;
    TwoPhaseFileHeader *hdr;
    int         fd;
    struct stat stat;
    uint32      crc_offset;
    pg_crc32c   calc_crc,
                file_crc;

    TwoPhaseFilePath(path, xid);

    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0);
    if (fd < 0)
    {
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not open two-phase state file \"%s\": %m",
                            path)));
        return NULL;
    }

    /*
     * 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))
    {
        CloseTransientFile(fd);
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not stat two-phase state file \"%s\": %m",
                            path)));
        return NULL;
    }

    if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                        MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
                        sizeof(pg_crc32c)) ||
        stat.st_size > MaxAllocSize)
    {
        CloseTransientFile(fd);
        return NULL;
    }

    crc_offset = stat.st_size - sizeof(pg_crc32c);
    if (crc_offset != MAXALIGN(crc_offset))
    {
        CloseTransientFile(fd);
        return NULL;
    }

    /*
     * OK, slurp in the file.
     */
    buf = (char *) palloc(stat.st_size);

    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_READ);
    if (read(fd, buf, stat.st_size) != stat.st_size)
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        if (give_warnings)
            ereport(WARNING,
                    (errcode_for_file_access(),
                     errmsg("could not read two-phase state file \"%s\": %m",
                            path)));
        pfree(buf);
        return NULL;
    }

    pgstat_report_wait_end();
    CloseTransientFile(fd);

    hdr = (TwoPhaseFileHeader *) buf;
    if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
    {
        pfree(buf);
        return NULL;
    }

    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))
    {
        pfree(buf);
        return NULL;
    }

    return buf;
}

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

static

Definition at line 2255 of file twophase.c.

References elog, END_CRIT_SECTION, GetCurrentTimestamp(), MyXactFlags, PANIC, START_CRIT_SECTION, SyncRepWaitForLSN(), TransactionIdAbortTree(), TransactionIdDidCommit(), XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK, XactLogAbortRecord(), and XLogFlush().

Referenced by FinishPreparedTransaction().

{
    XLogRecPtr  recptr;

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

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

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

static

Definition at line 2162 of file twophase.c.

References PGXACT::delayChkpt, DoNotReplicateId, END_CRIT_SECTION, GetCurrentTimestamp(), InvalidRepOriginId, MyPgXact, 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 */
    MyPgXact->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, false,
                                 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
                                 xid);


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

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

void RecoverPreparedTransactions

(

void

)

Definition at line 1912 of file twophase.c.

References Assert, buf, TwoPhaseFileHeader::database, ereport, errmsg(), TwoPhaseFileHeader::gidlen, GXactLoadSubxactData(), i, InHotStandby, GlobalTransactionData::inredo, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MarkAsPrepared(), MarkAsPreparingGuts(), MAXALIGN, TwoPhaseFileHeader::nabortrels, TwoPhaseFileHeader::ncommitrels, TwoPhaseFileHeader::ninvalmsgs, TwoPhaseFileHeader::nsubxacts, TwoPhaseStateData::numPrepXacts, GlobalTransactionData::ondisk, TwoPhaseFileHeader::owner, pfree(), PostPrepare_Twophase(), GlobalTransactionData::prepare_start_lsn, TwoPhaseFileHeader::prepared_at, TwoPhaseStateData::prepXacts, ProcessRecords(), ProcessTwoPhaseBuffer(), StandbyReleaseLockTree(), TransactionIdEquals, twophase_recover_callbacks, GlobalTransactionData::xid, and TwoPhaseFileHeader::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);
}

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

static

Definition at line 1570 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,
                           S_IRUSR | S_IWUSR);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not recreate two-phase state file \"%s\": %m",
                        path)));

    /* Write content and CRC */
    pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_WRITE);
    if (write(fd, content, len) != len)
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write two-phase state file: %m")));
    }
    if (write(fd, &statefile_crc, sizeof(pg_crc32c)) != sizeof(pg_crc32c))
    {
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not write two-phase state file: %m")));
    }
    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)
    {
        CloseTransientFile(fd);
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not fsync two-phase state file: %m")));
    }
    pgstat_report_wait_end();

    if (CloseTransientFile(fd) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not close two-phase state file: %m")));
}

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

Definition at line 1164 of file twophase.c.

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

Referenced by AtPrepare_Locks(), AtPrepare_MultiXact(), AtPrepare_PgStat(), 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);
}

static void RemoveGXact ( GlobalTransaction  gxact )

static

Definition at line 635 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);
}

static void RemoveTwoPhaseFile ( TransactionId  xid, bool  giveWarning  )

static

Definition at line 1550 of file twophase.c.

References ereport, errcode_for_file_access(), errmsg(), MAXPGPATH, TwoPhaseFilePath, unlink(), 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 two-phase state file \"%s\": %m",
                            path)));
}

void restoreTwoPhaseData

(

void

)

Definition at line 1734 of file twophase.c.

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

Referenced by StartupXLOG().

{
    DIR        *cldir;
    struct dirent *clde;

    cldir = AllocateDir(TWOPHASE_DIR);
    LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    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);
        }
    }
    LWLockRelease(TwoPhaseStateLock);
    FreeDir(cldir);
}

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

static

Definition at line 963 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;
}

void StandbyRecoverPreparedTransactions

(

void

)

Definition at line 1871 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);
}

bool StandbyTransactionIdIsPrepared

(

TransactionId

xid

)

Definition at line 1340 of file twophase.c.

References Assert, buf, max_prepared_xacts, pfree(), ReadTwoPhaseFile(), result, TransactionIdEquals, TransactionIdIsValid, and TwoPhaseFileHeader::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, false);
    if (buf == NULL)
        return false;

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

    return result;
}

void StartPrepare

(

GlobalTransaction

gxact

)

Definition at line 991 of file twophase.c.

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

Referenced by PrepareTransaction().

{
    PGPROC     *proc = &ProcGlobal->allProcs[gxact->pgprocno];
    PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
    TransactionId xid = pgxact->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);
    }
}

BackendId TwoPhaseGetDummyBackendId

(

TransactionId

xid

)

Definition at line 856 of file twophase.c.

References GlobalTransactionData::dummyBackendId, and TwoPhaseGetGXact().

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

{
    GlobalTransaction gxact = TwoPhaseGetGXact(xid);

    return gxact->dummyBackendId;
}

PGPROC* TwoPhaseGetDummyProc

(

TransactionId

xid

)

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

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

static GlobalTransaction TwoPhaseGetGXact ( TransactionId  xid )

static

Definition at line 807 of file twophase.c.

References PROC_HDR::allPgXact, elog, ERROR, i, InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), TwoPhaseStateData::numPrepXacts, GlobalTransactionData::pgprocno, TwoPhaseStateData::prepXacts, ProcGlobal, result, and PGXACT::xid.

Referenced by TwoPhaseGetDummyBackendId(), and TwoPhaseGetDummyProc().

{
    GlobalTransaction result = NULL;
    int         i;

    static TransactionId cached_xid = InvalidTransactionId;
    static GlobalTransaction cached_gxact = NULL;

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

    LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

    for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
    {
        GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
        PGXACT     *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];

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

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

void TwoPhaseShmemInit

(

void

)

Definition at line 254 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);
}

Size TwoPhaseShmemSize

(

void

)

Definition at line 238 of file twophase.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), 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;
}

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

static

Definition at line 1296 of file twophase.c.

References ereport, errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, palloc(), read_local_xlog_page(), wal_segment_size, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogReaderAllocate(), XLogReaderFree(), XLogReadRecord(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetInfo, and XLogRecGetRmid.

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

{
    XLogRecord *record;
    XLogReaderState *xlogreader;
    char       *errormsg;

    xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,
                                    NULL);
    if (!xlogreader)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("out of memory"),
                 errdetail("Failed while allocating a WAL reading processor.")));

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

int max_prepared_xacts = 0

Definition at line 117 of file twophase.c.

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

GlobalTransaction MyLockedGxact = NULL

static

Definition at line 203 of file twophase.c.

struct xllist records

static

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

bool twophaseExitRegistered = false

static

Definition at line 205 of file twophase.c.

Referenced by LockGXact(), and MarkAsPreparing().

TwoPhaseStateData* TwoPhaseState

static

Definition at line 195 of file twophase.c.