diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 8a22836..7119f22 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -45,8 +45,8 @@ * fsynced * * If COMMIT happens after checkpoint then backend reads state data from * files - * * In case of crash replay will move data from xlog to files, if that - * hasn't happened before. XXX TODO - move to shmem in replay also + * + * The same procedure happens during replication and crash recovery. * *------------------------------------------------------------------------- */ @@ -1248,8 +1248,6 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len) XLogReaderState *xlogreader; char *errormsg; - Assert(!RecoveryInProgress()); - xlogreader = XLogReaderAllocate(&read_local_xlog_page, NULL); if (!xlogreader) ereport(ERROR, @@ -1292,12 +1290,30 @@ StandbyTransactionIdIsPrepared(TransactionId xid) char *buf; TwoPhaseFileHeader *hdr; bool result; + int i; Assert(TransactionIdIsValid(xid)); if (max_prepared_xacts <= 0) return false; /* nothing to do */ + /* + * At first check prepared tx that have no saved state files. + */ + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + PGXACT *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno]; + + if (TransactionIdEquals(pgxact->xid, xid)) + { + LWLockRelease(TwoPhaseStateLock); + return true; + } + } + LWLockRelease(TwoPhaseStateLock); + /* Read and validate file */ buf = ReadTwoPhaseFile(xid, false); if (buf == NULL) @@ -1571,6 +1587,109 @@ RecreateTwoPhaseFile(TransactionId xid, void *content, int len) errmsg("could not close two-phase state file: %m"))); } + +/* + * XlogRedoFinishPrepared() + * + * This function is called during replay when xlog reader faces 2pc commit or + * abort record. That function should clean up memory state that was created + * while replaying prepare xlog record. + * Actions are the same as in FinishPreparedTransaction() but without any + * writes to xlog and files (as it was already done). + */ +void +XlogRedoFinishPrepared(TransactionId xid) +{ + int i; + char *buf; + char *bufptr; + TwoPhaseFileHeader *hdr; + TransactionId latestXid; + TransactionId *children; + + GlobalTransaction gxact; + PGPROC *proc; + PGXACT *pgxact; + + /* During replay that lock isn't really necessary, but let's take it anyway */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + gxact = TwoPhaseState->prepXacts[i]; + proc = &ProcGlobal->allProcs[gxact->pgprocno]; + pgxact = &ProcGlobal->allPgXact[gxact->pgprocno]; + + if (TransactionIdEquals(xid, pgxact->xid)) + { + gxact->locking_backend = MyBackendId; + MyLockedGxact = gxact; + break; + } + } + LWLockRelease(TwoPhaseStateLock); + + /* + * If requested xid isn't in numPrepXacts array that means that prepare + * record was moved to files before our replay started. That's okay and we + * have nothing to clean. + */ + if (i == TwoPhaseState->numPrepXacts) + return; + + if (gxact->ondisk) + buf = ReadTwoPhaseFile(xid, true); + else + XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); + + /* + * Disassemble the header area + */ + hdr = (TwoPhaseFileHeader *) buf; + + Assert(TransactionIdEquals(hdr->xid, xid)); + + bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader)); + children = (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)); + + /* + * Here we don't need to care about putting records to xlog or + * deleting files, as it already done by process that have written + * that xlog record. We need just to clen up memmory state. + */ + + latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children); + ProcArrayRemove(proc, latestXid); + gxact->valid = false; + + /* + * 2REVIEWER: I assume that we can skip invalidation callbacks here, + * aren't we? + */ + + /* And release locks */ + if (true) + ProcessRecords(bufptr, xid, twophase_postcommit_callbacks); + else + ProcessRecords(bufptr, xid, twophase_postabort_callbacks); + PredicateLockTwoPhaseFinish(xid, true); + RemoveGXact(gxact); + MyLockedGxact = NULL; + + /* + * And now we can clean up any files we may have left. + */ + if (gxact->ondisk) + RemoveTwoPhaseFile(xid, true); + + pfree(buf); +} + + + /* * CheckPointTwoPhase -- handle 2PC component of checkpointing. * @@ -1685,7 +1804,51 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p) TransactionId *xids = NULL; int nxids = 0; int allocsize = 0; + int i; + + /* + * This is usually called after end-of-recovery checkpoint, so all 2pc + * files moved xlog to files. But if we restart slave when master is + * switched off this function will be called before checkpoint ans we need + * to check PGXACT array as it can contain prepared transactions that + * didn't created any state files yet. + */ + 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) + continue; + if (TransactionIdPrecedes(pgxact->xid, result)) + result = pgxact->xid; + + if (xids_p) + { + if (nxids == allocsize) + { + if (nxids == 0) + { + allocsize = 10; + xids = palloc(allocsize * sizeof(TransactionId)); + } + else + { + allocsize = allocsize * 2; + xids = repalloc(xids, allocsize * sizeof(TransactionId)); + } + } + xids[nxids++] = pgxact->xid; + } + } + LWLockRelease(TwoPhaseStateLock); + + + /* + * And now scan files in pg_twophase directory + */ cldir = AllocateDir(TWOPHASE_DIR); while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL) { @@ -1696,7 +1859,6 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p) char *buf; TwoPhaseFileHeader *hdr; TransactionId *subxids; - int i; xid = (TransactionId) strtoul(clde->d_name, NULL, 16); @@ -1804,97 +1966,14 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p) } /* - * 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 + * RecoverPreparedFromFiles * * 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) +RecoverPreparedFromFiles(bool forceOverwriteOK) { char dir[MAXPGPATH]; DIR *cldir; @@ -1916,9 +1995,25 @@ RecoverPreparedTransactions(void) TransactionId *subxids; GlobalTransaction gxact; int i; + PGXACT *pgxact; xid = (TransactionId) strtoul(clde->d_name, NULL, 16); + /* Already recovered from WAL? */ + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + gxact = TwoPhaseState->prepXacts[i]; + pgxact = &ProcGlobal->allPgXact[gxact->pgprocno]; + + if (TransactionIdEquals(xid, pgxact->xid)) + { + LWLockRelease(TwoPhaseStateLock); + goto next_file; + } + } + LWLockRelease(TwoPhaseStateLock); + /* Already processed? */ if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid)) { @@ -1963,6 +2058,12 @@ RecoverPreparedTransactions(void) overwriteOK = true; /* + * Caller can also force overwriteOK. + */ + if (forceOverwriteOK) + 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 @@ -2004,10 +2105,94 @@ RecoverPreparedTransactions(void) pfree(buf); } + +next_file: + continue; + } FreeDir(cldir); } + +void +RecoverPreparedFromXLOG(XLogReaderState *record) +{ + bool overwriteOK = false; + TransactionId xid = XLogRecGetXid(record); + char *buf = (char *) XLogRecGetData(record); + char *bufptr; + TwoPhaseFileHeader *hdr; + TransactionId *subxids; + GlobalTransaction gxact; + int i; + + /* 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 || + XLogLogicalInfoActive())) + 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 + * + * MarkAsPreparing sets prepare_start_lsn to InvalidXLogRecPtr + * so next checkpoint will skip that transaction. + */ + gxact = MarkAsPreparing(xid, hdr->gid, + hdr->prepared_at, + hdr->owner, hdr->database); + GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids); + MarkAsPrepared(gxact); + + gxact->prepare_start_lsn = record->ReadRecPtr; + gxact->prepare_end_lsn = record->EndRecPtr; + + /* + * Recover other state (notably locks) using resource managers + */ + ProcessRecords(bufptr, xid, twophase_recover_callbacks); + + /* + * Release locks held by the standby process after we process each + * prepared transaction. As a result, we don't need too many + * additional locks at any one time. + */ + if (InHotStandby) + StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids); + + /* + * We're done with recovering this transaction. Clear + * MyLockedGxact, like we do in PrepareTransaction() during normal + * operation. + */ + PostPrepare_Twophase(); +} + + + /* * RecordTransactionCommitPrepared * diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index b0d5440..5e9ad8d 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -5567,7 +5567,7 @@ xact_redo(XLogReaderState *record) Assert(TransactionIdIsValid(parsed.twophase_xid)); xact_redo_commit(&parsed, parsed.twophase_xid, record->EndRecPtr, XLogRecGetOrigin(record)); - RemoveTwoPhaseFile(parsed.twophase_xid, false); + XlogRedoFinishPrepared(parsed.twophase_xid); } } else if (info == XLOG_XACT_ABORT || info == XLOG_XACT_ABORT_PREPARED) @@ -5587,14 +5587,21 @@ xact_redo(XLogReaderState *record) { Assert(TransactionIdIsValid(parsed.twophase_xid)); xact_redo_abort(&parsed, parsed.twophase_xid); - RemoveTwoPhaseFile(parsed.twophase_xid, false); + XlogRedoFinishPrepared(parsed.twophase_xid); } } else if (info == XLOG_XACT_PREPARE) { - /* the record contents are exactly the 2PC file */ - RecreateTwoPhaseFile(XLogRecGetXid(record), - XLogRecGetData(record), XLogRecGetDataLen(record)); + /* + * To avoid creation of state files during replay we registering + * prepare xlog resords in shared memory in the same way as it happens + * while not in recovery. If replay faces commit xlog record before + * checkpoint/restartpoint happens then we avoid using files at all. + * + * We need such behaviour because speed of 2PC replay on replica should + * be at least not slower than 2PC tx speed on master. + */ + RecoverPreparedFromXLOG(record); } else if (info == XLOG_XACT_ASSIGNMENT) { diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index a2846c4..4d5fee5 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -6605,7 +6605,7 @@ StartupXLOG(void) ProcArrayApplyRecoveryInfo(&running); - StandbyRecoverPreparedTransactions(false); + RecoverPreparedFromFiles(false); } } @@ -7351,7 +7351,7 @@ StartupXLOG(void) TrimMultiXact(); /* Reload shared-memory state for prepared transactions */ - RecoverPreparedTransactions(); + RecoverPreparedFromFiles(false); /* * Shutdown the recovery environment. This must occur after @@ -9269,7 +9269,7 @@ xlog_redo(XLogReaderState *record) ProcArrayApplyRecoveryInfo(&running); - StandbyRecoverPreparedTransactions(true); + RecoverPreparedFromFiles(true); } /* ControlFile->checkPointCopy always tracks the latest ckpt XID */ diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index b7ce0c6..6b7bc7d 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -17,6 +17,7 @@ #include "access/xlogdefs.h" #include "datatype/timestamp.h" #include "storage/lock.h" +#include "access/xlogreader.h" /* * GlobalTransactionData is defined in twophase.c; other places have no @@ -46,8 +47,8 @@ extern bool StandbyTransactionIdIsPrepared(TransactionId xid); extern TransactionId PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p); -extern void StandbyRecoverPreparedTransactions(bool overwriteOK); -extern void RecoverPreparedTransactions(void); +extern void RecoverPreparedFromFiles(bool overwriteOK); +extern void RecoverPreparedFromXLOG(XLogReaderState *record); extern void RecreateTwoPhaseFile(TransactionId xid, void *content, int len); extern void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning); @@ -56,4 +57,5 @@ extern void CheckPointTwoPhase(XLogRecPtr redo_horizon); extern void FinishPreparedTransaction(const char *gid, bool isCommit); +extern void XlogRedoFinishPrepared(TransactionId xid); #endif /* TWOPHASE_H */