diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c index d13d2e7..488a49a 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -211,6 +211,10 @@ typedef struct MultiXactStateData MultiXactId multiStopLimit; MultiXactId multiWrapLimit; + /* support for members anti-wraparound measures */ + MultiXactOffset offsetWarnLimit; + MultiXactOffset offsetStopLimit; + /* * Per-backend data starts here. We have two arrays stored in the area * immediately following the MultiXactStateData struct. Each is indexed by @@ -339,6 +343,8 @@ static bool MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2); static void ExtendMultiXactOffset(MultiXactId multi); static void ExtendMultiXactMember(MultiXactOffset offset, int nmembers); +static MultiXactOffset read_offset_for_multi(MultiXactId multi); +static void DetermineSafeOldestOffset(MultiXactId oldestMXact); static void WriteMZeroPageXlogRec(int pageno, uint8 info); @@ -975,7 +981,7 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset) /* * To avoid swamping the postmaster with signals, we issue the autovac - * request only once per 64K transaction starts. This still gives + * request only once per 64K multis generated. This still gives * plenty of chances before we get into real trouble. */ if (IsUnderPostmaster && (result % 65536) == 0) @@ -1051,6 +1057,36 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset) else *offset = nextOffset; + /* + * Protect against overrun of the members space as well, with the following + * rules: + * + * - if we're past offsetWarnLimit, emit a warning. + * - if we're past offsetStopLimit, refuse to generate more multis. + * + * Note we haven't updated the shared state yet, so if we fail at this + * point, the multixact ID we grabbed can still be used by the next guy. + * + * Note that thre is no point in forcing autovacuum runs here; the + * multixact freeze settings would have to be reduced for that to have any + * effect, and we can't do that from here. + */ + if (MultiXactOffsetPrecedes(MultiXactState->offsetStopLimit, nextOffset + nmembers)) + ereport(ERROR, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("database is not accepting commands that generate new MultiXactIds to avoid \"members\" wraparound data loss in database with OID %u", + MultiXactState->oldestMultiXactDB), + errhint("Execute a database-wide VACUUM in that database, with reduced vacuum_multixact_freeze_min_age and vacuum_multixact_freeze_table_age settings."))); + else if (MultiXactOffsetPrecedes(MultiXactState->offsetWarnLimit, nextOffset + nmembers)) + ereport(WARNING, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg_plural("database with OID %u must be vacuumed before %d more multixact member is used", + "database with OID %u must be vacuumed before %d more multixact members are used", + MultiXactState->offsetStopLimit - nextOffset + nmembers, + MultiXactState->oldestMultiXactDB, + MultiXactState->offsetStopLimit - nextOffset + nmembers), + errhint("Execute a database-wide VACUUM in that database, with reduced vacuum_multixact_freeze_min_age and vacuum_multixact_freeze_table_age settings."))); + ExtendMultiXactMember(nextOffset, nmembers); /* @@ -1921,6 +1957,12 @@ StartupMultiXact(void) */ pageno = MXOffsetToMemberPage(offset); MultiXactMemberCtl->shared->latest_page_number = pageno; + + /* + * compute the oldest member we need to keep around to avoid old member + * data overrun. + */ + DetermineSafeOldestOffset(MultiXactState->oldestMultiXactId); } /* @@ -2121,7 +2163,7 @@ SetMultiXactIdLimit(MultiXactId oldest_datminmxid, Oid oldest_datoid) * * Note: This differs from the magic number used in * SetTransactionIdLimit() since vacuum itself will never generate new - * multis. + * multis. XXX actually it does, if it needs to freeze old multis. */ multiStopLimit = multiWrapLimit - 100; if (multiStopLimit < FirstMultiXactId) @@ -2164,6 +2206,8 @@ SetMultiXactIdLimit(MultiXactId oldest_datminmxid, Oid oldest_datoid) curMulti = MultiXactState->nextMXact; LWLockRelease(MultiXactGenLock); + DetermineSafeOldestOffset(oldest_datminmxid); + /* Log the info */ ereport(DEBUG1, (errmsg("MultiXactId wrap limit is %u, limited by database with OID %u", @@ -2250,13 +2294,16 @@ MultiXactAdvanceNextMXact(MultiXactId minMulti, /* * Update our oldestMultiXactId value, but only if it's more recent than - * what we had. + * what we had. However, even if not, always update the oldest multixact + * offset limit. */ void MultiXactAdvanceOldest(MultiXactId oldestMulti, Oid oldestMultiDB) { if (MultiXactIdPrecedes(MultiXactState->oldestMultiXactId, oldestMulti)) SetMultiXactIdLimit(oldestMulti, oldestMultiDB); + else + DetermineSafeOldestOffset(oldestMulti); } /* @@ -2424,6 +2471,70 @@ GetOldestMultiXactId(void) } /* + * Read the offset of the first member of the given multixact. + */ +static MultiXactOffset +read_offset_for_multi(MultiXactId multi) +{ + MultiXactOffset offset; + int pageno; + int entryno; + int slotno; + MultiXactOffset *offptr; + + pageno = MultiXactIdToOffsetPage(multi); + entryno = MultiXactIdToOffsetEntry(multi); + + /* lock is acquired by SimpleLruReadPage_ReadOnly */ + slotno = SimpleLruReadPage_ReadOnly(MultiXactOffsetCtl, pageno, multi); + offptr = (MultiXactOffset *) MultiXactOffsetCtl->shared->page_buffer[slotno]; + offptr += entryno; + offset = *offptr; + LWLockRelease(MultiXactOffsetControlLock); + + return offset; +} + +/* + * Based on the given oldest MultiXactId, determine what's the oldest member + * offset and install the limit info in MultiXactState, where it can be used to + * prevent overrun of old data in the members SLRU area. + */ +static void +DetermineSafeOldestOffset(MultiXactId oldestMXact) +{ + MultiXactOffset oldestOffset; + + /* + * Can't do this while initdb'ing or in the startup process while replaying + * WAL: the segment file to read might have not yet been created, or + * already been removed. + */ + if (IsBootstrapProcessingMode() || InRecovery) + return; + + /* + * We determine the safe upper bound for offsets of new xacts by reading + * the offset of the oldest multixact, and going back one segment. This + * way, the sequence of multixact member segments will always have a + * one-segment hole at a minimum. We start spewing warnings 19 segments + * before that. + */ + oldestOffset = read_offset_for_multi(oldestMXact); + /* move back to start of this segment */ + oldestOffset -= oldestOffset / MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT; + + LWLockAcquire(MultiXactGenLock, LW_EXCLUSIVE); + /* always leave one segment before the wraparound point */ + MultiXactState->offsetStopLimit = oldestOffset - + (MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT); + /* start complaining 20 segments before the wraparound point */ + MultiXactState->offsetWarnLimit = oldestOffset - + (MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT * 20); + LWLockRelease(MultiXactGenLock); +} + +/* * SlruScanDirectory callback. * This callback deletes segments that are outside the range determined by * the given page numbers. @@ -2555,26 +2666,7 @@ TruncateMultiXact(void) * First, compute the safe truncation point for MultiXactMember. This is * the starting offset of the oldest multixact. */ - { - int pageno; - int slotno; - int entryno; - MultiXactOffset *offptr; - - /* lock is acquired by SimpleLruReadPage_ReadOnly */ - - pageno = MultiXactIdToOffsetPage(oldestMXact); - entryno = MultiXactIdToOffsetEntry(oldestMXact); - - slotno = SimpleLruReadPage_ReadOnly(MultiXactOffsetCtl, pageno, - oldestMXact); - offptr = (MultiXactOffset *) - MultiXactOffsetCtl->shared->page_buffer[slotno]; - offptr += entryno; - oldestOffset = *offptr; - - LWLockRelease(MultiXactOffsetControlLock); - } + oldestOffset = read_offset_for_multi(oldestMXact); /* * To truncate MultiXactMembers, we need to figure out the active page diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c index fc1b3c8..43a50e4 100644 --- a/src/backend/bootstrap/bootstrap.c +++ b/src/backend/bootstrap/bootstrap.c @@ -411,6 +411,7 @@ AuxiliaryProcessMain(int argc, char *argv[]) proc_exit(1); /* should never return */ case BootstrapProcess: + SetProcessingMode(BootstrapProcessing); bootstrap_signals(); BootStrapXLOG(); BootstrapModeMain(); @@ -473,8 +474,7 @@ BootstrapModeMain(void) int i; Assert(!IsUnderPostmaster); - - SetProcessingMode(BootstrapProcessing); + Assert(IsBootstrapProcessingMode()); /* * Do backend-like initialization for bootstrap mode