From 0e39b2cc4cc2c716847a536abc944be4823f3e11 Mon Sep 17 00:00:00 2001 From: Vignesh Date: Mon, 2 Jun 2025 11:37:24 +0530 Subject: [PATCH v8-PG16] Fix exponential memory allocation issue in logical decoding due to invalidation message distribution Commit 9a2f8b4f01 added logic to distribute invalidation messages from one transaction to all concurrent in-progress transactions. However, since the source of each invalidation message was not tracked, transactions could redundantly re-distribute already propagated messages. This led to an exponential increase in invalidation messages and, ultimately, memory allocation failures. To fix this, a new field is introduced to track distributed invalidation messages separately. This ensures that only transactions in the pre-existing list receive the invalidations, preventing redundant propagation. The accumulated distributed invalidations are now consumed at the end of transaction processing in `ReorderBufferProcessTXN()`. --- .../replication/logical/reorderbuffer.c | 175 ++++++++++++++---- src/backend/replication/logical/snapbuild.c | 5 +- src/include/replication/reorderbuffer.h | 17 ++ 3 files changed, 158 insertions(+), 39 deletions(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index fa04e829cc9..43b5583b1f8 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -103,6 +103,7 @@ #include "storage/sinval.h" #include "utils/builtins.h" #include "utils/combocid.h" +#include "utils/inval.h" #include "utils/memdebug.h" #include "utils/memutils.h" #include "utils/rel.h" @@ -461,6 +462,12 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) txn->invalidations = NULL; } + if (txn->distributed_invalidations) + { + pfree(txn->distributed_invalidations); + txn->distributed_invalidations = NULL; + } + /* Reset the toast hash */ ReorderBufferToastReset(rb, txn); @@ -2539,7 +2546,14 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, AbortCurrentTransaction(); /* make sure there's no cache pollution */ - ReorderBufferExecuteInvalidations(txn->ninvalidations, txn->invalidations); + if (rbtxn_inval_all_cache(txn)) + InvalidateSystemCaches(); + else + { + ReorderBufferExecuteInvalidations(txn->ninvalidations, txn->invalidations); + ReorderBufferExecuteInvalidations(txn->ninvalidations_distr, + txn->distributed_invalidations); + } if (using_subtxn) RollbackAndReleaseCurrentSubTransaction(); @@ -2615,6 +2629,20 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, errdata = NULL; curtxn->concurrent_abort = true; + /* + * Make sure there's no cache pollution. Unlike the PG_TRY section, + * this must be done unconditionally, since processing may fail + * before we reach the invalidation messages. This isn't needed + * in non-rollback cases, as the system cache is cleared either + * by the API's error handling or when the walsender exits. + */ + if (rbtxn_inval_all_cache(txn)) + InvalidateSystemCaches(); + else + ReorderBufferExecuteInvalidations(txn->ninvalidations_distr, + txn->distributed_invalidations); + + /* Reset the TXN so that it is allowed to stream remaining data. */ ReorderBufferResetTXN(rb, txn, snapshot_now, command_id, prev_lsn, @@ -3258,58 +3286,57 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, } /* - * Accumulate the invalidations for executing them later. + * Helper function to store invalidation messages and queue them as a + * change in the reorder buffer. Handles both regular and distributed + * invalidations. * - * This needs to be called for each XLOG_XACT_INVALIDATIONS message and - * accumulates all the invalidation messages in the toplevel transaction, if - * available, otherwise in the current transaction, as well as in the form of - * change in reorder buffer. We require to record it in form of the change + * This accumulates all the invalidation messages in the toplevel transaction, + * if available, otherwise in the current transaction, as well as in the form + * of change in reorder buffer. We require to record it in form of the change * so that we can execute only the required invalidations instead of executing * all the invalidations on each CommandId increment. We also need to * accumulate these in the txn buffer because in some cases where we skip * processing the transaction (see ReorderBufferForget), we need to execute * all the invalidations together. */ -void -ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, - XLogRecPtr lsn, Size nmsgs, - SharedInvalidationMessage *msgs) +static void +ReorderBufferAddInvalidationsCommon(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs, + ReorderBufferTXN *txn, + bool for_inval) { - ReorderBufferTXN *txn; MemoryContext oldcontext; ReorderBufferChange *change; - txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); - oldcontext = MemoryContextSwitchTo(rb->context); - /* - * Collect all the invalidations under the top transaction, if available, - * so that we can execute them all together. See comments atop this - * function. - */ - txn = rbtxn_get_toptxn(txn); + if (!for_inval || (for_inval && !rbtxn_inval_all_cache(txn))) + { + uint32 *ninvalidations; + SharedInvalidationMessage **invalidations; - Assert(nmsgs > 0); + ninvalidations = for_inval ? &txn->ninvalidations_distr : &txn->ninvalidations; + invalidations = for_inval ? &txn->distributed_invalidations : &txn->invalidations; - /* Accumulate invalidations. */ - if (txn->ninvalidations == 0) - { - txn->ninvalidations = nmsgs; - txn->invalidations = (SharedInvalidationMessage *) - palloc(sizeof(SharedInvalidationMessage) * nmsgs); - memcpy(txn->invalidations, msgs, - sizeof(SharedInvalidationMessage) * nmsgs); - } - else - { - txn->invalidations = (SharedInvalidationMessage *) - repalloc(txn->invalidations, sizeof(SharedInvalidationMessage) * - (txn->ninvalidations + nmsgs)); + /* Accumulate invalidations. */ + if (*ninvalidations == 0) + { + *ninvalidations = nmsgs; + *invalidations = (SharedInvalidationMessage *) + palloc(sizeof(SharedInvalidationMessage) * nmsgs); + memcpy(*invalidations, msgs, sizeof(SharedInvalidationMessage) * nmsgs); + } + else + { + *invalidations = (SharedInvalidationMessage *) + repalloc(*invalidations, sizeof(SharedInvalidationMessage) * + (*ninvalidations + nmsgs)); - memcpy(txn->invalidations + txn->ninvalidations, msgs, - nmsgs * sizeof(SharedInvalidationMessage)); - txn->ninvalidations += nmsgs; + memcpy(*invalidations + *ninvalidations, msgs, + nmsgs * sizeof(SharedInvalidationMessage)); + *ninvalidations += nmsgs; + } } change = ReorderBufferGetChange(rb); @@ -3325,6 +3352,80 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, MemoryContextSwitchTo(oldcontext); } +/* + * Accumulate the invalidations for executing them later. + * + * This needs to be called for each XLOG_XACT_INVALIDATIONS message. + */ +void +ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs) +{ + ReorderBufferTXN *txn; + + txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); + + /* + * Collect all the invalidations under the top transaction, if available, + * so that we can execute them all together. See comments atop + * ReorderBufferAddInvalidationsCommon function. + */ + txn = rbtxn_get_toptxn(txn); + + Assert(nmsgs > 0); + + ReorderBufferAddInvalidationsCommon(rb, xid, lsn, nmsgs, msgs, txn, false); +} + +/* + * Accumulate the invalidations sent by committed transactions for executing + * them later. + * + * This needs to be called by committed transactions to distribute the + * invalidations to the in-progress transactions. + */ +void +ReorderBufferAddDistributedInvalidtions(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs) +{ + ReorderBufferTXN *txn; + Size req_mem_size; + + txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); + + /* + * Collect all the invalidations under the top transaction, if available, + * so that we can execute them all together. See comments + * ReorderBufferAddInvalidationsCommon ReorderBufferAddInvalidations. + */ + txn = rbtxn_get_toptxn(txn); + + Assert(nmsgs > 0); + + req_mem_size = sizeof(SharedInvalidationMessage) * (txn->ninvalidations_distr + nmsgs); + + /* + * If the number of invalidation messages is larger than 8MB, it's more + * efficient to invalidate the entire cache rather than processing each + * message individually. + */ + if (req_mem_size > (8 * 1024 * 1024) || rbtxn_inval_all_cache(txn)) + { + txn->txn_flags |= RBTXN_INVAL_ALL_CACHE; + + if (txn->distributed_invalidations) + { + pfree(txn->distributed_invalidations); + txn->distributed_invalidations = NULL; + txn->ninvalidations_distr = 0; + } + } + + ReorderBufferAddInvalidationsCommon(rb, xid, lsn, nmsgs, msgs, txn, true); +} + /* * Apply all invalidations we know. Possibly we only need parts at this point * in the changestream but we don't know which those are. diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index d48ebb8337b..5f218ffaa84 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -931,8 +931,9 @@ SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, Transact { Assert(msgs != NULL); - ReorderBufferAddInvalidations(builder->reorder, txn->xid, lsn, - ninvalidations, msgs); + ReorderBufferAddDistributedInvalidtions(builder->reorder, + txn->xid, lsn, + ninvalidations, msgs); } } } diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 8a32bbe28df..4049ad4bf4e 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -186,6 +186,7 @@ typedef struct ReorderBufferChange #define RBTXN_PREPARE 0x0040 #define RBTXN_SKIPPED_PREPARE 0x0080 #define RBTXN_HAS_STREAMABLE_CHANGE 0x0100 +#define RBTXN_INVAL_ALL_CACHE 0x0200 /* Does the transaction have catalog changes? */ #define rbtxn_has_catalog_changes(txn) \ @@ -223,6 +224,12 @@ typedef struct ReorderBufferChange ((txn)->txn_flags & RBTXN_HAS_STREAMABLE_CHANGE) != 0 \ ) +/* Should the complete cache be invalidated? */ +#define rbtxn_inval_all_cache(txn) \ +( \ + ((txn)->txn_flags & RBTXN_INVAL_ALL_CACHE) != 0 \ +) + /* * Has this transaction been streamed to downstream? * @@ -435,6 +442,13 @@ typedef struct ReorderBufferTXN * Private data pointer of the output plugin. */ void *output_plugin_private; + + /* + * Stores the cache invalidation messages distributed by the committing + * transaction. + */ + uint32 ninvalidations_distr; + SharedInvalidationMessage *distributed_invalidations; } ReorderBufferTXN; /* so we can define the callbacks used inside struct ReorderBuffer itself */ @@ -728,6 +742,9 @@ extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, CommandId cmin, CommandId cmax, CommandId combocid); extern void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs); +extern void ReorderBufferAddDistributedInvalidtions(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs); extern void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations); extern void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); -- 2.43.0