From 4517d67fc14b95b2761381781949c2931cead70e Mon Sep 17 00:00:00 2001 From: Vignesh Date: Fri, 30 May 2025 12:06:07 +0530 Subject: [PATCH v9-master] Fix exponential memory allocation issue in logical decoding due to invalidation message distribution Commit 4909b38af0 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 | 174 +++++++++++++++--- src/backend/replication/logical/snapbuild.c | 5 +- src/include/replication/reorderbuffer.h | 18 ++ 3 files changed, 166 insertions(+), 31 deletions(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 67655111875..860e1d4b0bb 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -109,10 +109,18 @@ #include "storage/procarray.h" #include "storage/sinval.h" #include "utils/builtins.h" +#include "utils/inval.h" #include "utils/memutils.h" #include "utils/rel.h" #include "utils/relfilenumbermap.h" +/* + * Maximum number of distributed invalidation messages per transaction. + * Each message is ~16 bytes, this allows up to 8 MB of invalidation + * message data. + */ +#define MAX_DISTR_INVAL_MSG_PER_TXN 524288 + /* entry for a hash table we use to map from xid to our transaction state */ typedef struct ReorderBufferTXNByIdEnt { @@ -472,6 +480,12 @@ ReorderBufferFreeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) txn->invalidations = NULL; } + if (txn->invalidations_distributed) + { + pfree(txn->invalidations_distributed); + txn->invalidations_distributed = NULL; + } + /* Reset the toast hash */ ReorderBufferToastReset(rb, txn); @@ -2661,7 +2675,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_distributed, + txn->invalidations_distributed); + } if (using_subtxn) RollbackAndReleaseCurrentSubTransaction(); @@ -2710,8 +2731,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_distributed, + txn->invalidations_distributed); + } if (using_subtxn) RollbackAndReleaseCurrentSubTransaction(); @@ -3421,6 +3448,55 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, txn->ntuplecids++; } +/* + * Add a new invalidation message to the reorder buffer queue. + */ +static void +ReorderBufferQueueInvalidations(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs) +{ + ReorderBufferChange *change; + + change = ReorderBufferAllocChange(rb); + change->action = REORDER_BUFFER_CHANGE_INVALIDATION; + change->data.inval.ninvalidations = nmsgs; + change->data.inval.invalidations = (SharedInvalidationMessage *) + palloc(sizeof(SharedInvalidationMessage) * nmsgs); + memcpy(change->data.inval.invalidations, msgs, + sizeof(SharedInvalidationMessage) * nmsgs); + + ReorderBufferQueueChange(rb, xid, lsn, change, false); +} + +/* + * Accumulate the invalidations. + */ +static void +ReorderBufferAccumulateInvalidations(uint32 *ninvalidations, + SharedInvalidationMessage **invalidations, + Size nmsgs, + SharedInvalidationMessage *msgs) +{ + 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(*invalidations + *ninvalidations, msgs, + nmsgs * sizeof(SharedInvalidationMessage)); + *ninvalidations += nmsgs; + } +} + /* * Accumulate the invalidations for executing them later. * @@ -3441,7 +3517,6 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, { ReorderBufferTXN *txn; MemoryContext oldcontext; - ReorderBufferChange *change; txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); @@ -3456,35 +3531,76 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, Assert(nmsgs > 0); - /* Accumulate invalidations. */ - if (txn->ninvalidations == 0) - { - txn->ninvalidations = nmsgs; - txn->invalidations = (SharedInvalidationMessage *) - palloc(sizeof(SharedInvalidationMessage) * nmsgs); - memcpy(txn->invalidations, msgs, - sizeof(SharedInvalidationMessage) * nmsgs); - } - else + /* + * If the complete cache will be invalidated, we don't need to accumulate + * the invalidations. + */ + if (!rbtxn_inval_all_cache(txn)) + ReorderBufferAccumulateInvalidations(&txn->ninvalidations, + &txn->invalidations, nmsgs, msgs); + + ReorderBufferQueueInvalidations(rb, xid, lsn, nmsgs, msgs); + + MemoryContextSwitchTo(oldcontext); +} + +/* + * 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 +ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size nmsgs, + SharedInvalidationMessage *msgs) +{ + ReorderBufferTXN *txn; + MemoryContext oldcontext; + + 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 + * ReorderBufferAddInvalidations. + */ + txn = rbtxn_get_toptxn(txn); + + Assert(nmsgs > 0); + + /* + * If the number of invalidation messages is high, performing a full cache + * invalidation is more efficient than handling each message separately. + */ + if (((nmsgs + txn->ninvalidations_distributed) > MAX_DISTR_INVAL_MSG_PER_TXN) || + rbtxn_inval_all_cache(txn)) { - txn->invalidations = (SharedInvalidationMessage *) - repalloc(txn->invalidations, sizeof(SharedInvalidationMessage) * - (txn->ninvalidations + nmsgs)); + txn->txn_flags |= RBTXN_INVAL_ALL_CACHE; - memcpy(txn->invalidations + txn->ninvalidations, msgs, - nmsgs * sizeof(SharedInvalidationMessage)); - txn->ninvalidations += nmsgs; - } + if (txn->invalidations_distributed) + { + pfree(txn->invalidations_distributed); + txn->invalidations_distributed = NULL; + txn->ninvalidations_distributed = 0; + } - change = ReorderBufferAllocChange(rb); - change->action = REORDER_BUFFER_CHANGE_INVALIDATION; - change->data.inval.ninvalidations = nmsgs; - change->data.inval.invalidations = (SharedInvalidationMessage *) - palloc(sizeof(SharedInvalidationMessage) * nmsgs); - memcpy(change->data.inval.invalidations, msgs, - sizeof(SharedInvalidationMessage) * nmsgs); + if (txn->invalidations) + { + pfree(txn->invalidations); + txn->invalidations = NULL; + txn->ninvalidations = 0; + } + } + else + ReorderBufferAccumulateInvalidations(&txn->ninvalidations_distributed, + &txn->invalidations_distributed, + nmsgs, msgs); - ReorderBufferQueueChange(rb, xid, lsn, change, false); + ReorderBufferQueueInvalidations(rb, xid, lsn, nmsgs, msgs); MemoryContextSwitchTo(oldcontext); } diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index 0d7bddbe4ed..3e052ea1d58 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -807,8 +807,9 @@ SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, Transact { Assert(msgs != NULL); - ReorderBufferAddInvalidations(builder->reorder, txn->xid, lsn, - ninvalidations, msgs); + ReorderBufferAddDistributedInvalidations(builder->reorder, + txn->xid, lsn, + ninvalidations, msgs); } } } diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 24e88c409ba..15d839f3d90 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -176,6 +176,7 @@ typedef struct ReorderBufferChange #define RBTXN_SENT_PREPARE 0x0200 #define RBTXN_IS_COMMITTED 0x0400 #define RBTXN_IS_ABORTED 0x0800 +#define RBTXN_INVAL_ALL_CACHE 0x1000 #define RBTXN_PREPARE_STATUS_MASK (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE | RBTXN_SENT_PREPARE) @@ -265,6 +266,12 @@ typedef struct ReorderBufferChange ((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \ ) +/* Should the complete cache be invalidated? */ +#define rbtxn_inval_all_cache(txn) \ +( \ + ((txn)->txn_flags & RBTXN_INVAL_ALL_CACHE) != 0 \ +) + /* Is this a top-level transaction? */ #define rbtxn_is_toptxn(txn) \ ( \ @@ -422,6 +429,14 @@ typedef struct ReorderBufferTXN uint32 ninvalidations; SharedInvalidationMessage *invalidations; + /* + * Stores the cache invalidation messages distributed by the committing + * transaction. + */ + uint32 ninvalidations_distributed; + SharedInvalidationMessage *invalidations_distributed; + + /* --- * Position in one of two lists: * * list of subtransactions if we are *known* to be subxact @@ -738,6 +753,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 ReorderBufferAddDistributedInvalidations(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