database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-08 11:42:09 +03:00
Code Activity

Integrate FullTransactionIds deeper into two-phase code

Browse Source 
This refactoring is a follow-up of the work done in 5a1dfde833, that
has switched 2PC file names to use FullTransactionIds when written on
disk.  This will help with the integration of a follow-up solution
related to the handling of two-phase files during recovery, to address
older defects while reading these from disk after a crash.

This change is useful in itself as it reduces the need to build the
file names from epoch numbers and TransactionIds, because we can use
directly FullTransactionIds from which the 2PC file names are guessed.
So this avoids a lot of back-and-forth between the FullTransactionIds
retrieved from the file names and how these are passed around in the
internal 2PC logic.

Note that the core of the change is the use of a FullTransactionId
instead of a TransactionId in GlobalTransactionData, that tracks 2PC
file information in shared memory.  The change in TwoPhaseCallback makes
this commit unfit for stable branches.

Noah has contributed a good chunk of this patch.  I have spent some time
on it as well while working on the issues with two-phase state files and
recovery.

Author: Noah Misch <noah@leadboat.com>
Co-Authored-by: Michael Paquier <michael@paquier.xyz>
Discussion: https://postgr.es/m/Z5sd5O9JO7NYNK-C@paquier.xyz
Discussion: https://postgr.es/m/20250116205254.65.nmisch@google.com
This commit is contained in:
Michael Paquier
2025-07-07 12:50:40 +09:00
parent 8aa54aa7ee
commit 62a17a9283
12 changed files with 200 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/backend/access/transam/multixact.c
View File

@ -1847,7 +1847,7 @@ AtPrepare_MultiXact(void)
* Clean up after successful PREPARE TRANSACTION * Clean up after successful PREPARE TRANSACTION
*/ */
void void
PostPrepare_MultiXact(TransactionId xid) PostPrepare_MultiXact(FullTransactionId fxid)
{ {
MultiXactId myOldestMember; MultiXactId myOldestMember;
@ -1858,7 +1858,7 @@ PostPrepare_MultiXact(TransactionId xid)
myOldestMember = OldestMemberMXactId[MyProcNumber]; myOldestMember = OldestMemberMXactId[MyProcNumber];
if (MultiXactIdIsValid(myOldestMember)) if (MultiXactIdIsValid(myOldestMember))
{ {
ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(xid, false); ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(fxid, false);
/* /*
* Even though storing MultiXactId is atomic, acquire lock to make * Even though storing MultiXactId is atomic, acquire lock to make
@ -1896,10 +1896,10 @@ PostPrepare_MultiXact(TransactionId xid)
* Recover the state of a prepared transaction at startup * Recover the state of a prepared transaction at startup
*/ */
void void
multixact_twophase_recover(TransactionId xid, uint16 info, multixact_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(xid, false); ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(fxid, false);
MultiXactId oldestMember; MultiXactId oldestMember;
/* /*
@ -1917,10 +1917,10 @@ multixact_twophase_recover(TransactionId xid, uint16 info,
* Similar to AtEOXact_MultiXact but for COMMIT PREPARED * Similar to AtEOXact_MultiXact but for COMMIT PREPARED
*/ */
void void
multixact_twophase_postcommit(TransactionId xid, uint16 info, multixact_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(xid, true); ProcNumber dummyProcNumber = TwoPhaseGetDummyProcNumber(fxid, true);
Assert(len == sizeof(MultiXactId)); Assert(len == sizeof(MultiXactId));
@ -1932,10 +1932,10 @@ multixact_twophase_postcommit(TransactionId xid, uint16 info,
* This is actually just the same as the COMMIT case. * This is actually just the same as the COMMIT case.
*/ */
void void
multixact_twophase_postabort(TransactionId xid, uint16 info, multixact_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
multixact_twophase_postcommit(xid, info, recdata, len); multixact_twophase_postcommit(fxid, info, recdata, len);
} }
/* /*

245
src/backend/access/transam/twophase.c
View File

@ -159,7 +159,7 @@ typedef struct GlobalTransactionData
*/ */
XLogRecPtr prepare_start_lsn; /* XLOG offset of prepare record start */ XLogRecPtr prepare_start_lsn; /* XLOG offset of prepare record start */
XLogRecPtr prepare_end_lsn; /* XLOG offset of prepare record end */ XLogRecPtr prepare_end_lsn; /* XLOG offset of prepare record end */
TransactionId xid; /* The GXACT id */ FullTransactionId fxid; /* The GXACT full xid */
Oid owner; /* ID of user that executed the xact */ Oid owner; /* ID of user that executed the xact */
ProcNumber locking_backend; /* backend currently working on the xact */ ProcNumber locking_backend; /* backend currently working on the xact */
@ -197,6 +197,7 @@ static GlobalTransaction MyLockedGxact = NULL;
static bool twophaseExitRegistered = false; static bool twophaseExitRegistered = false;
static void PrepareRedoRemoveFull(FullTransactionId fxid, bool giveWarning);
static void RecordTransactionCommitPrepared(TransactionId xid, static void RecordTransactionCommitPrepared(TransactionId xid,
int nchildren, int nchildren,
TransactionId *children, TransactionId *children,
@ -216,19 +217,19 @@ static void RecordTransactionAbortPrepared(TransactionId xid,
int nstats, int nstats,
xl_xact_stats_item *stats, xl_xact_stats_item *stats,
const char *gid); const char *gid);
static void ProcessRecords(char *bufptr, TransactionId xid, static void ProcessRecords(char *bufptr, FullTransactionId fxid,
const TwoPhaseCallback callbacks[]); const TwoPhaseCallback callbacks[]);
static void RemoveGXact(GlobalTransaction gxact); static void RemoveGXact(GlobalTransaction gxact);
static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len); static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len);
static char *ProcessTwoPhaseBuffer(TransactionId xid, static char *ProcessTwoPhaseBuffer(FullTransactionId fxid,
XLogRecPtr prepare_start_lsn, XLogRecPtr prepare_start_lsn,
bool fromdisk, bool setParent, bool setNextXid); bool fromdisk, bool setParent, bool setNextXid);
static void MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, static void MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
const char *gid, TimestampTz prepared_at, Oid owner, const char *gid, TimestampTz prepared_at, Oid owner,
Oid databaseid); Oid databaseid);
static void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning); static void RemoveTwoPhaseFile(FullTransactionId fxid, bool giveWarning);
static void RecreateTwoPhaseFile(TransactionId xid, void *content, int len); static void RecreateTwoPhaseFile(FullTransactionId fxid, void *content, int len);
/* /*
* Initialization of shared memory * Initialization of shared memory
@ -356,7 +357,7 @@ PostPrepare_Twophase(void)
* Reserve the GID for the given transaction. * Reserve the GID for the given transaction.
*/ */
GlobalTransaction GlobalTransaction
MarkAsPreparing(TransactionId xid, const char *gid, MarkAsPreparing(FullTransactionId fxid, const char *gid,
TimestampTz prepared_at, Oid owner, Oid databaseid) TimestampTz prepared_at, Oid owner, Oid databaseid)
{ {
GlobalTransaction gxact; GlobalTransaction gxact;
@ -407,7 +408,7 @@ MarkAsPreparing(TransactionId xid, const char *gid,
gxact = TwoPhaseState->freeGXacts; gxact = TwoPhaseState->freeGXacts;
TwoPhaseState->freeGXacts = gxact->next; TwoPhaseState->freeGXacts = gxact->next;
MarkAsPreparingGuts(gxact, xid, gid, prepared_at, owner, databaseid); MarkAsPreparingGuts(gxact, fxid, gid, prepared_at, owner, databaseid);
gxact->ondisk = false; gxact->ondisk = false;
@ -430,11 +431,13 @@ MarkAsPreparing(TransactionId xid, const char *gid,
* Note: This function should be called with appropriate locks held. * Note: This function should be called with appropriate locks held.
*/ */
static void static void
MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid, MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
TimestampTz prepared_at, Oid owner, Oid databaseid) const char *gid, TimestampTz prepared_at, Oid owner,
Oid databaseid)
{ {
PGPROC *proc; PGPROC *proc;
int i; int i;
TransactionId xid = XidFromFullTransactionId(fxid);
Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE)); Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
@ -479,7 +482,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
proc->subxidStatus.count = 0; proc->subxidStatus.count = 0;
gxact->prepared_at = prepared_at; gxact->prepared_at = prepared_at;
gxact->xid = xid; gxact->fxid = fxid;
gxact->owner = owner; gxact->owner = owner;
gxact->locking_backend = MyProcNumber; gxact->locking_backend = MyProcNumber;
gxact->valid = false; gxact->valid = false;
@ -797,12 +800,12 @@ pg_prepared_xact(PG_FUNCTION_ARGS)
* caller had better hold it. * caller had better hold it.
*/ */
static GlobalTransaction static GlobalTransaction
TwoPhaseGetGXact(TransactionId xid, bool lock_held) TwoPhaseGetGXact(FullTransactionId fxid, bool lock_held)
{ {
GlobalTransaction result = NULL; GlobalTransaction result = NULL;
int i; int i;
static TransactionId cached_xid = InvalidTransactionId; static FullTransactionId cached_fxid = {InvalidTransactionId};
static GlobalTransaction cached_gxact = NULL; static GlobalTransaction cached_gxact = NULL;
Assert(!lock_held || LWLockHeldByMe(TwoPhaseStateLock)); Assert(!lock_held || LWLockHeldByMe(TwoPhaseStateLock));
@ -811,7 +814,7 @@ TwoPhaseGetGXact(TransactionId xid, bool lock_held)
* During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called * 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. * repeatedly for the same XID. We can save work with a simple cache.
*/ */
if (xid == cached_xid) if (FullTransactionIdEquals(fxid, cached_fxid))
return cached_gxact; return cached_gxact;
if (!lock_held) if (!lock_held)
@ -821,7 +824,7 @@ TwoPhaseGetGXact(TransactionId xid, bool lock_held)
{ {
GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
if (gxact->xid == xid) if (FullTransactionIdEquals(gxact->fxid, fxid))
{ {
result = gxact; result = gxact;
break; break;
@ -832,9 +835,10 @@ TwoPhaseGetGXact(TransactionId xid, bool lock_held)
LWLockRelease(TwoPhaseStateLock); LWLockRelease(TwoPhaseStateLock);
if (result == NULL) /* should not happen */ if (result == NULL) /* should not happen */
elog(ERROR, "failed to find GlobalTransaction for xid %u", xid); elog(ERROR, "failed to find GlobalTransaction for xid %u",
XidFromFullTransactionId(fxid));
cached_xid = xid; cached_fxid = fxid;
cached_gxact = result; cached_gxact = result;
return result; return result;
@ -881,7 +885,7 @@ TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid,
*have_more = true; *have_more = true;
break; break;
} }
result = gxact->xid; result = XidFromFullTransactionId(gxact->fxid);
} }
} }
@ -892,7 +896,7 @@ TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid,
/* /*
* TwoPhaseGetDummyProcNumber * TwoPhaseGetDummyProcNumber
* Get the dummy proc number for prepared transaction specified by XID * Get the dummy proc number for prepared transaction
* *
* Dummy proc numbers are similar to proc numbers of real backends. They * Dummy proc numbers are similar to proc numbers of real backends. They
* start at MaxBackends, and are unique across all currently active real * start at MaxBackends, and are unique across all currently active real
@ -900,24 +904,24 @@ TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid,
* TwoPhaseStateLock will not be taken, so the caller had better hold it. * TwoPhaseStateLock will not be taken, so the caller had better hold it.
*/ */
ProcNumber ProcNumber
TwoPhaseGetDummyProcNumber(TransactionId xid, bool lock_held) TwoPhaseGetDummyProcNumber(FullTransactionId fxid, bool lock_held)
{ {
GlobalTransaction gxact = TwoPhaseGetGXact(xid, lock_held); GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
return gxact->pgprocno; return gxact->pgprocno;
} }
/* /*
* TwoPhaseGetDummyProc * TwoPhaseGetDummyProc
* Get the PGPROC that represents a prepared transaction specified by XID * Get the PGPROC that represents a prepared transaction
* *
* If lock_held is set to true, TwoPhaseStateLock will not be taken, so the * If lock_held is set to true, TwoPhaseStateLock will not be taken, so the
* caller had better hold it. * caller had better hold it.
*/ */
PGPROC * PGPROC *
TwoPhaseGetDummyProc(TransactionId xid, bool lock_held) TwoPhaseGetDummyProc(FullTransactionId fxid, bool lock_held)
{ {
GlobalTransaction gxact = TwoPhaseGetGXact(xid, lock_held); GlobalTransaction gxact = TwoPhaseGetGXact(fxid, lock_held);
return GetPGProcByNumber(gxact->pgprocno); return GetPGProcByNumber(gxact->pgprocno);
} }
@ -942,10 +946,8 @@ AdjustToFullTransactionId(TransactionId xid)
} }
static inline int static inline int
TwoPhaseFilePath(char *path, TransactionId xid) TwoPhaseFilePath(char *path, FullTransactionId fxid)
{ {
FullTransactionId fxid = AdjustToFullTransactionId(xid);
return snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X%08X", return snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X%08X",
EpochFromFullTransactionId(fxid), EpochFromFullTransactionId(fxid),
XidFromFullTransactionId(fxid)); XidFromFullTransactionId(fxid));
@ -1049,7 +1051,7 @@ void
StartPrepare(GlobalTransaction gxact) StartPrepare(GlobalTransaction gxact)
{ {
PGPROC *proc = GetPGProcByNumber(gxact->pgprocno); PGPROC *proc = GetPGProcByNumber(gxact->pgprocno);
TransactionId xid = gxact->xid; TransactionId xid = XidFromFullTransactionId(gxact->fxid);
TwoPhaseFileHeader hdr; TwoPhaseFileHeader hdr;
TransactionId *children; TransactionId *children;
RelFileLocator *commitrels; RelFileLocator *commitrels;
@ -1281,10 +1283,11 @@ RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
* If it looks OK (has a valid magic number and CRC), return the palloc'd * If it looks OK (has a valid magic number and CRC), return the palloc'd
* contents of the file, issuing an error when finding corrupted data. If * contents of the file, issuing an error when finding corrupted data. If
* missing_ok is true, which indicates that missing files can be safely * missing_ok is true, which indicates that missing files can be safely
* ignored, then return NULL. This state can be reached when doing recovery. * ignored, then return NULL. This state can be reached when doing recovery
* after discarding two-phase files from frozen epochs.
*/ */
static char * static char *
ReadTwoPhaseFile(TransactionId xid, bool missing_ok) ReadTwoPhaseFile(FullTransactionId fxid, bool missing_ok)
{ {
char path[MAXPGPATH]; char path[MAXPGPATH];
char *buf; char *buf;
@ -1296,7 +1299,7 @@ ReadTwoPhaseFile(TransactionId xid, bool missing_ok)
file_crc; file_crc;
int r; int r;
TwoPhaseFilePath(path, xid); TwoPhaseFilePath(path, fxid);
fd = OpenTransientFile(path, O_RDONLY | PG_BINARY); fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
if (fd < 0) if (fd < 0)
@ -1461,6 +1464,7 @@ StandbyTransactionIdIsPrepared(TransactionId xid)
char *buf; char *buf;
TwoPhaseFileHeader *hdr; TwoPhaseFileHeader *hdr;
bool result; bool result;
FullTransactionId fxid;
Assert(TransactionIdIsValid(xid)); Assert(TransactionIdIsValid(xid));
@ -1468,7 +1472,8 @@ StandbyTransactionIdIsPrepared(TransactionId xid)
return false; /* nothing to do */ return false; /* nothing to do */
/* Read and validate file */ /* Read and validate file */
buf = ReadTwoPhaseFile(xid, true); fxid = AdjustToFullTransactionId(xid);
buf = ReadTwoPhaseFile(fxid, true);
if (buf == NULL) if (buf == NULL)
return false; return false;
@ -1488,6 +1493,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
{ {
GlobalTransaction gxact; GlobalTransaction gxact;
PGPROC *proc; PGPROC *proc;
FullTransactionId fxid;
TransactionId xid; TransactionId xid;
bool ondisk; bool ondisk;
char *buf; char *buf;
@ -1509,7 +1515,8 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
*/ */
gxact = LockGXact(gid, GetUserId()); gxact = LockGXact(gid, GetUserId());
proc = GetPGProcByNumber(gxact->pgprocno); proc = GetPGProcByNumber(gxact->pgprocno);
xid = gxact->xid; fxid = gxact->fxid;
xid = XidFromFullTransactionId(fxid);
/* /*
* Read and validate 2PC state data. State data will typically be stored * Read and validate 2PC state data. State data will typically be stored
@ -1517,7 +1524,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
* to disk if for some reason they have lived for a long time. * to disk if for some reason they have lived for a long time.
*/ */
if (gxact->ondisk) if (gxact->ondisk)
buf = ReadTwoPhaseFile(xid, false); buf = ReadTwoPhaseFile(fxid, false);
else else
XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
@ -1636,11 +1643,11 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
/* And now do the callbacks */ /* And now do the callbacks */
if (isCommit) if (isCommit)
ProcessRecords(bufptr, xid, twophase_postcommit_callbacks); ProcessRecords(bufptr, fxid, twophase_postcommit_callbacks);
else else
ProcessRecords(bufptr, xid, twophase_postabort_callbacks); ProcessRecords(bufptr, fxid, twophase_postabort_callbacks);
PredicateLockTwoPhaseFinish(xid, isCommit); PredicateLockTwoPhaseFinish(fxid, isCommit);
/* /*
* Read this value while holding the two-phase lock, as the on-disk 2PC * Read this value while holding the two-phase lock, as the on-disk 2PC
@ -1664,7 +1671,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
* And now we can clean up any files we may have left. * And now we can clean up any files we may have left.
*/ */
if (ondisk) if (ondisk)
RemoveTwoPhaseFile(xid, true); RemoveTwoPhaseFile(fxid, true);
MyLockedGxact = NULL; MyLockedGxact = NULL;
@ -1677,7 +1684,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
* Scan 2PC state data in memory and call the indicated callbacks for each 2PC record. * Scan 2PC state data in memory and call the indicated callbacks for each 2PC record.
*/ */
static void static void
ProcessRecords(char *bufptr, TransactionId xid, ProcessRecords(char *bufptr, FullTransactionId fxid,
const TwoPhaseCallback callbacks[]) const TwoPhaseCallback callbacks[])
{ {
for (;;) for (;;)
@ -1691,24 +1698,28 @@ ProcessRecords(char *bufptr, TransactionId xid,
bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk)); bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));
if (callbacks[record->rmid] != NULL) if (callbacks[record->rmid] != NULL)
callbacks[record->rmid] (xid, record->info, bufptr, record->len); callbacks[record->rmid] (fxid, record->info, bufptr, record->len);
bufptr += MAXALIGN(record->len); bufptr += MAXALIGN(record->len);
} }
} }
/* /*
* Remove the 2PC file for the specified XID. * Remove the 2PC file.
* *
* If giveWarning is false, do not complain about file-not-present; * If giveWarning is false, do not complain about file-not-present;
* this is an expected case during WAL replay. * this is an expected case during WAL replay.
*
* This routine is used at early stages at recovery where future and
* past orphaned files are checked, hence the FullTransactionId to build
* a complete file name fit for the removal.
*/ */
static void static void
RemoveTwoPhaseFile(TransactionId xid, bool giveWarning) RemoveTwoPhaseFile(FullTransactionId fxid, bool giveWarning)
{ {
char path[MAXPGPATH]; char path[MAXPGPATH];
TwoPhaseFilePath(path, xid); TwoPhaseFilePath(path, fxid);
if (unlink(path)) if (unlink(path))
if (errno != ENOENT || giveWarning) if (errno != ENOENT || giveWarning)
ereport(WARNING, ereport(WARNING,
@ -1723,7 +1734,7 @@ RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
* Note: content and len don't include CRC. * Note: content and len don't include CRC.
*/ */
static void static void
RecreateTwoPhaseFile(TransactionId xid, void *content, int len) RecreateTwoPhaseFile(FullTransactionId fxid, void *content, int len)
{ {
char path[MAXPGPATH]; char path[MAXPGPATH];
pg_crc32c statefile_crc; pg_crc32c statefile_crc;
@ -1734,7 +1745,7 @@ RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
COMP_CRC32C(statefile_crc, content, len); COMP_CRC32C(statefile_crc, content, len);
FIN_CRC32C(statefile_crc); FIN_CRC32C(statefile_crc);
TwoPhaseFilePath(path, xid); TwoPhaseFilePath(path, fxid);
fd = OpenTransientFile(path, fd = OpenTransientFile(path,
O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY); O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY);
@ -1846,7 +1857,7 @@ CheckPointTwoPhase(XLogRecPtr redo_horizon)
int len; int len;
XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len); XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len);
RecreateTwoPhaseFile(gxact->xid, buf, len); RecreateTwoPhaseFile(gxact->fxid, buf, len);
gxact->ondisk = true; gxact->ondisk = true;
gxact->prepare_start_lsn = InvalidXLogRecPtr; gxact->prepare_start_lsn = InvalidXLogRecPtr;
gxact->prepare_end_lsn = InvalidXLogRecPtr; gxact->prepare_end_lsn = InvalidXLogRecPtr;
@ -1897,19 +1908,17 @@ restoreTwoPhaseData(void)
if (strlen(clde->d_name) == 16 && if (strlen(clde->d_name) == 16 &&
strspn(clde->d_name, "0123456789ABCDEF") == 16) strspn(clde->d_name, "0123456789ABCDEF") == 16)
{ {
TransactionId xid;
FullTransactionId fxid; FullTransactionId fxid;
char *buf; char *buf;
fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16)); fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16));
xid = XidFromFullTransactionId(fxid);
buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr, buf = ProcessTwoPhaseBuffer(fxid, InvalidXLogRecPtr,
true, false, false); true, false, false);
if (buf == NULL) if (buf == NULL)
continue; continue;
PrepareRedoAdd(buf, InvalidXLogRecPtr, PrepareRedoAdd(fxid, buf, InvalidXLogRecPtr,
InvalidXLogRecPtr, InvalidRepOriginId); InvalidXLogRecPtr, InvalidRepOriginId);
} }
} }
@ -1968,9 +1977,7 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
Assert(gxact->inredo); Assert(gxact->inredo);
xid = gxact->xid; buf = ProcessTwoPhaseBuffer(gxact->fxid,
buf = ProcessTwoPhaseBuffer(xid,
gxact->prepare_start_lsn, gxact->prepare_start_lsn,
gxact->ondisk, false, true); gxact->ondisk, false, true);
@ -1981,6 +1988,7 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
* OK, we think this file is valid. Incorporate xid into the * OK, we think this file is valid. Incorporate xid into the
* running-minimum result. * running-minimum result.
*/ */
xid = XidFromFullTransactionId(gxact->fxid);
if (TransactionIdPrecedes(xid, result)) if (TransactionIdPrecedes(xid, result))
result = xid; result = xid;
@ -2036,15 +2044,12 @@ StandbyRecoverPreparedTransactions(void)
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
for (i = 0; i < TwoPhaseState->numPrepXacts; i++) for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
{ {
TransactionId xid;
char *buf; char *buf;
GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
Assert(gxact->inredo); Assert(gxact->inredo);
xid = gxact->xid; buf = ProcessTwoPhaseBuffer(gxact->fxid,
buf = ProcessTwoPhaseBuffer(xid,
gxact->prepare_start_lsn, gxact->prepare_start_lsn,
gxact->ondisk, true, false); gxact->ondisk, true, false);
if (buf != NULL) if (buf != NULL)
@ -2077,16 +2082,14 @@ RecoverPreparedTransactions(void)
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
for (i = 0; i < TwoPhaseState->numPrepXacts; i++) for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
{ {
TransactionId xid;
char *buf; char *buf;
GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
FullTransactionId fxid = gxact->fxid;
char *bufptr; char *bufptr;
TwoPhaseFileHeader *hdr; TwoPhaseFileHeader *hdr;
TransactionId *subxids; TransactionId *subxids;
const char *gid; const char *gid;
xid = gxact->xid;
/* /*
* Reconstruct subtrans state for the transaction --- needed because * Reconstruct subtrans state for the transaction --- needed because
* pg_subtrans is not preserved over a restart. Note that we are * pg_subtrans is not preserved over a restart. Note that we are
@ -2096,17 +2099,20 @@ RecoverPreparedTransactions(void)
* SubTransSetParent has been set before, if the prepared transaction * SubTransSetParent has been set before, if the prepared transaction
* generated xid assignment records. * generated xid assignment records.
*/ */
buf = ProcessTwoPhaseBuffer(xid, buf = ProcessTwoPhaseBuffer(gxact->fxid,
gxact->prepare_start_lsn, gxact->prepare_start_lsn,
gxact->ondisk, true, false); gxact->ondisk, true, false);
if (buf == NULL) if (buf == NULL)
continue; continue;
ereport(LOG, ereport(LOG,
(errmsg("recovering prepared transaction %u from shared memory", xid))); (errmsg("recovering prepared transaction %u of epoch %u from shared memory",
XidFromFullTransactionId(gxact->fxid),
EpochFromFullTransactionId(gxact->fxid))));
hdr = (TwoPhaseFileHeader *) buf; hdr = (TwoPhaseFileHeader *) buf;
Assert(TransactionIdEquals(hdr->xid, xid)); Assert(TransactionIdEquals(hdr->xid,
XidFromFullTransactionId(gxact->fxid)));
bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader)); bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
gid = (const char *) bufptr; gid = (const char *) bufptr;
bufptr += MAXALIGN(hdr->gidlen); bufptr += MAXALIGN(hdr->gidlen);
@ -2122,7 +2128,7 @@ RecoverPreparedTransactions(void)
* Recreate its GXACT and dummy PGPROC. But, check whether it was * Recreate its GXACT and dummy PGPROC. But, check whether it was
* added in redo and already has a shmem entry for it. * added in redo and already has a shmem entry for it.
*/ */
MarkAsPreparingGuts(gxact, xid, gid, MarkAsPreparingGuts(gxact, gxact->fxid, gid,
hdr->prepared_at, hdr->prepared_at,
hdr->owner, hdr->database); hdr->owner, hdr->database);
@ -2137,7 +2143,7 @@ RecoverPreparedTransactions(void)
/* /*
* Recover other state (notably locks) using resource managers. * Recover other state (notably locks) using resource managers.
*/ */
ProcessRecords(bufptr, xid, twophase_recover_callbacks); ProcessRecords(bufptr, fxid, twophase_recover_callbacks);
/* /*
* Release locks held by the standby process after we process each * Release locks held by the standby process after we process each
@ -2145,7 +2151,7 @@ RecoverPreparedTransactions(void)
* additional locks at any one time. * additional locks at any one time.
*/ */
if (InHotStandby) if (InHotStandby)
StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids); StandbyReleaseLockTree(hdr->xid, hdr->nsubxacts, subxids);
/* /*
* We're done with recovering this transaction. Clear MyLockedGxact, * We're done with recovering this transaction. Clear MyLockedGxact,
@ -2164,7 +2170,7 @@ RecoverPreparedTransactions(void)
/* /*
* ProcessTwoPhaseBuffer * ProcessTwoPhaseBuffer
* *
* Given a transaction id, read it either from disk or read it directly * Given a FullTransactionId, read it either from disk or read it directly
* via shmem xlog record pointer using the provided "prepare_start_lsn". * via shmem xlog record pointer using the provided "prepare_start_lsn".
* *
* If setParent is true, set up subtransaction parent linkages. * If setParent is true, set up subtransaction parent linkages.
@ -2173,13 +2179,12 @@ RecoverPreparedTransactions(void)
* value scanned. * value scanned.
*/ */
static char * static char *
ProcessTwoPhaseBuffer(TransactionId xid, ProcessTwoPhaseBuffer(FullTransactionId fxid,
XLogRecPtr prepare_start_lsn, XLogRecPtr prepare_start_lsn,
bool fromdisk, bool fromdisk,
bool setParent, bool setNextXid) bool setParent, bool setNextXid)
{ {
FullTransactionId nextXid = TransamVariables->nextXid; FullTransactionId nextXid = TransamVariables->nextXid;
TransactionId origNextXid = XidFromFullTransactionId(nextXid);
TransactionId *subxids; TransactionId *subxids;
char *buf; char *buf;
TwoPhaseFileHeader *hdr; TwoPhaseFileHeader *hdr;
@ -2191,41 +2196,46 @@ ProcessTwoPhaseBuffer(TransactionId xid,
Assert(prepare_start_lsn != InvalidXLogRecPtr); Assert(prepare_start_lsn != InvalidXLogRecPtr);
/* Already processed? */ /* Already processed? */
if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid)) if (TransactionIdDidCommit(XidFromFullTransactionId(fxid)) ||
TransactionIdDidAbort(XidFromFullTransactionId(fxid)))
{ {
if (fromdisk) if (fromdisk)
{ {
ereport(WARNING, ereport(WARNING,
(errmsg("removing stale two-phase state file for transaction %u", (errmsg("removing stale two-phase state file for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
RemoveTwoPhaseFile(xid, true); EpochFromFullTransactionId(fxid))));
RemoveTwoPhaseFile(fxid, true);
} }
else else
{ {
ereport(WARNING, ereport(WARNING,
(errmsg("removing stale two-phase state from memory for transaction %u", (errmsg("removing stale two-phase state from memory for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
PrepareRedoRemove(xid, true); EpochFromFullTransactionId(fxid))));
PrepareRedoRemoveFull(fxid, true);
} }
return NULL; return NULL;
} }
/* Reject XID if too new */ /* Reject XID if too new */
if (TransactionIdFollowsOrEquals(xid, origNextXid)) if (FullTransactionIdFollowsOrEquals(fxid, nextXid))
{ {
if (fromdisk) if (fromdisk)
{ {
ereport(WARNING, ereport(WARNING,
(errmsg("removing future two-phase state file for transaction %u", (errmsg("removing future two-phase state file for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
RemoveTwoPhaseFile(xid, true); EpochFromFullTransactionId(fxid))));
RemoveTwoPhaseFile(fxid, true);
} }
else else
{ {
ereport(WARNING, ereport(WARNING,
(errmsg("removing future two-phase state from memory for transaction %u", (errmsg("removing future two-phase state from memory for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
PrepareRedoRemove(xid, true); EpochFromFullTransactionId(fxid))));
PrepareRedoRemoveFull(fxid, true);
} }
return NULL; return NULL;
} }
@ -2233,7 +2243,7 @@ ProcessTwoPhaseBuffer(TransactionId xid,
if (fromdisk) if (fromdisk)
{ {
/* Read and validate file */ /* Read and validate file */
buf = ReadTwoPhaseFile(xid, false); buf = ReadTwoPhaseFile(fxid, false);
} }
else else
{ {
@ -2243,18 +2253,20 @@ ProcessTwoPhaseBuffer(TransactionId xid,
/* Deconstruct header */ /* Deconstruct header */
hdr = (TwoPhaseFileHeader *) buf; hdr = (TwoPhaseFileHeader *) buf;
if (!TransactionIdEquals(hdr->xid, xid)) if (!TransactionIdEquals(hdr->xid, XidFromFullTransactionId(fxid)))
{ {
if (fromdisk) if (fromdisk)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_DATA_CORRUPTED), (errcode(ERRCODE_DATA_CORRUPTED),
errmsg("corrupted two-phase state file for transaction %u", errmsg("corrupted two-phase state file for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
EpochFromFullTransactionId(fxid))));
else else
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_DATA_CORRUPTED), (errcode(ERRCODE_DATA_CORRUPTED),
errmsg("corrupted two-phase state in memory for transaction %u", errmsg("corrupted two-phase state in memory for transaction %u of epoch %u",
xid))); XidFromFullTransactionId(fxid),
EpochFromFullTransactionId(fxid))));
} }
/* /*
@ -2268,14 +2280,14 @@ ProcessTwoPhaseBuffer(TransactionId xid,
{ {
TransactionId subxid = subxids[i]; TransactionId subxid = subxids[i];
Assert(TransactionIdFollows(subxid, xid)); Assert(TransactionIdFollows(subxid, XidFromFullTransactionId(fxid)));
/* update nextXid if needed */ /* update nextXid if needed */
if (setNextXid) if (setNextXid)
AdvanceNextFullTransactionIdPastXid(subxid); AdvanceNextFullTransactionIdPastXid(subxid);
if (setParent) if (setParent)
SubTransSetParent(subxid, xid); SubTransSetParent(subxid, XidFromFullTransactionId(fxid));
} }
return buf; return buf;
@ -2466,8 +2478,9 @@ RecordTransactionAbortPrepared(TransactionId xid,
* data, the entry is marked as located on disk. * data, the entry is marked as located on disk.
*/ */
void void
PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, PrepareRedoAdd(FullTransactionId fxid, char *buf,
XLogRecPtr end_lsn, RepOriginId origin_id) XLogRecPtr start_lsn, XLogRecPtr end_lsn,
RepOriginId origin_id)
{ {
TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf; TwoPhaseFileHeader *hdr = (TwoPhaseFileHeader *) buf;
char *bufptr; char *bufptr;
@ -2477,6 +2490,13 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE)); Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
Assert(RecoveryInProgress()); Assert(RecoveryInProgress());
if (!FullTransactionIdIsValid(fxid))
{
Assert(InRecovery);
fxid = FullTransactionIdFromAllowableAt(TransamVariables->nextXid,
hdr->xid);
}
bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader)); bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
gid = (const char *) bufptr; gid = (const char *) bufptr;
@ -2505,7 +2525,8 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
{ {
char path[MAXPGPATH]; char path[MAXPGPATH];
TwoPhaseFilePath(path, hdr->xid); Assert(InRecovery);
TwoPhaseFilePath(path, fxid);
if (access(path, F_OK) == 0) if (access(path, F_OK) == 0)
{ {
@ -2536,7 +2557,7 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
gxact->prepared_at = hdr->prepared_at; gxact->prepared_at = hdr->prepared_at;
gxact->prepare_start_lsn = start_lsn; gxact->prepare_start_lsn = start_lsn;
gxact->prepare_end_lsn = end_lsn; gxact->prepare_end_lsn = end_lsn;
gxact->xid = hdr->xid; gxact->fxid = fxid;
gxact->owner = hdr->owner; gxact->owner = hdr->owner;
gxact->locking_backend = INVALID_PROC_NUMBER; gxact->locking_backend = INVALID_PROC_NUMBER;
gxact->valid = false; gxact->valid = false;
@ -2555,11 +2576,13 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
false /* backward */ , false /* WAL */ ); false /* backward */ , false /* WAL */ );
} }
elog(DEBUG2, "added 2PC data in shared memory for transaction %u", gxact->xid); elog(DEBUG2, "added 2PC data in shared memory for transaction %u of epoch %u",
XidFromFullTransactionId(gxact->fxid),
EpochFromFullTransactionId(gxact->fxid));
} }
/* /*
* PrepareRedoRemove * PrepareRedoRemoveFull
* *
* Remove the corresponding gxact entry from TwoPhaseState. Also remove * Remove the corresponding gxact entry from TwoPhaseState. Also remove
* the 2PC file if a prepared transaction was saved via an earlier checkpoint. * the 2PC file if a prepared transaction was saved via an earlier checkpoint.
@ -2567,8 +2590,8 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
* Caller must hold TwoPhaseStateLock in exclusive mode, because TwoPhaseState * Caller must hold TwoPhaseStateLock in exclusive mode, because TwoPhaseState
* is updated. * is updated.
*/ */
void static void
PrepareRedoRemove(TransactionId xid, bool giveWarning) PrepareRedoRemoveFull(FullTransactionId fxid, bool giveWarning)
{ {
GlobalTransaction gxact = NULL; GlobalTransaction gxact = NULL;
int i; int i;
@ -2581,7 +2604,7 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
{ {
gxact = TwoPhaseState->prepXacts[i]; gxact = TwoPhaseState->prepXacts[i];
if (gxact->xid == xid) if (FullTransactionIdEquals(gxact->fxid, fxid))
{ {
Assert(gxact->inredo); Assert(gxact->inredo);
found = true; found = true;
@ -2598,12 +2621,28 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
/* /*
* And now we can clean up any files we may have left. * And now we can clean up any files we may have left.
*/ */
elog(DEBUG2, "removing 2PC data for transaction %u", xid); elog(DEBUG2, "removing 2PC data for transaction %u of epoch %u ",
XidFromFullTransactionId(fxid),
EpochFromFullTransactionId(fxid));
if (gxact->ondisk) if (gxact->ondisk)
RemoveTwoPhaseFile(xid, giveWarning); RemoveTwoPhaseFile(fxid, giveWarning);
RemoveGXact(gxact); RemoveGXact(gxact);
} }
/*
* Wrapper of PrepareRedoRemoveFull(), for TransactionIds.
*/
void
PrepareRedoRemove(TransactionId xid, bool giveWarning)
{
FullTransactionId fxid =
FullTransactionIdFromAllowableAt(TransamVariables->nextXid, xid);
PrepareRedoRemoveFull(fxid, giveWarning);
}
/* /*
* LookupGXact * LookupGXact
* Check if the prepared transaction with the given GID, lsn and timestamp * Check if the prepared transaction with the given GID, lsn and timestamp
@ -2648,7 +2687,7 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
* between publisher and subscriber. * between publisher and subscriber.
*/ */
if (gxact->ondisk) if (gxact->ondisk)
buf = ReadTwoPhaseFile(gxact->xid, false); buf = ReadTwoPhaseFile(gxact->fxid, false);
else else
{ {
Assert(gxact->prepare_start_lsn); Assert(gxact->prepare_start_lsn);

13
src/backend/access/transam/xact.c
View File

@ -2515,7 +2515,7 @@ static void
PrepareTransaction(void) PrepareTransaction(void)
{ {
TransactionState s = CurrentTransactionState; TransactionState s = CurrentTransactionState;
TransactionId xid = GetCurrentTransactionId(); FullTransactionId fxid = GetCurrentFullTransactionId();
GlobalTransaction gxact; GlobalTransaction gxact;
TimestampTz prepared_at; TimestampTz prepared_at;
@ -2644,7 +2644,7 @@ PrepareTransaction(void)
* Reserve the GID for this transaction. This could fail if the requested * Reserve the GID for this transaction. This could fail if the requested
* GID is invalid or already in use. * GID is invalid or already in use.
*/ */
gxact = MarkAsPreparing(xid, prepareGID, prepared_at, gxact = MarkAsPreparing(fxid, prepareGID, prepared_at,
GetUserId(), MyDatabaseId); GetUserId(), MyDatabaseId);
prepareGID = NULL; prepareGID = NULL;
@ -2694,7 +2694,7 @@ PrepareTransaction(void)
* ProcArrayClearTransaction(). Otherwise, a GetLockConflicts() would * ProcArrayClearTransaction(). Otherwise, a GetLockConflicts() would
* conclude "xact already committed or aborted" for our locks. * conclude "xact already committed or aborted" for our locks.
*/ */
PostPrepare_Locks(xid); PostPrepare_Locks(fxid);
/* /*
* Let others know about no transaction in progress by me. This has to be * Let others know about no transaction in progress by me. This has to be
@ -2738,9 +2738,9 @@ PrepareTransaction(void)
PostPrepare_smgr(); PostPrepare_smgr();
PostPrepare_MultiXact(xid); PostPrepare_MultiXact(fxid);
PostPrepare_PredicateLocks(xid); PostPrepare_PredicateLocks(fxid);
ResourceOwnerRelease(TopTransactionResourceOwner, ResourceOwnerRelease(TopTransactionResourceOwner,
RESOURCE_RELEASE_LOCKS, RESOURCE_RELEASE_LOCKS,
@ -6420,7 +6420,8 @@ xact_redo(XLogReaderState *record)
* gxact entry. * gxact entry.
*/ */
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
PrepareRedoAdd(XLogRecGetData(record), PrepareRedoAdd(InvalidFullTransactionId,
XLogRecGetData(record),
record->ReadRecPtr, record->ReadRecPtr,
record->EndRecPtr, record->EndRecPtr,
XLogRecGetOrigin(record)); XLogRecGetOrigin(record));

20
src/backend/storage/lmgr/lock.c
View File

@ -3539,9 +3539,9 @@ AtPrepare_Locks(void)
* but that probably costs more cycles. * but that probably costs more cycles.
*/ */
void void
PostPrepare_Locks(TransactionId xid) PostPrepare_Locks(FullTransactionId fxid)
{ {
PGPROC *newproc = TwoPhaseGetDummyProc(xid, false); PGPROC *newproc = TwoPhaseGetDummyProc(fxid, false);
HASH_SEQ_STATUS status; HASH_SEQ_STATUS status;
LOCALLOCK *locallock; LOCALLOCK *locallock;
LOCK *lock; LOCK *lock;
@ -4324,11 +4324,11 @@ DumpAllLocks(void)
* and PANIC anyway. * and PANIC anyway.
*/ */
void void
lock_twophase_recover(TransactionId xid, uint16 info, lock_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata; TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata;
PGPROC *proc = TwoPhaseGetDummyProc(xid, false); PGPROC *proc = TwoPhaseGetDummyProc(fxid, false);
LOCKTAG *locktag; LOCKTAG *locktag;
LOCKMODE lockmode; LOCKMODE lockmode;
LOCKMETHODID lockmethodid; LOCKMETHODID lockmethodid;
@ -4505,7 +4505,7 @@ lock_twophase_recover(TransactionId xid, uint16 info,
* starting up into hot standby mode. * starting up into hot standby mode.
*/ */
void void
lock_twophase_standby_recover(TransactionId xid, uint16 info, lock_twophase_standby_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata; TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata;
@ -4524,7 +4524,7 @@ lock_twophase_standby_recover(TransactionId xid, uint16 info,
if (lockmode == AccessExclusiveLock && if (lockmode == AccessExclusiveLock &&
locktag->locktag_type == LOCKTAG_RELATION) locktag->locktag_type == LOCKTAG_RELATION)
{ {
StandbyAcquireAccessExclusiveLock(xid, StandbyAcquireAccessExclusiveLock(XidFromFullTransactionId(fxid),
locktag->locktag_field1 /* dboid */ , locktag->locktag_field1 /* dboid */ ,
locktag->locktag_field2 /* reloid */ ); locktag->locktag_field2 /* reloid */ );
} }
@ -4537,11 +4537,11 @@ lock_twophase_standby_recover(TransactionId xid, uint16 info,
* Find and release the lock indicated by the 2PC record. * Find and release the lock indicated by the 2PC record.
*/ */
void void
lock_twophase_postcommit(TransactionId xid, uint16 info, lock_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata; TwoPhaseLockRecord *rec = (TwoPhaseLockRecord *) recdata;
PGPROC *proc = TwoPhaseGetDummyProc(xid, true); PGPROC *proc = TwoPhaseGetDummyProc(fxid, true);
LOCKTAG *locktag; LOCKTAG *locktag;
LOCKMETHODID lockmethodid; LOCKMETHODID lockmethodid;
LockMethod lockMethodTable; LockMethod lockMethodTable;
@ -4563,10 +4563,10 @@ lock_twophase_postcommit(TransactionId xid, uint16 info,
* This is actually just the same as the COMMIT case. * This is actually just the same as the COMMIT case.
*/ */
void void
lock_twophase_postabort(TransactionId xid, uint16 info, lock_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
lock_twophase_postcommit(xid, info, recdata, len); lock_twophase_postcommit(fxid, info, recdata, len);
} }
/* /*

11
src/backend/storage/lmgr/predicate.c
View File

@ -191,7 +191,7 @@
* AtPrepare_PredicateLocks(void); * AtPrepare_PredicateLocks(void);
* PostPrepare_PredicateLocks(TransactionId xid); * PostPrepare_PredicateLocks(TransactionId xid);
* PredicateLockTwoPhaseFinish(TransactionId xid, bool isCommit); * PredicateLockTwoPhaseFinish(TransactionId xid, bool isCommit);
* predicatelock_twophase_recover(TransactionId xid, uint16 info, * predicatelock_twophase_recover(FullTransactionId fxid, uint16 info,
* void *recdata, uint32 len); * void *recdata, uint32 len);
*/ */
@ -4856,7 +4856,7 @@ AtPrepare_PredicateLocks(void)
* anyway. We only need to clean up our local state. * anyway. We only need to clean up our local state.
*/ */
void void
PostPrepare_PredicateLocks(TransactionId xid) PostPrepare_PredicateLocks(FullTransactionId fxid)
{ {
if (MySerializableXact == InvalidSerializableXact) if (MySerializableXact == InvalidSerializableXact)
return; return;
@ -4879,12 +4879,12 @@ PostPrepare_PredicateLocks(TransactionId xid)
* commits or aborts. * commits or aborts.
*/ */
void void
PredicateLockTwoPhaseFinish(TransactionId xid, bool isCommit) PredicateLockTwoPhaseFinish(FullTransactionId fxid, bool isCommit)
{ {
SERIALIZABLEXID *sxid; SERIALIZABLEXID *sxid;
SERIALIZABLEXIDTAG sxidtag; SERIALIZABLEXIDTAG sxidtag;
sxidtag.xid = xid; sxidtag.xid = XidFromFullTransactionId(fxid);
LWLockAcquire(SerializableXactHashLock, LW_SHARED); LWLockAcquire(SerializableXactHashLock, LW_SHARED);
sxid = (SERIALIZABLEXID *) sxid = (SERIALIZABLEXID *)
@ -4906,10 +4906,11 @@ PredicateLockTwoPhaseFinish(TransactionId xid, bool isCommit)
* Re-acquire a predicate lock belonging to a transaction that was prepared. * Re-acquire a predicate lock belonging to a transaction that was prepared.
*/ */
void void
predicatelock_twophase_recover(TransactionId xid, uint16 info, predicatelock_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhasePredicateRecord *record; TwoPhasePredicateRecord *record;
TransactionId xid = XidFromFullTransactionId(fxid);
Assert(len == sizeof(TwoPhasePredicateRecord)); Assert(len == sizeof(TwoPhasePredicateRecord));

4
src/backend/utils/activity/pgstat_relation.c
View File

@ -744,7 +744,7 @@ PostPrepare_PgStat_Relations(PgStat_SubXactStatus *xact_state)
* Load the saved counts into our local pgstats state. * Load the saved counts into our local pgstats state.
*/ */
void void
pgstat_twophase_postcommit(TransactionId xid, uint16 info, pgstat_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata; TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata;
@ -780,7 +780,7 @@ pgstat_twophase_postcommit(TransactionId xid, uint16 info,
* as aborted. * as aborted.
*/ */
void void
pgstat_twophase_postabort(TransactionId xid, uint16 info, pgstat_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len) void *recdata, uint32 len)
{ {
TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata; TwoPhasePgStatRecord *rec = (TwoPhasePgStatRecord *) recdata;

9
src/include/access/multixact.h
View File

@ -11,6 +11,7 @@
#ifndef MULTIXACT_H #ifndef MULTIXACT_H
#define MULTIXACT_H #define MULTIXACT_H
#include "access/transam.h"
#include "access/xlogreader.h" #include "access/xlogreader.h"
#include "lib/stringinfo.h" #include "lib/stringinfo.h"
#include "storage/sync.h" #include "storage/sync.h"
@ -119,7 +120,7 @@ extern int multixactmemberssyncfiletag(const FileTag *ftag, char *path);
extern void AtEOXact_MultiXact(void); extern void AtEOXact_MultiXact(void);
extern void AtPrepare_MultiXact(void); extern void AtPrepare_MultiXact(void);
extern void PostPrepare_MultiXact(TransactionId xid); extern void PostPrepare_MultiXact(FullTransactionId fxid);
extern Size MultiXactShmemSize(void); extern Size MultiXactShmemSize(void);
extern void MultiXactShmemInit(void); extern void MultiXactShmemInit(void);
@ -145,11 +146,11 @@ extern void MultiXactAdvanceNextMXact(MultiXactId minMulti,
extern void MultiXactAdvanceOldest(MultiXactId oldestMulti, Oid oldestMultiDB); extern void MultiXactAdvanceOldest(MultiXactId oldestMulti, Oid oldestMultiDB);
extern int MultiXactMemberFreezeThreshold(void); extern int MultiXactMemberFreezeThreshold(void);
extern void multixact_twophase_recover(TransactionId xid, uint16 info, extern void multixact_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void multixact_twophase_postcommit(TransactionId xid, uint16 info, extern void multixact_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void multixact_twophase_postabort(TransactionId xid, uint16 info, extern void multixact_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void multixact_redo(XLogReaderState *record); extern void multixact_redo(XLogReaderState *record);

11
src/include/access/twophase.h
View File

@ -36,10 +36,10 @@ extern void PostPrepare_Twophase(void);
extern TransactionId TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid, extern TransactionId TwoPhaseGetXidByVirtualXID(VirtualTransactionId vxid,
bool *have_more); bool *have_more);
extern PGPROC *TwoPhaseGetDummyProc(TransactionId xid, bool lock_held); extern PGPROC *TwoPhaseGetDummyProc(FullTransactionId fxid, bool lock_held);
extern int TwoPhaseGetDummyProcNumber(TransactionId xid, bool lock_held); extern int TwoPhaseGetDummyProcNumber(FullTransactionId fxid, bool lock_held);
extern GlobalTransaction MarkAsPreparing(TransactionId xid, const char *gid, extern GlobalTransaction MarkAsPreparing(FullTransactionId fxid, const char *gid,
TimestampTz prepared_at, TimestampTz prepared_at,
Oid owner, Oid databaseid); Oid owner, Oid databaseid);
@ -56,8 +56,9 @@ extern void CheckPointTwoPhase(XLogRecPtr redo_horizon);
extern void FinishPreparedTransaction(const char *gid, bool isCommit); extern void FinishPreparedTransaction(const char *gid, bool isCommit);
extern void PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, extern void PrepareRedoAdd(FullTransactionId fxid, char *buf,
XLogRecPtr end_lsn, RepOriginId origin_id); XLogRecPtr start_lsn, XLogRecPtr end_lsn,
RepOriginId origin_id);
extern void PrepareRedoRemove(TransactionId xid, bool giveWarning); extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
extern void restoreTwoPhaseData(void); extern void restoreTwoPhaseData(void);
extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, extern bool LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,

4
src/include/access/twophase_rmgr.h
View File

@ -14,7 +14,9 @@
#ifndef TWOPHASE_RMGR_H #ifndef TWOPHASE_RMGR_H
#define TWOPHASE_RMGR_H #define TWOPHASE_RMGR_H
typedef void (*TwoPhaseCallback) (TransactionId xid, uint16 info, #include "access/transam.h"
typedef void (*TwoPhaseCallback) (FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
typedef uint8 TwoPhaseRmgrId; typedef uint8 TwoPhaseRmgrId;

4
src/include/pgstat.h
View File

@ -718,9 +718,9 @@ extern void pgstat_count_heap_delete(Relation rel);
extern void pgstat_count_truncate(Relation rel); extern void pgstat_count_truncate(Relation rel);
extern void pgstat_update_heap_dead_tuples(Relation rel, int delta); extern void pgstat_update_heap_dead_tuples(Relation rel, int delta);
extern void pgstat_twophase_postcommit(TransactionId xid, uint16 info, extern void pgstat_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void pgstat_twophase_postabort(TransactionId xid, uint16 info, extern void pgstat_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern PgStat_StatTabEntry *pgstat_fetch_stat_tabentry(Oid relid); extern PgStat_StatTabEntry *pgstat_fetch_stat_tabentry(Oid relid);

11
src/include/storage/lock.h
View File

@ -18,6 +18,7 @@
#error "lock.h may not be included from frontend code" #error "lock.h may not be included from frontend code"
#endif #endif
#include "access/transam.h"
#include "lib/ilist.h" #include "lib/ilist.h"
#include "storage/lockdefs.h" #include "storage/lockdefs.h"
#include "storage/lwlock.h" #include "storage/lwlock.h"
@ -581,7 +582,7 @@ extern bool LockHasWaiters(const LOCKTAG *locktag,
extern VirtualTransactionId *GetLockConflicts(const LOCKTAG *locktag, extern VirtualTransactionId *GetLockConflicts(const LOCKTAG *locktag,
LOCKMODE lockmode, int *countp); LOCKMODE lockmode, int *countp);
extern void AtPrepare_Locks(void); extern void AtPrepare_Locks(void);
extern void PostPrepare_Locks(TransactionId xid); extern void PostPrepare_Locks(FullTransactionId fxid);
extern bool LockCheckConflicts(LockMethod lockMethodTable, extern bool LockCheckConflicts(LockMethod lockMethodTable,
LOCKMODE lockmode, LOCKMODE lockmode,
LOCK *lock, PROCLOCK *proclock); LOCK *lock, PROCLOCK *proclock);
@ -597,13 +598,13 @@ extern BlockedProcsData *GetBlockerStatusData(int blocked_pid);
extern xl_standby_lock *GetRunningTransactionLocks(int *nlocks); extern xl_standby_lock *GetRunningTransactionLocks(int *nlocks);
extern const char *GetLockmodeName(LOCKMETHODID lockmethodid, LOCKMODE mode); extern const char *GetLockmodeName(LOCKMETHODID lockmethodid, LOCKMODE mode);
extern void lock_twophase_recover(TransactionId xid, uint16 info, extern void lock_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void lock_twophase_postcommit(TransactionId xid, uint16 info, extern void lock_twophase_postcommit(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void lock_twophase_postabort(TransactionId xid, uint16 info, extern void lock_twophase_postabort(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern void lock_twophase_standby_recover(TransactionId xid, uint16 info, extern void lock_twophase_standby_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
extern DeadLockState DeadLockCheck(PGPROC *proc); extern DeadLockState DeadLockCheck(PGPROC *proc);

6
src/include/storage/predicate.h
View File

@ -72,9 +72,9 @@ extern void PreCommit_CheckForSerializationFailure(void);
/* two-phase commit support */ /* two-phase commit support */
extern void AtPrepare_PredicateLocks(void); extern void AtPrepare_PredicateLocks(void);
extern void PostPrepare_PredicateLocks(TransactionId xid); extern void PostPrepare_PredicateLocks(FullTransactionId fxid);
extern void PredicateLockTwoPhaseFinish(TransactionId xid, bool isCommit); extern void PredicateLockTwoPhaseFinish(FullTransactionId xid, bool isCommit);
extern void predicatelock_twophase_recover(TransactionId xid, uint16 info, extern void predicatelock_twophase_recover(FullTransactionId fxid, uint16 info,
void *recdata, uint32 len); void *recdata, uint32 len);
/* parallel query support */ /* parallel query support */