mirror of
https://git.postgresql.org/git/postgresql.git
synced 2024-08-16 18:03:22 +02:00
When the "hot" members of PGPROC were split off to separate PGXACT structs, many PGPROC fields referred to in comments were moved to PGXACT, but the comments were neglected in the commit. Mostly this is just a search/replace of PGPROC with PGXACT, but the way the dummy PGPROC entries are created for prepared transactions changed more, making some of the comments totally bogus. Noah Misch
2161 lines
59 KiB
C
2161 lines
59 KiB
C
/*-------------------------------------------------------------------------
|
|
*
|
|
* twophase.c
|
|
* Two-phase commit support functions.
|
|
*
|
|
* Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
|
|
* Portions Copyright (c) 1994, Regents of the University of California
|
|
*
|
|
* IDENTIFICATION
|
|
* src/backend/access/transam/twophase.c
|
|
*
|
|
* NOTES
|
|
* Each global transaction is associated with a global transaction
|
|
* identifier (GID). The client assigns a GID to a postgres
|
|
* transaction with the PREPARE TRANSACTION command.
|
|
*
|
|
* We keep all active global transactions in a shared memory array.
|
|
* When the PREPARE TRANSACTION command is issued, the GID is
|
|
* reserved for the transaction in the array. This is done before
|
|
* a WAL entry is made, because the reservation checks for duplicate
|
|
* GIDs and aborts the transaction if there already is a global
|
|
* transaction in prepared state with the same GID.
|
|
*
|
|
* A global transaction (gxact) also has dummy PGXACT and PGPROC; this is
|
|
* what keeps the XID considered running by TransactionIdIsInProgress.
|
|
* It is also convenient as a PGPROC to hook the gxact's locks to.
|
|
*
|
|
* In order to survive crashes and shutdowns, all prepared
|
|
* transactions must be stored in permanent storage. This includes
|
|
* locking information, pending notifications etc. All that state
|
|
* information is written to the per-transaction state file in
|
|
* the pg_twophase directory.
|
|
*
|
|
*-------------------------------------------------------------------------
|
|
*/
|
|
#include "postgres.h"
|
|
|
|
#include <fcntl.h>
|
|
#include <sys/stat.h>
|
|
#include <sys/types.h>
|
|
#include <time.h>
|
|
#include <unistd.h>
|
|
|
|
#include "access/htup.h"
|
|
#include "access/subtrans.h"
|
|
#include "access/transam.h"
|
|
#include "access/twophase.h"
|
|
#include "access/twophase_rmgr.h"
|
|
#include "access/xact.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/walsender.h"
|
|
#include "replication/syncrep.h"
|
|
#include "storage/fd.h"
|
|
#include "storage/predicate.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"
|
|
|
|
|
|
/*
|
|
* Directory where Two-phase commit files reside within PGDATA
|
|
*/
|
|
#define TWOPHASE_DIR "pg_twophase"
|
|
|
|
/* GUC variable, can't be changed after startup */
|
|
int max_prepared_xacts = 0;
|
|
|
|
/*
|
|
* This struct describes one global transaction that is in prepared state
|
|
* or attempting to become prepared.
|
|
*
|
|
* The lifecycle of a global transaction is:
|
|
*
|
|
* 1. After checking that the requested GID is not in use, set up an
|
|
* entry in the TwoPhaseState->prepXacts array with the correct XID and GID,
|
|
* with locking_xid = my own XID and valid = false.
|
|
*
|
|
* 2. After successfully completing prepare, set valid = true and enter the
|
|
* referenced PGPROC into the global ProcArray.
|
|
*
|
|
* 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry
|
|
* is valid and its locking_xid is no longer active, then store my current
|
|
* XID into locking_xid. This prevents concurrent attempts to commit or
|
|
* rollback the same prepared xact.
|
|
*
|
|
* 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
|
|
* from the ProcArray and the TwoPhaseState->prepXacts array and return it to
|
|
* the freelist.
|
|
*
|
|
* Note that if the preparing transaction fails between steps 1 and 2, the
|
|
* entry will remain in prepXacts until recycled. We can detect recyclable
|
|
* entries by checking for valid = false and locking_xid no longer active.
|
|
*
|
|
* typedef struct GlobalTransactionData *GlobalTransaction appears in
|
|
* twophase.h
|
|
*/
|
|
#define GIDSIZE 200
|
|
|
|
typedef struct GlobalTransactionData
|
|
{
|
|
GlobalTransaction next;
|
|
int pgprocno; /* dummy proc */
|
|
BackendId dummyBackendId; /* similar to backend id for backends */
|
|
TimestampTz prepared_at; /* time of preparation */
|
|
XLogRecPtr prepare_lsn; /* XLOG offset of prepare record */
|
|
Oid owner; /* ID of user that executed the xact */
|
|
TransactionId locking_xid; /* top-level XID of backend working on xact */
|
|
bool valid; /* TRUE if fully prepared */
|
|
char gid[GIDSIZE]; /* The GID assigned to the prepared xact */
|
|
} GlobalTransactionData;
|
|
|
|
/*
|
|
* Two Phase Commit shared state. Access to this struct is protected
|
|
* by TwoPhaseStateLock.
|
|
*/
|
|
typedef struct TwoPhaseStateData
|
|
{
|
|
/* Head of linked list of free GlobalTransactionData structs */
|
|
GlobalTransaction freeGXacts;
|
|
|
|
/* Number of valid prepXacts entries. */
|
|
int numPrepXacts;
|
|
|
|
/*
|
|
* There are max_prepared_xacts items in this array, but C wants a
|
|
* fixed-size array.
|
|
*/
|
|
GlobalTransaction prepXacts[1]; /* VARIABLE LENGTH ARRAY */
|
|
} TwoPhaseStateData; /* VARIABLE LENGTH STRUCT */
|
|
|
|
static TwoPhaseStateData *TwoPhaseState;
|
|
|
|
|
|
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[]);
|
|
|
|
|
|
/*
|
|
* Initialization of shared memory
|
|
*/
|
|
Size
|
|
TwoPhaseShmemSize(void)
|
|
{
|
|
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;
|
|
}
|
|
|
|
void
|
|
TwoPhaseShmemInit(void)
|
|
{
|
|
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++)
|
|
{
|
|
gxacts[i].pgprocno = PreparedXactProcs[i].pgprocno;
|
|
gxacts[i].next = TwoPhaseState->freeGXacts;
|
|
TwoPhaseState->freeGXacts = &gxacts[i];
|
|
|
|
/*
|
|
* 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);
|
|
}
|
|
|
|
|
|
/*
|
|
* MarkAsPreparing
|
|
* Reserve the GID for the given transaction.
|
|
*
|
|
* Internally, this creates a gxact struct and puts it into the active array.
|
|
* NOTE: this is also used when reloading a gxact after a crash; so avoid
|
|
* assuming that we can use very much backend context.
|
|
*/
|
|
GlobalTransaction
|
|
MarkAsPreparing(TransactionId xid, const char *gid,
|
|
TimestampTz prepared_at, Oid owner, Oid databaseid)
|
|
{
|
|
GlobalTransaction gxact;
|
|
PGPROC *proc;
|
|
PGXACT *pgxact;
|
|
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.")));
|
|
|
|
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
|
|
|
|
/*
|
|
* First, find and recycle any gxacts that failed during prepare. We do
|
|
* this partly to ensure we don't mistakenly say their GIDs are still
|
|
* reserved, and partly so we don't fail on out-of-slots unnecessarily.
|
|
*/
|
|
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
|
|
{
|
|
gxact = TwoPhaseState->prepXacts[i];
|
|
if (!gxact->valid && !TransactionIdIsActive(gxact->locking_xid))
|
|
{
|
|
/* It's dead Jim ... 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;
|
|
/* Back up index count too, so we don't miss scanning one */
|
|
i--;
|
|
}
|
|
}
|
|
|
|
/* 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 = (GlobalTransaction) gxact->next;
|
|
|
|
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->inCommit = false;
|
|
pgxact->vacuumFlags = 0;
|
|
proc->pid = 0;
|
|
proc->backendId = InvalidBackendId;
|
|
proc->databaseId = databaseid;
|
|
proc->roleId = owner;
|
|
proc->lwWaiting = false;
|
|
proc->lwWaitMode = 0;
|
|
proc->lwWaitLink = NULL;
|
|
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;
|
|
/* initialize LSN to 0 (start of WAL) */
|
|
gxact->prepare_lsn.xlogid = 0;
|
|
gxact->prepare_lsn.xrecoff = 0;
|
|
gxact->owner = owner;
|
|
gxact->locking_xid = xid;
|
|
gxact->valid = false;
|
|
strcpy(gxact->gid, gid);
|
|
|
|
/* And insert it into the active array */
|
|
Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
|
|
TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
return gxact;
|
|
}
|
|
|
|
/*
|
|
* GXactLoadSubxactData
|
|
*
|
|
* If the transaction being persisted had any subtransactions, this must
|
|
* be called before MarkAsPrepared() to load information into the dummy
|
|
* PGPROC.
|
|
*/
|
|
static void
|
|
GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
|
|
TransactionId *children)
|
|
{
|
|
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;
|
|
}
|
|
}
|
|
|
|
/*
|
|
* MarkAsPrepared
|
|
* Mark the GXACT as fully valid, and enter it into the global ProcArray.
|
|
*/
|
|
static void
|
|
MarkAsPrepared(GlobalTransaction gxact)
|
|
{
|
|
/* Lock here may be overkill, but I'm not convinced of that ... */
|
|
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
|
|
Assert(!gxact->valid);
|
|
gxact->valid = true;
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
/*
|
|
* Put it into the global ProcArray so TransactionIdIsInProgress considers
|
|
* the XID as still running.
|
|
*/
|
|
ProcArrayAdd(&ProcGlobal->allProcs[gxact->pgprocno]);
|
|
}
|
|
|
|
/*
|
|
* LockGXact
|
|
* Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
|
|
*/
|
|
static GlobalTransaction
|
|
LockGXact(const char *gid, Oid user)
|
|
{
|
|
int i;
|
|
|
|
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 (TransactionIdIsValid(gxact->locking_xid))
|
|
{
|
|
if (TransactionIdIsActive(gxact->locking_xid))
|
|
ereport(ERROR,
|
|
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
|
|
errmsg("prepared transaction with identifier \"%s\" is busy",
|
|
gid)));
|
|
gxact->locking_xid = InvalidTransactionId;
|
|
}
|
|
|
|
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_xid = GetTopTransactionId();
|
|
|
|
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;
|
|
}
|
|
|
|
/*
|
|
* RemoveGXact
|
|
* Remove the prepared transaction from the shared memory array.
|
|
*
|
|
* NB: caller should have already removed it from ProcArray
|
|
*/
|
|
static void
|
|
RemoveGXact(GlobalTransaction gxact)
|
|
{
|
|
int i;
|
|
|
|
LWLockAcquire(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;
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
return;
|
|
}
|
|
}
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
|
|
}
|
|
|
|
/*
|
|
* TransactionIdIsPrepared
|
|
* True iff transaction associated with the identifier is prepared
|
|
* for two-phase commit
|
|
*
|
|
* Note: only gxacts marked "valid" are considered; but notice we do not
|
|
* check the locking status.
|
|
*
|
|
* This is not currently exported, because it is only needed internally.
|
|
*/
|
|
static bool
|
|
TransactionIdIsPrepared(TransactionId xid)
|
|
{
|
|
bool result = false;
|
|
int i;
|
|
|
|
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
|
|
|
|
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
|
|
{
|
|
GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
|
|
PGXACT *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
|
|
|
|
if (gxact->valid && pgxact->xid == xid)
|
|
{
|
|
result = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
return result;
|
|
}
|
|
|
|
/*
|
|
* Returns an array of all prepared transactions for the user-level
|
|
* function pg_prepared_xact.
|
|
*
|
|
* The returned array and all its elements are copies of internal data
|
|
* structures, to minimize the time we need to hold the TwoPhaseStateLock.
|
|
*
|
|
* WARNING -- we return even those transactions that are not fully prepared
|
|
* yet. The caller should filter them out if he doesn't want them.
|
|
*
|
|
* The returned array is palloc'd.
|
|
*/
|
|
static int
|
|
GetPreparedTransactionList(GlobalTransaction *gxacts)
|
|
{
|
|
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;
|
|
}
|
|
|
|
|
|
/* Working status for pg_prepared_xact */
|
|
typedef struct
|
|
{
|
|
GlobalTransaction array;
|
|
int ngxacts;
|
|
int currIdx;
|
|
} Working_State;
|
|
|
|
/*
|
|
* pg_prepared_xact
|
|
* Produce a view with one row per prepared transaction.
|
|
*
|
|
* This function is here so we don't have to export the
|
|
* GlobalTransactionData struct definition.
|
|
*/
|
|
Datum
|
|
pg_prepared_xact(PG_FUNCTION_ARGS)
|
|
{
|
|
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);
|
|
}
|
|
|
|
/*
|
|
* TwoPhaseGetDummyProc
|
|
* Get the dummy backend ID for prepared transaction specified by XID
|
|
*
|
|
* Dummy backend IDs are similar to real backend IDs of real backends.
|
|
* They start at MaxBackends + 1, and are unique across all currently active
|
|
* real backends and prepared transactions.
|
|
*/
|
|
BackendId
|
|
TwoPhaseGetDummyBackendId(TransactionId xid)
|
|
{
|
|
PGPROC *proc = TwoPhaseGetDummyProc(xid);
|
|
|
|
return ((GlobalTransaction) proc)->dummyBackendId;
|
|
}
|
|
|
|
/*
|
|
* TwoPhaseGetDummyProc
|
|
* Get the PGPROC that represents a prepared transaction specified by XID
|
|
*/
|
|
PGPROC *
|
|
TwoPhaseGetDummyProc(TransactionId xid)
|
|
{
|
|
PGPROC *result = NULL;
|
|
int i;
|
|
|
|
static TransactionId cached_xid = InvalidTransactionId;
|
|
static PGPROC *cached_proc = 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_proc;
|
|
|
|
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 = &ProcGlobal->allProcs[gxact->pgprocno];
|
|
break;
|
|
}
|
|
}
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
if (result == NULL) /* should not happen */
|
|
elog(ERROR, "failed to find dummy PGPROC for xid %u", xid);
|
|
|
|
cached_xid = xid;
|
|
cached_proc = result;
|
|
|
|
return result;
|
|
}
|
|
|
|
/************************************************************************/
|
|
/* State file support */
|
|
/************************************************************************/
|
|
|
|
#define TwoPhaseFilePath(path, xid) \
|
|
snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)
|
|
|
|
/*
|
|
* 2PC state file format:
|
|
*
|
|
* 1. TwoPhaseFileHeader
|
|
* 2. TransactionId[] (subtransactions)
|
|
* 3. RelFileNode[] (files to be deleted at commit)
|
|
* 4. RelFileNode[] (files to be deleted at abort)
|
|
* 5. SharedInvalidationMessage[] (inval messages to be sent at commit)
|
|
* 6. TwoPhaseRecordOnDisk
|
|
* 7. ...
|
|
* 8. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
|
|
* 9. CRC32
|
|
*
|
|
* Each segment except the final CRC32 is MAXALIGN'd.
|
|
*/
|
|
|
|
/*
|
|
* Header for a 2PC state file
|
|
*/
|
|
#define TWOPHASE_MAGIC 0x57F94532 /* format identifier */
|
|
|
|
typedef struct TwoPhaseFileHeader
|
|
{
|
|
uint32 magic; /* format identifier */
|
|
uint32 total_len; /* actual file length */
|
|
TransactionId xid; /* original transaction XID */
|
|
Oid database; /* OID of database it was in */
|
|
TimestampTz prepared_at; /* time of preparation */
|
|
Oid owner; /* user running the transaction */
|
|
int32 nsubxacts; /* number of following subxact XIDs */
|
|
int32 ncommitrels; /* number of delete-on-commit rels */
|
|
int32 nabortrels; /* number of delete-on-abort rels */
|
|
int32 ninvalmsgs; /* number of cache invalidation messages */
|
|
bool initfileinval; /* does relcache init file need invalidation? */
|
|
char gid[GIDSIZE]; /* GID for transaction */
|
|
} TwoPhaseFileHeader;
|
|
|
|
/*
|
|
* Header for each record in a state file
|
|
*
|
|
* NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
|
|
* The rmgr data will be stored starting on a MAXALIGN boundary.
|
|
*/
|
|
typedef struct TwoPhaseRecordOnDisk
|
|
{
|
|
uint32 len; /* length of rmgr data */
|
|
TwoPhaseRmgrId rmid; /* resource manager for this record */
|
|
uint16 info; /* flag bits for use by rmgr */
|
|
} TwoPhaseRecordOnDisk;
|
|
|
|
/*
|
|
* During prepare, the state file is assembled in memory before writing it
|
|
* to WAL and the actual state file. We use a chain of XLogRecData blocks
|
|
* so that we will be able to pass the state file contents directly to
|
|
* XLogInsert.
|
|
*/
|
|
static struct xllist
|
|
{
|
|
XLogRecData *head; /* first data block in the chain */
|
|
XLogRecData *tail; /* last block in chain */
|
|
uint32 bytes_free; /* free bytes left in tail block */
|
|
uint32 total_len; /* total data bytes in chain */
|
|
} records;
|
|
|
|
|
|
/*
|
|
* Append a block of data to records data structure.
|
|
*
|
|
* NB: each block is padded to a MAXALIGN multiple. This must be
|
|
* accounted for when the file is later read!
|
|
*
|
|
* The data is copied, so the caller is free to modify it afterwards.
|
|
*/
|
|
static void
|
|
save_state_data(const void *data, uint32 len)
|
|
{
|
|
uint32 padlen = MAXALIGN(len);
|
|
|
|
if (padlen > records.bytes_free)
|
|
{
|
|
records.tail->next = palloc0(sizeof(XLogRecData));
|
|
records.tail = records.tail->next;
|
|
records.tail->buffer = InvalidBuffer;
|
|
records.tail->len = 0;
|
|
records.tail->next = NULL;
|
|
|
|
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;
|
|
}
|
|
|
|
/*
|
|
* Start preparing a state file.
|
|
*
|
|
* Initializes data structure and inserts the 2PC file header record.
|
|
*/
|
|
void
|
|
StartPrepare(GlobalTransaction gxact)
|
|
{
|
|
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(XLogRecData));
|
|
records.head->buffer = InvalidBuffer;
|
|
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.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);
|
|
StrNCpy(hdr.gid, gxact->gid, GIDSIZE);
|
|
|
|
save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
|
|
|
|
/*
|
|
* 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);
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Finish preparing state file.
|
|
*
|
|
* Calculates CRC and writes state file to WAL and in pg_twophase directory.
|
|
*/
|
|
void
|
|
EndPrepare(GlobalTransaction gxact)
|
|
{
|
|
PGXACT *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
|
|
TransactionId xid = pgxact->xid;
|
|
TwoPhaseFileHeader *hdr;
|
|
char path[MAXPGPATH];
|
|
XLogRecData *record;
|
|
pg_crc32 statefile_crc;
|
|
pg_crc32 bogus_crc;
|
|
int fd;
|
|
|
|
/* 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_crc32);
|
|
|
|
/*
|
|
* If the file size exceeds MaxAllocSize, we won't be able to read it in
|
|
* ReadTwoPhaseFile. Check for that now, rather than fail at commit time.
|
|
*/
|
|
if (hdr->total_len > MaxAllocSize)
|
|
ereport(ERROR,
|
|
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
|
|
errmsg("two-phase state file maximum length exceeded")));
|
|
|
|
/*
|
|
* Create the 2PC state file.
|
|
*
|
|
* Note: because we use BasicOpenFile(), we are responsible for ensuring
|
|
* the FD gets closed in any error exit path. Once we get into the
|
|
* critical section, though, it doesn't matter since any failure causes
|
|
* PANIC anyway.
|
|
*/
|
|
TwoPhaseFilePath(path, xid);
|
|
|
|
fd = BasicOpenFile(path,
|
|
O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
|
|
S_IRUSR | S_IWUSR);
|
|
if (fd < 0)
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not create two-phase state file \"%s\": %m",
|
|
path)));
|
|
|
|
/* Write data to file, and calculate CRC as we pass over it */
|
|
INIT_CRC32(statefile_crc);
|
|
|
|
for (record = records.head; record != NULL; record = record->next)
|
|
{
|
|
COMP_CRC32(statefile_crc, record->data, record->len);
|
|
if ((write(fd, record->data, record->len)) != record->len)
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not write two-phase state file: %m")));
|
|
}
|
|
}
|
|
|
|
FIN_CRC32(statefile_crc);
|
|
|
|
/*
|
|
* Write a deliberately bogus CRC to the state file; this is just paranoia
|
|
* to catch the case where four more bytes will run us out of disk space.
|
|
*/
|
|
bogus_crc = ~statefile_crc;
|
|
|
|
if ((write(fd, &bogus_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not write two-phase state file: %m")));
|
|
}
|
|
|
|
/* Back up to prepare for rewriting the CRC */
|
|
if (lseek(fd, -((off_t) sizeof(pg_crc32)), SEEK_CUR) < 0)
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not seek in two-phase state file: %m")));
|
|
}
|
|
|
|
/*
|
|
* The state file isn't valid yet, because we haven't written the correct
|
|
* CRC yet. Before we do that, insert entry in WAL and flush it to disk.
|
|
*
|
|
* Between the time we have written the WAL entry and the time we write
|
|
* out the correct state file CRC, we have an inconsistency: the xact is
|
|
* prepared according to WAL but not according to our on-disk state. We
|
|
* use a critical section to force a PANIC if we are unable to complete
|
|
* the write --- then, WAL replay should repair the inconsistency. The
|
|
* odds of a PANIC actually occurring should be very tiny given that we
|
|
* were able to write the bogus CRC above.
|
|
*
|
|
* We have to set inCommit 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 inCommit for; see notes there.)
|
|
*
|
|
* We save the PREPARE record's location in the gxact for later use by
|
|
* CheckPointTwoPhase.
|
|
*/
|
|
START_CRIT_SECTION();
|
|
|
|
MyPgXact->inCommit = true;
|
|
|
|
gxact->prepare_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE,
|
|
records.head);
|
|
XLogFlush(gxact->prepare_lsn);
|
|
|
|
/* If we crash now, we have prepared: WAL replay will fix things */
|
|
|
|
/*
|
|
* Wake up all walsenders to send WAL up to the PREPARE record immediately
|
|
* if replication is enabled
|
|
*/
|
|
if (max_wal_senders > 0)
|
|
WalSndWakeup();
|
|
|
|
/* write correct CRC and close file */
|
|
if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not write two-phase state file: %m")));
|
|
}
|
|
|
|
if (close(fd) != 0)
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not close two-phase state file: %m")));
|
|
|
|
/*
|
|
* 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);
|
|
|
|
/*
|
|
* 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->inCommit = false;
|
|
|
|
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_lsn);
|
|
|
|
records.tail = records.head = NULL;
|
|
}
|
|
|
|
/*
|
|
* Register a 2PC record to be written to state file.
|
|
*/
|
|
void
|
|
RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
|
|
const void *data, uint32 len)
|
|
{
|
|
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);
|
|
}
|
|
|
|
|
|
/*
|
|
* Read and validate the state file for xid.
|
|
*
|
|
* If it looks OK (has a valid magic number and CRC), return the palloc'd
|
|
* contents of the file. Otherwise return NULL.
|
|
*/
|
|
static char *
|
|
ReadTwoPhaseFile(TransactionId xid, bool give_warnings)
|
|
{
|
|
char path[MAXPGPATH];
|
|
char *buf;
|
|
TwoPhaseFileHeader *hdr;
|
|
int fd;
|
|
struct stat stat;
|
|
uint32 crc_offset;
|
|
pg_crc32 calc_crc,
|
|
file_crc;
|
|
|
|
TwoPhaseFilePath(path, xid);
|
|
|
|
fd = BasicOpenFile(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))
|
|
{
|
|
close(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_crc32)) ||
|
|
stat.st_size > MaxAllocSize)
|
|
{
|
|
close(fd);
|
|
return NULL;
|
|
}
|
|
|
|
crc_offset = stat.st_size - sizeof(pg_crc32);
|
|
if (crc_offset != MAXALIGN(crc_offset))
|
|
{
|
|
close(fd);
|
|
return NULL;
|
|
}
|
|
|
|
/*
|
|
* OK, slurp in the file.
|
|
*/
|
|
buf = (char *) palloc(stat.st_size);
|
|
|
|
if (read(fd, buf, stat.st_size) != stat.st_size)
|
|
{
|
|
close(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;
|
|
}
|
|
|
|
close(fd);
|
|
|
|
hdr = (TwoPhaseFileHeader *) buf;
|
|
if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
|
|
{
|
|
pfree(buf);
|
|
return NULL;
|
|
}
|
|
|
|
INIT_CRC32(calc_crc);
|
|
COMP_CRC32(calc_crc, buf, crc_offset);
|
|
FIN_CRC32(calc_crc);
|
|
|
|
file_crc = *((pg_crc32 *) (buf + crc_offset));
|
|
|
|
if (!EQ_CRC32(calc_crc, file_crc))
|
|
{
|
|
pfree(buf);
|
|
return NULL;
|
|
}
|
|
|
|
return buf;
|
|
}
|
|
|
|
/*
|
|
* Confirms an xid is prepared, during recovery
|
|
*/
|
|
bool
|
|
StandbyTransactionIdIsPrepared(TransactionId xid)
|
|
{
|
|
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;
|
|
}
|
|
|
|
/*
|
|
* FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
|
|
*/
|
|
void
|
|
FinishPreparedTransaction(const char *gid, bool isCommit)
|
|
{
|
|
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 the state file
|
|
*/
|
|
buf = ReadTwoPhaseFile(xid, true);
|
|
if (buf == NULL)
|
|
ereport(ERROR,
|
|
(errcode(ERRCODE_DATA_CORRUPTED),
|
|
errmsg("two-phase state file for transaction %u is corrupt",
|
|
xid)));
|
|
|
|
/*
|
|
* Disassemble the header area
|
|
*/
|
|
hdr = (TwoPhaseFileHeader *) buf;
|
|
Assert(TransactionIdEquals(hdr->xid, xid));
|
|
bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
|
|
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_clog, 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 can be recycled
|
|
* properly later. It is still locked by our XID 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);
|
|
ForkNumber fork;
|
|
|
|
for (fork = 0; fork <= MAX_FORKNUM; fork++)
|
|
{
|
|
smgrdounlink(srel, fork, 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 our mess.
|
|
*/
|
|
RemoveTwoPhaseFile(xid, true);
|
|
|
|
RemoveGXact(gxact);
|
|
|
|
pfree(buf);
|
|
}
|
|
|
|
/*
|
|
* Scan a 2PC state file (already read into memory by ReadTwoPhaseFile)
|
|
* and call the indicated callbacks for each 2PC record.
|
|
*/
|
|
static void
|
|
ProcessRecords(char *bufptr, TransactionId xid,
|
|
const TwoPhaseCallback callbacks[])
|
|
{
|
|
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);
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Remove the 2PC file for the specified XID.
|
|
*
|
|
* If giveWarning is false, do not complain about file-not-present;
|
|
* this is an expected case during WAL replay.
|
|
*/
|
|
void
|
|
RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
|
|
{
|
|
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)));
|
|
}
|
|
|
|
/*
|
|
* Recreates a state file. This is used in WAL replay.
|
|
*
|
|
* Note: content and len don't include CRC.
|
|
*/
|
|
void
|
|
RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
|
|
{
|
|
char path[MAXPGPATH];
|
|
pg_crc32 statefile_crc;
|
|
int fd;
|
|
|
|
/* Recompute CRC */
|
|
INIT_CRC32(statefile_crc);
|
|
COMP_CRC32(statefile_crc, content, len);
|
|
FIN_CRC32(statefile_crc);
|
|
|
|
TwoPhaseFilePath(path, xid);
|
|
|
|
fd = BasicOpenFile(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 */
|
|
if (write(fd, content, len) != len)
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not write two-phase state file: %m")));
|
|
}
|
|
if (write(fd, &statefile_crc, sizeof(pg_crc32)) != sizeof(pg_crc32))
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not write two-phase state file: %m")));
|
|
}
|
|
|
|
/*
|
|
* 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.
|
|
*/
|
|
if (pg_fsync(fd) != 0)
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not fsync two-phase state file: %m")));
|
|
}
|
|
|
|
if (close(fd) != 0)
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not close two-phase state file: %m")));
|
|
}
|
|
|
|
/*
|
|
* CheckPointTwoPhase -- handle 2PC component of checkpointing.
|
|
*
|
|
* We must fsync the state file of any GXACT that is valid and has a PREPARE
|
|
* LSN <= the checkpoint's redo horizon. (If the gxact isn't valid yet or
|
|
* has a later LSN, this checkpoint is not responsible for fsyncing it.)
|
|
*
|
|
* This is deliberately run as late as possible in the checkpoint sequence,
|
|
* because GXACTs ordinarily have short lifespans, and so it is quite
|
|
* possible that GXACTs that were valid at checkpoint start will no longer
|
|
* exist if we wait a little bit.
|
|
*
|
|
* If a GXACT remains valid across multiple checkpoints, it'll be fsynced
|
|
* each time. This is considered unusual enough that we don't bother to
|
|
* expend any extra code to avoid the redundant fsyncs. (They should be
|
|
* reasonably cheap anyway, since they won't cause I/O.)
|
|
*/
|
|
void
|
|
CheckPointTwoPhase(XLogRecPtr redo_horizon)
|
|
{
|
|
TransactionId *xids;
|
|
int nxids;
|
|
char path[MAXPGPATH];
|
|
int i;
|
|
|
|
/*
|
|
* We don't want to hold the TwoPhaseStateLock while doing I/O, so we grab
|
|
* it just long enough to make a list of the XIDs that require fsyncing,
|
|
* and then do the I/O afterwards.
|
|
*
|
|
* This approach creates a race condition: someone else could delete a
|
|
* GXACT between the time we release TwoPhaseStateLock and the time we try
|
|
* to open its state file. We handle this by special-casing ENOENT
|
|
* failures: if we see that, we verify that the GXACT is no longer valid,
|
|
* and if so ignore the failure.
|
|
*/
|
|
if (max_prepared_xacts <= 0)
|
|
return; /* nothing to do */
|
|
|
|
TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();
|
|
|
|
xids = (TransactionId *) palloc(max_prepared_xacts * sizeof(TransactionId));
|
|
nxids = 0;
|
|
|
|
LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
|
|
|
|
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
|
|
{
|
|
GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
|
|
PGXACT *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
|
|
|
|
if (gxact->valid &&
|
|
XLByteLE(gxact->prepare_lsn, redo_horizon))
|
|
xids[nxids++] = pgxact->xid;
|
|
}
|
|
|
|
LWLockRelease(TwoPhaseStateLock);
|
|
|
|
for (i = 0; i < nxids; i++)
|
|
{
|
|
TransactionId xid = xids[i];
|
|
int fd;
|
|
|
|
TwoPhaseFilePath(path, xid);
|
|
|
|
fd = BasicOpenFile(path, O_RDWR | PG_BINARY, 0);
|
|
if (fd < 0)
|
|
{
|
|
if (errno == ENOENT)
|
|
{
|
|
/* OK if gxact is no longer valid */
|
|
if (!TransactionIdIsPrepared(xid))
|
|
continue;
|
|
/* Restore errno in case it was changed */
|
|
errno = ENOENT;
|
|
}
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not open two-phase state file \"%s\": %m",
|
|
path)));
|
|
}
|
|
|
|
if (pg_fsync(fd) != 0)
|
|
{
|
|
close(fd);
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not fsync two-phase state file \"%s\": %m",
|
|
path)));
|
|
}
|
|
|
|
if (close(fd) != 0)
|
|
ereport(ERROR,
|
|
(errcode_for_file_access(),
|
|
errmsg("could not close two-phase state file \"%s\": %m",
|
|
path)));
|
|
}
|
|
|
|
pfree(xids);
|
|
|
|
TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
|
|
}
|
|
|
|
/*
|
|
* PrescanPreparedTransactions
|
|
*
|
|
* Scan the pg_twophase directory and determine the range of valid XIDs
|
|
* present. This is run during database startup, after we have completed
|
|
* reading WAL. ShmemVariableCache->nextXid has been set to one more than
|
|
* the highest XID for which evidence exists in WAL.
|
|
*
|
|
* We throw away any prepared xacts with main XID beyond nextXid --- if any
|
|
* are present, it suggests that the DBA has done a PITR recovery to an
|
|
* earlier point in time without cleaning out pg_twophase. We dare not
|
|
* try to recover such prepared xacts since they likely depend on database
|
|
* state that doesn't exist now.
|
|
*
|
|
* However, we will advance nextXid beyond any subxact XIDs belonging to
|
|
* valid prepared xacts. We need to do this since subxact commit doesn't
|
|
* write a WAL entry, and so there might be no evidence in WAL of those
|
|
* subxact XIDs.
|
|
*
|
|
* Our other responsibility is to determine and return the oldest valid XID
|
|
* among the prepared xacts (if none, return ShmemVariableCache->nextXid).
|
|
* This is needed to synchronize pg_subtrans startup properly.
|
|
*
|
|
* If xids_p and nxids_p are not NULL, pointer to a palloc'd array of all
|
|
* top-level xids is stored in *xids_p. The number of entries in the array
|
|
* is returned in *nxids_p.
|
|
*/
|
|
TransactionId
|
|
PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
|
|
{
|
|
TransactionId origNextXid = ShmemVariableCache->nextXid;
|
|
TransactionId result = origNextXid;
|
|
DIR *cldir;
|
|
struct dirent *clde;
|
|
TransactionId *xids = NULL;
|
|
int nxids = 0;
|
|
int allocsize = 0;
|
|
|
|
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;
|
|
TwoPhaseFileHeader *hdr;
|
|
TransactionId *subxids;
|
|
int i;
|
|
|
|
xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
|
|
|
|
/* Reject XID if too new */
|
|
if (TransactionIdFollowsOrEquals(xid, origNextXid))
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing future two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
/*
|
|
* Note: we can't check if already processed because clog
|
|
* subsystem isn't up yet.
|
|
*/
|
|
|
|
/* Read and validate file */
|
|
buf = ReadTwoPhaseFile(xid, true);
|
|
if (buf == NULL)
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing corrupt two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
/* Deconstruct header */
|
|
hdr = (TwoPhaseFileHeader *) buf;
|
|
if (!TransactionIdEquals(hdr->xid, xid))
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing corrupt two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
pfree(buf);
|
|
continue;
|
|
}
|
|
|
|
/*
|
|
* OK, we think this file is valid. Incorporate xid into the
|
|
* running-minimum result.
|
|
*/
|
|
if (TransactionIdPrecedes(xid, result))
|
|
result = xid;
|
|
|
|
/*
|
|
* Examine subtransaction XIDs ... they should all follow main
|
|
* XID, and they may force us to advance 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.
|
|
*/
|
|
subxids = (TransactionId *)
|
|
(buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
|
|
for (i = 0; i < hdr->nsubxacts; i++)
|
|
{
|
|
TransactionId subxid = subxids[i];
|
|
|
|
Assert(TransactionIdFollows(subxid, xid));
|
|
if (TransactionIdFollowsOrEquals(subxid,
|
|
ShmemVariableCache->nextXid))
|
|
{
|
|
LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
|
|
ShmemVariableCache->nextXid = subxid;
|
|
TransactionIdAdvance(ShmemVariableCache->nextXid);
|
|
LWLockRelease(XidGenLock);
|
|
}
|
|
}
|
|
|
|
|
|
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);
|
|
}
|
|
}
|
|
FreeDir(cldir);
|
|
|
|
if (xids_p)
|
|
{
|
|
*xids_p = xids;
|
|
*nxids_p = nxids;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/*
|
|
* StandbyRecoverPreparedTransactions
|
|
*
|
|
* Scan the pg_twophase directory and setup all the required information to
|
|
* allow standby queries to treat prepared transactions as still active.
|
|
* This is never called at the end of recovery - we use
|
|
* RecoverPreparedTransactions() at that point.
|
|
*
|
|
* Currently we simply call SubTransSetParent() for any subxids of prepared
|
|
* transactions. If overwriteOK is true, it's OK if some XIDs have already
|
|
* been marked in pg_subtrans.
|
|
*/
|
|
void
|
|
StandbyRecoverPreparedTransactions(bool overwriteOK)
|
|
{
|
|
DIR *cldir;
|
|
struct dirent *clde;
|
|
|
|
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;
|
|
TwoPhaseFileHeader *hdr;
|
|
TransactionId *subxids;
|
|
int i;
|
|
|
|
xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
|
|
|
|
/* Already processed? */
|
|
if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing stale two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
/* Read and validate file */
|
|
buf = ReadTwoPhaseFile(xid, true);
|
|
if (buf == NULL)
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing corrupt two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
/* Deconstruct header */
|
|
hdr = (TwoPhaseFileHeader *) buf;
|
|
if (!TransactionIdEquals(hdr->xid, xid))
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing corrupt two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
pfree(buf);
|
|
continue;
|
|
}
|
|
|
|
/*
|
|
* Examine subtransaction XIDs ... they should all follow main
|
|
* XID.
|
|
*/
|
|
subxids = (TransactionId *)
|
|
(buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
|
|
for (i = 0; i < hdr->nsubxacts; i++)
|
|
{
|
|
TransactionId subxid = subxids[i];
|
|
|
|
Assert(TransactionIdFollows(subxid, xid));
|
|
SubTransSetParent(xid, subxid, overwriteOK);
|
|
}
|
|
}
|
|
}
|
|
FreeDir(cldir);
|
|
}
|
|
|
|
/*
|
|
* RecoverPreparedTransactions
|
|
*
|
|
* Scan the pg_twophase directory and reload shared-memory state for each
|
|
* prepared transaction (reacquire locks, etc). This is run during database
|
|
* startup.
|
|
*/
|
|
void
|
|
RecoverPreparedTransactions(void)
|
|
{
|
|
char dir[MAXPGPATH];
|
|
DIR *cldir;
|
|
struct dirent *clde;
|
|
bool overwriteOK = false;
|
|
|
|
snprintf(dir, MAXPGPATH, "%s", TWOPHASE_DIR);
|
|
|
|
cldir = AllocateDir(dir);
|
|
while ((clde = ReadDir(cldir, dir)) != NULL)
|
|
{
|
|
if (strlen(clde->d_name) == 8 &&
|
|
strspn(clde->d_name, "0123456789ABCDEF") == 8)
|
|
{
|
|
TransactionId xid;
|
|
char *buf;
|
|
char *bufptr;
|
|
TwoPhaseFileHeader *hdr;
|
|
TransactionId *subxids;
|
|
GlobalTransaction gxact;
|
|
int i;
|
|
|
|
xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
|
|
|
|
/* Already processed? */
|
|
if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing stale two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
/* Read and validate file */
|
|
buf = ReadTwoPhaseFile(xid, true);
|
|
if (buf == NULL)
|
|
{
|
|
ereport(WARNING,
|
|
(errmsg("removing corrupt two-phase state file \"%s\"",
|
|
clde->d_name)));
|
|
RemoveTwoPhaseFile(xid, true);
|
|
continue;
|
|
}
|
|
|
|
ereport(LOG,
|
|
(errmsg("recovering prepared transaction %u", xid)));
|
|
|
|
/* Deconstruct header */
|
|
hdr = (TwoPhaseFileHeader *) buf;
|
|
Assert(TransactionIdEquals(hdr->xid, xid));
|
|
bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
|
|
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));
|
|
|
|
/*
|
|
* It's possible that SubTransSetParent has been set before, if
|
|
* the prepared transaction generated xid assignment records. Test
|
|
* here must match one used in AssignTransactionId().
|
|
*/
|
|
if (InHotStandby && hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS)
|
|
overwriteOK = true;
|
|
|
|
/*
|
|
* 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.
|
|
*/
|
|
for (i = 0; i < hdr->nsubxacts; i++)
|
|
SubTransSetParent(subxids[i], xid, overwriteOK);
|
|
|
|
/*
|
|
* Recreate its GXACT and dummy PGPROC
|
|
*
|
|
* Note: since we don't have the PREPARE record's WAL location at
|
|
* hand, we leave prepare_lsn zeroes. This means the GXACT will
|
|
* be fsync'd on every future checkpoint. We assume this
|
|
* situation is infrequent enough that the performance cost is
|
|
* negligible (especially since we know the state file has already
|
|
* been fsynced).
|
|
*/
|
|
gxact = MarkAsPreparing(xid, hdr->gid,
|
|
hdr->prepared_at,
|
|
hdr->owner, hdr->database);
|
|
GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
|
|
MarkAsPrepared(gxact);
|
|
|
|
/*
|
|
* 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);
|
|
|
|
pfree(buf);
|
|
}
|
|
}
|
|
FreeDir(cldir);
|
|
}
|
|
|
|
/*
|
|
* RecordTransactionCommitPrepared
|
|
*
|
|
* This is basically the same as RecordTransactionCommit: in particular,
|
|
* we must set the inCommit flag to avoid a race condition.
|
|
*
|
|
* We know the transaction made at least one XLOG entry (its PREPARE),
|
|
* so it is never possible to optimize out the commit record.
|
|
*/
|
|
static void
|
|
RecordTransactionCommitPrepared(TransactionId xid,
|
|
int nchildren,
|
|
TransactionId *children,
|
|
int nrels,
|
|
RelFileNode *rels,
|
|
int ninvalmsgs,
|
|
SharedInvalidationMessage *invalmsgs,
|
|
bool initfileinval)
|
|
{
|
|
XLogRecData rdata[4];
|
|
int lastrdata = 0;
|
|
xl_xact_commit_prepared xlrec;
|
|
XLogRecPtr recptr;
|
|
|
|
START_CRIT_SECTION();
|
|
|
|
/* See notes in RecordTransactionCommit */
|
|
MyPgXact->inCommit = true;
|
|
|
|
/* Emit the XLOG commit record */
|
|
xlrec.xid = xid;
|
|
xlrec.crec.xact_time = GetCurrentTimestamp();
|
|
xlrec.crec.xinfo = initfileinval ? XACT_COMPLETION_UPDATE_RELCACHE_FILE : 0;
|
|
xlrec.crec.nmsgs = 0;
|
|
xlrec.crec.nrels = nrels;
|
|
xlrec.crec.nsubxacts = nchildren;
|
|
xlrec.crec.nmsgs = ninvalmsgs;
|
|
|
|
rdata[0].data = (char *) (&xlrec);
|
|
rdata[0].len = MinSizeOfXactCommitPrepared;
|
|
rdata[0].buffer = InvalidBuffer;
|
|
/* dump rels to delete */
|
|
if (nrels > 0)
|
|
{
|
|
rdata[0].next = &(rdata[1]);
|
|
rdata[1].data = (char *) rels;
|
|
rdata[1].len = nrels * sizeof(RelFileNode);
|
|
rdata[1].buffer = InvalidBuffer;
|
|
lastrdata = 1;
|
|
}
|
|
/* dump committed child Xids */
|
|
if (nchildren > 0)
|
|
{
|
|
rdata[lastrdata].next = &(rdata[2]);
|
|
rdata[2].data = (char *) children;
|
|
rdata[2].len = nchildren * sizeof(TransactionId);
|
|
rdata[2].buffer = InvalidBuffer;
|
|
lastrdata = 2;
|
|
}
|
|
/* dump cache invalidation messages */
|
|
if (ninvalmsgs > 0)
|
|
{
|
|
rdata[lastrdata].next = &(rdata[3]);
|
|
rdata[3].data = (char *) invalmsgs;
|
|
rdata[3].len = ninvalmsgs * sizeof(SharedInvalidationMessage);
|
|
rdata[3].buffer = InvalidBuffer;
|
|
lastrdata = 3;
|
|
}
|
|
rdata[lastrdata].next = NULL;
|
|
|
|
recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT_PREPARED, rdata);
|
|
|
|
/*
|
|
* 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);
|
|
|
|
/*
|
|
* Wake up all walsenders to send WAL up to the COMMIT PREPARED record
|
|
* immediately if replication is enabled
|
|
*/
|
|
if (max_wal_senders > 0)
|
|
WalSndWakeup();
|
|
|
|
/* Mark the transaction committed in pg_clog */
|
|
TransactionIdCommitTree(xid, nchildren, children);
|
|
|
|
/* Checkpoint can proceed now */
|
|
MyPgXact->inCommit = 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);
|
|
}
|
|
|
|
/*
|
|
* RecordTransactionAbortPrepared
|
|
*
|
|
* This is basically the same as RecordTransactionAbort.
|
|
*
|
|
* We know the transaction made at least one XLOG entry (its PREPARE),
|
|
* so it is never possible to optimize out the abort record.
|
|
*/
|
|
static void
|
|
RecordTransactionAbortPrepared(TransactionId xid,
|
|
int nchildren,
|
|
TransactionId *children,
|
|
int nrels,
|
|
RelFileNode *rels)
|
|
{
|
|
XLogRecData rdata[3];
|
|
int lastrdata = 0;
|
|
xl_xact_abort_prepared xlrec;
|
|
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 abort record */
|
|
xlrec.xid = xid;
|
|
xlrec.arec.xact_time = GetCurrentTimestamp();
|
|
xlrec.arec.nrels = nrels;
|
|
xlrec.arec.nsubxacts = nchildren;
|
|
rdata[0].data = (char *) (&xlrec);
|
|
rdata[0].len = MinSizeOfXactAbortPrepared;
|
|
rdata[0].buffer = InvalidBuffer;
|
|
/* dump rels to delete */
|
|
if (nrels > 0)
|
|
{
|
|
rdata[0].next = &(rdata[1]);
|
|
rdata[1].data = (char *) rels;
|
|
rdata[1].len = nrels * sizeof(RelFileNode);
|
|
rdata[1].buffer = InvalidBuffer;
|
|
lastrdata = 1;
|
|
}
|
|
/* dump committed child Xids */
|
|
if (nchildren > 0)
|
|
{
|
|
rdata[lastrdata].next = &(rdata[2]);
|
|
rdata[2].data = (char *) children;
|
|
rdata[2].len = nchildren * sizeof(TransactionId);
|
|
rdata[2].buffer = InvalidBuffer;
|
|
lastrdata = 2;
|
|
}
|
|
rdata[lastrdata].next = NULL;
|
|
|
|
recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ABORT_PREPARED, rdata);
|
|
|
|
/* Always flush, since we're about to remove the 2PC state file */
|
|
XLogFlush(recptr);
|
|
|
|
/*
|
|
* Wake up all walsenders to send WAL up to the ABORT PREPARED record
|
|
* immediately if replication is enabled
|
|
*/
|
|
if (max_wal_senders > 0)
|
|
WalSndWakeup();
|
|
|
|
/*
|
|
* 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);
|
|
}
|