diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index e2566a4..365cd17 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -1719,6 +1719,88 @@ 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. + */ +void +StandbyRecoverPreparedTransactions(bool can_overwrite) +{ + 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, and they may force us to advance nextXid. + */ + subxids = (TransactionId *) + (buf + MAXALIGN(sizeof(TwoPhaseFileHeader))); + for (i = 0; i < hdr->nsubxacts; i++) + { + TransactionId subxid = subxids[i]; + + Assert(TransactionIdFollows(subxid, xid)); + SubTransSetParent(xid, subxid, can_overwrite); + } + } + } + FreeDir(cldir); +} + +/* * RecoverPreparedTransactions * * Scan the pg_twophase directory and reload shared-memory state for each diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 3000ab7..d26b369 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -180,6 +180,7 @@ static TimestampTz recoveryLastXTime = 0; /* options taken from recovery.conf for XLOG streaming */ static bool StandbyMode = false; +static bool SnapshotMode = false; static char *PrimaryConnInfo = NULL; char *TriggerFile = NULL; @@ -3845,7 +3846,7 @@ next_record_is_invalid: } /* In standby-mode, keep trying */ - if (StandbyMode) + if (StandbyMode && !SnapshotMode) goto retry; else return NULL; @@ -5163,10 +5164,15 @@ readRecoveryCommandFile(void) if (StandbyMode) { if (PrimaryConnInfo == NULL && recoveryRestoreCommand == NULL) - ereport(WARNING, + { + if (XLogRequestRecoveryConnections) + SnapshotMode = true; + else + ereport(WARNING, (errmsg("recovery command file \"%s\" specified neither primary_conninfo nor restore_command", RECOVERY_COMMAND_FILE), errhint("The database server will regularly poll the pg_xlog subdirectory to check for files placed there."))); + } } else { @@ -5643,12 +5649,15 @@ StartupXLOG(void) if (InArchiveRecovery) { - if (StandbyMode) + if (SnapshotMode) + ereport(LOG, + (errmsg("entering snapshot mode"))); + else if (StandbyMode) ereport(LOG, (errmsg("entering standby mode"))); else if (recoveryTarget == RECOVERY_TARGET_XID) ereport(LOG, - (errmsg("starting point-in-time recovery to XID %u", + (errmsg("starting point-in-time recovery to XID %u", recoveryTargetXid))); else if (recoveryTarget == RECOVERY_TARGET_TIME) ereport(LOG, @@ -5880,6 +5889,33 @@ StartupXLOG(void) StartupMultiXact(); ProcArrayInitRecoveryInfo(oldestActiveXID); + + /* + * If we're beginning at a shutdown checkpoint, we know that + * nothing was running on the master at this point. So fake-up + * an empty running-xacts record and use that here and now. + * Recover additional standby state for prepared transactions. + */ + if (wasShutdown) + { + RunningTransactionsData running; + + /* + * Construct a RunningTransactions snapshot representing a shut + * down server, with only prepared transactions still alive. + * We're never overflowed at this point because all subxids + * are listed with their parent prepared transactions. + */ + running.xcnt = nxids; + running.subxid_overflow = false; + running.nextXid = checkPoint.nextXid; + running.oldestRunningXid = oldestActiveXID; + running.xids = xids; + + ProcArrayApplyRecoveryInfo(&running); + + StandbyRecoverPreparedTransactions(false); + } } /* Initialize resource managers */ @@ -6070,6 +6106,40 @@ StartupXLOG(void) timestamptz_to_str(recoveryLastXTime)))); InRedo = false; } + else if (SnapshotMode) + { + if (standbyState == STANDBY_SNAPSHOT_READY && + IsUnderPostmaster && + XLByteLE(minRecoveryPoint, EndRecPtr) && + XLogRecPtrIsInvalid(ControlFile->backupStartPoint)) + { + SetForwardFsyncRequests(); + SendPostmasterSignal(PMSIGNAL_RECOVERY_STARTED); + + ereport(LOG, + (errmsg("consistent recovery state reached at %X/%X", + EndRecPtr.xlogid, EndRecPtr.xrecoff))); + + pg_usleep(2000000L); //wait for Postmaster to absorb signal + + SendPostmasterSignal(PMSIGNAL_RECOVERY_CONSISTENT); + + /* + * Cue the elevator music... + */ + while (true) + { + pg_usleep(5000000L); + if (CheckForStandbyTrigger()) + break; + + } + } + else + ereport(LOG, + (errmsg("unable to reach consistent recovery state at %X/%X", + EndRecPtr.xlogid, EndRecPtr.xrecoff))); + } else { /* there are no WAL records following the checkpoint */ @@ -7592,13 +7662,34 @@ xlog_redo(XLogRecPtr lsn, XLogRecord *record) if (standbyState != STANDBY_DISABLED) CheckRequiredParameterValues(checkPoint); + /* + * If we're beginning at a shutdown checkpoint, we know that + * nothing was running on the master at this point. So fake-up + * an empty running-xacts record and use that here and now. + * Recover additional standby state for prepared transactions. + */ if (standbyState >= STANDBY_INITIALIZED) { + TransactionId *xids; + int nxids; + TransactionId oldestActiveXID = PrescanPreparedTransactions(&xids, &nxids); + RunningTransactionsData running; + /* - * Remove stale transactions, if any. + * Construct a RunningTransactions snapshot representing a shut + * down server, with only prepared transactions still alive. + * We're never overflowed at this point because all subxids + * are listed with their parent prepared transactions. */ - ExpireOldKnownAssignedTransactionIds(checkPoint.nextXid); - StandbyReleaseOldLocks(checkPoint.nextXid); + running.xcnt = nxids; + running.subxid_overflow = false; + running.nextXid = checkPoint.nextXid; + running.oldestRunningXid = oldestActiveXID; + running.xids = xids; + + ProcArrayApplyRecoveryInfo(&running); + + StandbyRecoverPreparedTransactions(true); } /* ControlFile->checkPointCopy always tracks the latest ckpt XID */ diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 5d9d6cf..46f5fca 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -44,6 +44,7 @@ extern bool StandbyTransactionIdIsPrepared(TransactionId xid); extern TransactionId PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p); +extern void StandbyRecoverPreparedTransactions(bool can_overwrite); extern void RecoverPreparedTransactions(void); extern void RecreateTwoPhaseFile(TransactionId xid, void *content, int len);