From 6fa76418466a511a8af32fc3032b4f2ac9daae77 Mon Sep 17 00:00:00 2001 From: Vitaly Davydov Date: Wed, 22 Jan 2025 11:12:32 +0300 Subject: [PATCH] Use FullTransactionId in TwoPhaseFilePath and in some other functions The patch fixes TwoPhaseFilePath function. The current version gets TransactionId as an argument and makes a path to the file with twophase transaction state. The problem - this logic is ambiguous, because xid doesn't contain the epoch. By replacing the argument with FullTransactionId, the logic of this function becomes straightforward. All the responsibility to properly form a full xid were delegated to the caller functions. --- src/backend/access/transam/twophase.c | 84 ++++++++++++++++----------- 1 file changed, 51 insertions(+), 33 deletions(-) diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index ab2f4a8a92..e215170cca 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -221,14 +221,14 @@ static void ProcessRecords(char *bufptr, TransactionId xid, static void RemoveGXact(GlobalTransaction gxact); static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len); -static char *ProcessTwoPhaseBuffer(TransactionId xid, +static char *ProcessTwoPhaseBuffer(FullTransactionId xid, XLogRecPtr prepare_start_lsn, bool fromdisk, bool setParent, bool setNextXid); static void MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid, TimestampTz prepared_at, Oid owner, Oid databaseid); -static void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning); -static void RecreateTwoPhaseFile(TransactionId xid, void *content, int len); +static void RemoveTwoPhaseFile(FullTransactionId xid, bool giveWarning); +static void RecreateTwoPhaseFile(FullTransactionId xid, void *content, int len); /* * Initialization of shared memory @@ -958,10 +958,8 @@ AdjustToFullTransactionId(TransactionId xid) } 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", EpochFromFullTransactionId(fxid), XidFromFullTransactionId(fxid)); @@ -1300,7 +1298,7 @@ RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info, * ignored, then return NULL. This state can be reached when doing recovery. */ static char * -ReadTwoPhaseFile(TransactionId xid, bool missing_ok) +ReadTwoPhaseFile(FullTransactionId xid, bool missing_ok) { char path[MAXPGPATH]; char *buf; @@ -1477,14 +1475,17 @@ StandbyTransactionIdIsPrepared(TransactionId xid) char *buf; TwoPhaseFileHeader *hdr; bool result; + FullTransactionId fxid; Assert(TransactionIdIsValid(xid)); if (max_prepared_xacts <= 0) return false; /* nothing to do */ + fxid = AdjustToFullTransactionId(xid); + /* Read and validate file */ - buf = ReadTwoPhaseFile(xid, true); + buf = ReadTwoPhaseFile(fxid, true); if (buf == NULL) return false; @@ -1518,6 +1519,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit) xl_xact_stats_item *commitstats; xl_xact_stats_item *abortstats; SharedInvalidationMessage *invalmsgs; + FullTransactionId fxid = InvalidFullTransactionId; /* * Validate the GID, and lock the GXACT to ensure that two backends do not @@ -1532,8 +1534,10 @@ FinishPreparedTransaction(const char *gid, bool isCommit) * in WAL files if the LSN is after the last checkpoint record, or moved * to disk if for some reason they have lived for a long time. */ - if (gxact->ondisk) - buf = ReadTwoPhaseFile(xid, false); + if (gxact->ondisk) { + fxid = AdjustToFullTransactionId(xid); + buf = ReadTwoPhaseFile(fxid, false); + } else XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); @@ -1679,8 +1683,10 @@ FinishPreparedTransaction(const char *gid, bool isCommit) /* * And now we can clean up any files we may have left. */ - if (ondisk) - RemoveTwoPhaseFile(xid, true); + if (ondisk) { + Assert(FullTransactionIdIsValid(fxid)); + RemoveTwoPhaseFile(fxid, true); + } MyLockedGxact = NULL; @@ -1720,11 +1726,11 @@ ProcessRecords(char *bufptr, TransactionId xid, * this is an expected case during WAL replay. */ static void -RemoveTwoPhaseFile(TransactionId xid, bool giveWarning) +RemoveTwoPhaseFile(FullTransactionId fxid, bool giveWarning) { char path[MAXPGPATH]; - TwoPhaseFilePath(path, xid); + TwoPhaseFilePath(path, fxid); if (unlink(path)) if (errno != ENOENT || giveWarning) ereport(WARNING, @@ -1739,7 +1745,7 @@ RemoveTwoPhaseFile(TransactionId xid, bool giveWarning) * Note: content and len don't include CRC. */ static void -RecreateTwoPhaseFile(TransactionId xid, void *content, int len) +RecreateTwoPhaseFile(FullTransactionId xid, void *content, int len) { char path[MAXPGPATH]; pg_crc32c statefile_crc; @@ -1860,9 +1866,10 @@ CheckPointTwoPhase(XLogRecPtr redo_horizon) { char *buf; int len; + FullTransactionId fxid = AdjustToFullTransactionId(gxact->xid); XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len); - RecreateTwoPhaseFile(gxact->xid, buf, len); + RecreateTwoPhaseFile(fxid, buf, len); gxact->ondisk = true; gxact->prepare_start_lsn = InvalidXLogRecPtr; gxact->prepare_end_lsn = InvalidXLogRecPtr; @@ -1913,14 +1920,12 @@ restoreTwoPhaseData(void) if (strlen(clde->d_name) == 16 && strspn(clde->d_name, "0123456789ABCDEF") == 16) { - TransactionId xid; FullTransactionId fxid; char *buf; fxid = FullTransactionIdFromU64(strtou64(clde->d_name, NULL, 16)); - xid = XidFromFullTransactionId(fxid); - buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr, + buf = ProcessTwoPhaseBuffer(fxid, InvalidXLogRecPtr, true, false, false); if (buf == NULL) continue; @@ -1981,12 +1986,14 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p) TransactionId xid; char *buf; GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + FullTransactionId fxid; Assert(gxact->inredo); xid = gxact->xid; + fxid = AdjustToFullTransactionId(xid); - buf = ProcessTwoPhaseBuffer(xid, + buf = ProcessTwoPhaseBuffer(fxid, gxact->prepare_start_lsn, gxact->ondisk, false, true); @@ -2055,12 +2062,14 @@ StandbyRecoverPreparedTransactions(void) TransactionId xid; char *buf; GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + FullTransactionId fxid; Assert(gxact->inredo); xid = gxact->xid; + fxid = AdjustToFullTransactionId(xid); - buf = ProcessTwoPhaseBuffer(xid, + buf = ProcessTwoPhaseBuffer(fxid, gxact->prepare_start_lsn, gxact->ondisk, true, false); if (buf != NULL) @@ -2100,8 +2109,10 @@ RecoverPreparedTransactions(void) TwoPhaseFileHeader *hdr; TransactionId *subxids; const char *gid; + FullTransactionId fxid; xid = gxact->xid; + fxid = AdjustToFullTransactionId(xid); /* * Reconstruct subtrans state for the transaction --- needed because @@ -2112,7 +2123,7 @@ RecoverPreparedTransactions(void) * SubTransSetParent has been set before, if the prepared transaction * generated xid assignment records. */ - buf = ProcessTwoPhaseBuffer(xid, + buf = ProcessTwoPhaseBuffer(fxid, gxact->prepare_start_lsn, gxact->ondisk, true, false); if (buf == NULL) @@ -2189,13 +2200,13 @@ RecoverPreparedTransactions(void) * value scanned. */ static char * -ProcessTwoPhaseBuffer(TransactionId xid, +ProcessTwoPhaseBuffer(FullTransactionId fxid, XLogRecPtr prepare_start_lsn, bool fromdisk, bool setParent, bool setNextXid) { FullTransactionId nextXid = TransamVariables->nextXid; - TransactionId origNextXid = XidFromFullTransactionId(nextXid); + TransactionId xid = XidFromFullTransactionId(fxid); TransactionId *subxids; char *buf; TwoPhaseFileHeader *hdr; @@ -2214,7 +2225,7 @@ ProcessTwoPhaseBuffer(TransactionId xid, ereport(WARNING, (errmsg("removing stale two-phase state file for transaction %u", xid))); - RemoveTwoPhaseFile(xid, true); + RemoveTwoPhaseFile(fxid, true); } else { @@ -2227,14 +2238,14 @@ ProcessTwoPhaseBuffer(TransactionId xid, } /* Reject XID if too new */ - if (TransactionIdFollowsOrEquals(xid, origNextXid)) + if (FullTransactionIdFollowsOrEquals(fxid, nextXid)) { if (fromdisk) { ereport(WARNING, (errmsg("removing future two-phase state file for transaction %u", xid))); - RemoveTwoPhaseFile(xid, true); + RemoveTwoPhaseFile(fxid, true); } else { @@ -2249,7 +2260,7 @@ ProcessTwoPhaseBuffer(TransactionId xid, if (fromdisk) { /* Read and validate file */ - buf = ReadTwoPhaseFile(xid, false); + buf = ReadTwoPhaseFile(fxid, false); } else { @@ -2520,8 +2531,9 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, if (!XLogRecPtrIsInvalid(start_lsn)) { char path[MAXPGPATH]; + FullTransactionId fxid = AdjustToFullTransactionId(hdr->xid); - TwoPhaseFilePath(path, hdr->xid); + TwoPhaseFilePath(path, fxid); if (access(path, F_OK) == 0) { @@ -2615,8 +2627,11 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning) * And now we can clean up any files we may have left. */ elog(DEBUG2, "removing 2PC data for transaction %u", xid); - if (gxact->ondisk) - RemoveTwoPhaseFile(xid, giveWarning); + if (gxact->ondisk) { + FullTransactionId fxid = AdjustToFullTransactionId(xid); + + RemoveTwoPhaseFile(fxid, giveWarning); + } RemoveGXact(gxact); } @@ -2663,8 +2678,11 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, * do this optimization if we encounter many collisions in GID * between publisher and subscriber. */ - if (gxact->ondisk) - buf = ReadTwoPhaseFile(gxact->xid, false); + if (gxact->ondisk) { + FullTransactionId fxid = AdjustToFullTransactionId(gxact->xid); + + buf = ReadTwoPhaseFile(fxid, false); + } else { Assert(gxact->prepare_start_lsn); -- 2.34.1