From 35a6c5263fb0c88895990f8b23401443d208f337 Mon Sep 17 00:00:00 2001 From: Sebastian Nagel Date: Tue, 21 Jul 2026 23:58:21 +0200 Subject: [PATCH 1/5] Add mempool-state-bench: concurrent shared-state access benchmark Models the proto-devnet baseline mempool contention against the real mempool (openMempoolWithoutSyncThread) over a latency-injected mocked ledger interface. Concurrent roles mirror a node under tx-submission load: one server (reader) + one client (adder) per peer, local firehose clients, and a syncer that revalidates the whole mempool while holding the istate lock. Validated against the frozen baseline (V1, 100ms/50Mbit, V2LSM): - low load (100 TPS): ~90 tx/s, keeps up (devnet ~93) - backpressure ceiling ~550 tx/s (offered 2000 == unbounded) - high load (~40k txs): ~8s sync stall, matching the devnet's ~8s intake gaps; sync cost linear at ~200us/key - server-read stall == sync time at every load point (the contention the double-buffer design targets) Adds mkInitialLedgerState + advanceTip helpers to the bench TestBlock. --- ouroboros-consensus.cabal | 30 ++ .../Bench/Consensus/Mempool/TestBlock.hs | 21 +- .../bench/mempool-state-bench/Main.hs | 348 ++++++++++++++++++ 3 files changed, 397 insertions(+), 2 deletions(-) create mode 100644 ouroboros-consensus/bench/mempool-state-bench/Main.hs diff --git a/ouroboros-consensus.cabal b/ouroboros-consensus.cabal index cd0e788665..3016bbfa42 100644 --- a/ouroboros-consensus.cabal +++ b/ouroboros-consensus.cabal @@ -949,6 +949,36 @@ benchmark mempool-bench unstable-mempool-test-utils, with-utf8, +benchmark mempool-state-bench + import: common-bench + type: exitcode-stdio-1.0 + hs-source-dirs: + ouroboros-consensus/bench/mempool-state-bench + ouroboros-consensus/bench/mempool-bench + + main-is: Main.hs + other-modules: + Bench.Consensus.Mempool.TestBlock + + ghc-options: -threaded -rtsopts -with-rtsopts=-N4 + build-depends: + async, + base, + cardano-ledger-core, + cardano-slotting, + containers, + contra-tracer, + deepseq, + io-classes:si-timers, + mempack, + nothunks, + ouroboros-consensus, + serialise, + time, + transformers, + tree-diff, + unstable-consensus-testlib, + benchmark ChainSync-client-bench import: common-bench type: exitcode-stdio-1.0 diff --git a/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs b/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs index af1bc802f2..ade6029f54 100644 --- a/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs +++ b/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs @@ -18,6 +18,8 @@ module Bench.Consensus.Mempool.TestBlock -- * Initial parameters , initialLedgerState + , mkInitialLedgerState + , advanceTip , sampleLedgerConfig -- * Transactions @@ -37,6 +39,7 @@ import Data.MemPack import Data.Set (Set) import qualified Data.Set as Set import Data.TreeDiff (ToExpr) +import Data.Word (Word64) import GHC.Generics (Generic) import NoThunks.Class (NoThunks) import qualified Ouroboros.Consensus.Block as Block @@ -86,15 +89,29 @@ mkTx cons prod = -------------------------------------------------------------------------------} initialLedgerState :: LedgerState (TestBlockWith Tx) ValuesMK -initialLedgerState = +initialLedgerState = mkInitialLedgerState [] + +-- | Like 'initialLedgerState' but seeded with a set of available tokens (the +-- UTxO). Chains of transactions can then be built by consuming a seed token and +-- producing the next one. +mkInitialLedgerState :: [Token] -> LedgerState (TestBlockWith Tx) ValuesMK +mkInitialLedgerState toks = TestLedger { lastAppliedPoint = Block.GenesisPoint , payloadDependentState = TestPLDS - { getTestPLDS = ValuesMK Map.empty + { getTestPLDS = ValuesMK (Map.fromList [(t, ()) | t <- toks]) } } +-- | Move the tip to a fresh point (distinct per @n@) while keeping the ledger +-- tables unchanged. Used to force the mempool to resync/revalidate against a +-- "new" tip without invalidating any of its transactions. +advanceTip :: + Word64 -> LedgerState (TestBlockWith Tx) ValuesMK -> LedgerState (TestBlockWith Tx) ValuesMK +advanceTip n st = + st{lastAppliedPoint = Block.blockPoint (firstBlockWithPayload n (Tx Set.empty Set.empty))} + sampleLedgerConfig :: Ledger.LedgerConfig TestBlock sampleLedgerConfig = testBlockLedgerConfigFrom $ diff --git a/ouroboros-consensus/bench/mempool-state-bench/Main.hs b/ouroboros-consensus/bench/mempool-state-bench/Main.hs new file mode 100644 index 0000000000..53bac588a1 --- /dev/null +++ b/ouroboros-consensus/bench/mempool-state-bench/Main.hs @@ -0,0 +1,348 @@ +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | Concurrent benchmark of the __real__ mempool's shared-state access patterns +-- under a Leios-scale transaction load. +-- +-- It opens the actual mempool ('openMempoolWithoutSyncThread') over a mocked +-- ledger interface whose forker reads inject a configurable latency to model +-- on-disk UTxO reads. Three roles run concurrently against it, as in a node +-- under tx-submission load: +-- +-- * __Adders__ (tx-submission clients and local clients): each submits an +-- independent chain of transactions via the real 'addTx', rate-limited to a +-- target rate. +-- +-- * __Readers__ (tx-submission servers / block forging): call the real +-- 'getSnapshot' (@readTMVar istate@) on a configurable per-peer cadence, +-- measuring how long a read blocks. +-- +-- * __Syncer__ (the mempool sync thread): periodically advances the ledger tip +-- and runs the real 'testSyncWithLedger', which revalidates the mempool +-- through the latency-injected forker. +-- +-- With the mempool holding many transactions, this reproduces the contention +-- between revalidation, ingestion and serving that the mempool sync targets. +module Main (main) where + +import Bench.Consensus.Mempool.TestBlock + ( TestBlock + , Token (Token) + , advanceTip + , mkInitialLedgerState + , mkTx + , sampleLedgerConfig + ) +import qualified Control.Concurrent as Conc +import Control.Concurrent.Async (async, wait) +import Control.Exception (evaluate) +import Control.Monad (forM, when) +import Control.Monad.Class.MonadTime.SI (diffTime, getMonotonicTime) +import Control.Tracer (nullTracer) +import Data.IORef +import qualified Data.Set as Set +import Data.Time.Clock (DiffTime) +import Ouroboros.Consensus.Ledger.Basics (LedgerState) +import Ouroboros.Consensus.Ledger.SupportsMempool (ByteSize32 (ByteSize32)) +import Ouroboros.Consensus.Ledger.Tables + ( KeysMK (KeysMK) + , LedgerTables (LedgerTables) + , ValuesMK + , projectLedgerTables + ) +import Ouroboros.Consensus.Ledger.Tables.Utils + ( emptyLedgerTables + , forgetLedgerTables + , restrictValues' + ) +import Ouroboros.Consensus.Mempool + ( Mempool (addTx, getSnapshot, testSyncWithLedger) + , MempoolCapacityBytesOverride (MempoolCapacityBytesOverride) + , openMempoolWithoutSyncThread + , snapshotTxs + ) +import Ouroboros.Consensus.Mempool.API + ( AddTxOnBehalfOf (AddTxForLocalClient, AddTxForRemotePeer) + , isMempoolTxAdded + ) +import Ouroboros.Consensus.Mempool.Impl.Common + ( LedgerInterface (LedgerInterface, getCurrentLedgerState) + , MempoolLedgerDBView (MempoolLedgerDBView) + ) +import Ouroboros.Consensus.Storage.LedgerDB.Forker + ( ReadOnlyForker (..) + , Statistics (Statistics) + ) +import Ouroboros.Consensus.Util.IOLike + ( StrictTVar + , atomically + , newTVarIO + , readTVar + , writeTVar + ) +import System.Environment (lookupEnv) +import System.IO.Unsafe (unsafePerformIO) +import Test.Util.Orphans.IOLike () +import Text.Read (readMaybe) + +-- * Configuration (all overridable via environment variables) + +envInt :: String -> Int -> Int +envInt k d = unsafePerformIO $ maybe d id . (>>= readMaybe) <$> lookupEnv k + +envDouble :: String -> Double -> Double +envDouble k d = unsafePerformIO $ maybe d id . (>>= readMaybe) <$> lookupEnv k + +{-# NOINLINE durationSec #-} +durationSec :: Double +durationSec = envDouble "DURATION" 20 + +-- | Number of N2N peers. Each peer contributes one tx-submission __server__ (a +-- reader serving txs to that peer) and one tx-submission __client__ (a remote +-- adder feeding txs received from that peer). +{-# NOINLINE numPeers #-} +numPeers :: Int +numPeers = envInt "PEERS" 2 + +-- | Number of local (N2C) clients. They add on behalf of a local client (higher +-- fifo priority). +{-# NOINLINE numLocalClients #-} +numLocalClients :: Int +numLocalClients = envInt "LOCAL_CLIENTS" 1 + +-- | tx-submission servers = one per peer. +numReaders :: Int +numReaders = numPeers + +-- | tx-submission clients = one per peer (N2N) + the local clients (N2C). +numAdders :: Int +numAdders = numPeers + numLocalClients + +-- | Total target submission rate across all adders (tx/s). @TPS=0@ means +-- unbounded (adders submit as fast as they can), to measure the mempool's max +-- sustained rate. +{-# NOINLINE targetTpsTotal #-} +targetTpsTotal :: Double +targetTpsTotal = envDouble "TPS" 100 + +-- | How often the syncer advances the tip + revalidates. A chain adopts a block +-- roughly every ~20 s; a shorter period here exercises the sync contention more +-- often. This is strictly periodic, unlike real block adoption. +{-# NOINLINE syncPeriodSec #-} +syncPeriodSec :: Double +syncPeriodSec = envDouble "SYNC_PERIOD" 5 + +-- | Fixed cost of a forker table read (models one on-disk round-trip), microseconds. +{-# NOINLINE readBaseMicros #-} +readBaseMicros :: Int +readBaseMicros = envInt "READ_BASE_US" 500 + +-- | Additional per-key cost of a forker table read (models per-UTxO on-disk +-- lookup), microseconds. This is what makes a full-mempool sync read scale with +-- occupancy. +{-# NOINLINE readPerKeyMicros #-} +readPerKeyMicros :: Int +readPerKeyMicros = envInt "READ_PERKEY_US" 200 + +-- | Pause between successive 'getSnapshot's per reader, microseconds. A reader +-- models the tx-submission /server/ for one downstream peer, which reads the +-- mempool on request rather than in a spin. On a Leios testnet each downstream +-- peer pulled ~3–4 tx-body requests/s from a relay, and with the txid requests +-- on top a server reads roughly 5–8×/s per peer — a read every ~125–200ms. The +-- default models ~7 reads/s/peer; set @0@ for a tight loop (only sensible for a +-- handful of readers, else hundreds of spinning O(occupancy) readers just +-- measure CPU saturation). +{-# NOINLINE readPeriodMicros #-} +readPeriodMicros :: Int +readPeriodMicros = envInt "READ_PERIOD_US" 150000 + +-- | Disjoint token namespace per adder so their chains never collide. +chainStride :: Int +chainStride = 1_000_000_000 + +capacityOverride :: MempoolCapacityBytesOverride +capacityOverride = MempoolCapacityBytesOverride (ByteSize32 100_000_000) + +-- * Main + +main :: IO () +main = do + putStr $ + unlines + [ "Mempool shared-state concurrent benchmark (real mempool)" + , " duration : " <> show durationSec <> " s" + , " peers : " <> show numPeers <> " (=> " <> show numReaders <> " servers/readers)" + , " clients : " + <> show numAdders + <> " (" + <> show numPeers + <> " N2N + " + <> show numLocalClients + <> " local, target " + <> show targetTpsTotal + <> " tx/s total)" + , " sync period : " <> show syncPeriodSec <> " s" + , " forker read : " <> show readBaseMicros <> " us + " <> show readPerKeyMicros <> " us/key" + , "" + ] + let seeds = [Token (j * chainStride) | j <- [0 .. numAdders - 1]] + baseLedger = mkInitialLedgerState seeds + ledgerVar <- newTVarIO baseLedger + mempool <- + openMempoolWithoutSyncThread + (latencyLedgerInterface ledgerVar) + sampleLedgerConfig + capacityOverride + Nothing + nullTracer + + addedRef <- newIORef (0 :: Int) + readRef <- newIORef (0 :: Int) + maxReadLatRef <- newIORef (0 :: DiffTime) + syncDursRef <- newIORef ([] :: [DiffTime]) + + start <- getMonotonicTime + let expired = do + now <- getMonotonicTime + pure (realToFrac (diffTime now start) >= durationSec) + + syncer <- async (runSyncer expired mempool ledgerVar baseLedger syncDursRef) + readers <- + forM [1 .. numReaders] $ \_ -> + async (runReader expired mempool readRef maxReadLatRef) + adders <- + forM [0 .. numAdders - 1] $ \j -> do + let onBehalf = if j < numPeers then AddTxForRemotePeer else AddTxForLocalClient + async (runAdder expired mempool onBehalf j addedRef) + + mapM_ wait adders + mapM_ wait readers + wait syncer + end <- getMonotonicTime + + finalOccupancy <- length . snapshotTxs <$> atomically (getSnapshot mempool) + added <- readIORef addedRef + reads' <- readIORef readRef + maxReadLat <- readIORef maxReadLatRef + syncDurs <- readIORef syncDursRef + let elapsed = realToFrac (diffTime end start) :: Double + putStr $ + unlines + [ "Results:" + , " elapsed : " <> showT (diffTime end start) + , " txs added : " <> show added + , " final occupancy : " <> show finalOccupancy <> " txs in mempool" + , " throughput : " <> show (round (fromIntegral added / elapsed) :: Int) <> " tx/s" + , " snapshot reads : " <> show reads' + , " max read stall : " <> showT maxReadLat + , " syncs : " <> show (length syncDurs) + , " max sync time : " <> showT (if null syncDurs then 0 else maximum syncDurs) + , " avg sync time : " + <> showT (if null syncDurs then 0 else sum syncDurs / fromIntegral (length syncDurs)) + ] + +-- * Roles + +-- | Adder @j@ submits its own chain: consume token @base+i@, produce @base+i+1@. +runAdder :: IO Bool -> Mempool IO TestBlock -> AddTxOnBehalfOf -> Int -> IORef Int -> IO () +runAdder expired mempool onBehalf j addedRef = go 0 + where + base = j * chainStride + intervalMicros = + if targetTpsTotal <= 0 + then 0 + else round (1_000_000 * fromIntegral numAdders / targetTpsTotal) + go :: Int -> IO () + go i = do + done <- expired + if done + then pure () + else do + let tx = mkTx [Token (base + i)] [Token (base + i + 1)] + r <- addTx mempool onBehalf tx + when (isMempoolTxAdded r) $ atomicModifyIORef' addedRef (\c -> (c + 1, ())) + when (intervalMicros > 0) $ Conc.threadDelay intervalMicros + go (i + 1) + +-- | Reader: real 'getSnapshot' on the configured per-reader cadence +-- ('readPeriodMicros'), recording max read latency. +runReader :: IO Bool -> Mempool IO TestBlock -> IORef Int -> IORef DiffTime -> IO () +runReader expired mempool readRef maxLatRef = go + where + go = do + done <- expired + if done + then pure () + else do + t0 <- getMonotonicTime + snap <- atomically (getSnapshot mempool) + _ <- evaluate (length (snapshotTxs snap)) + t1 <- getMonotonicTime + let lat = diffTime t1 t0 + atomicModifyIORef' readRef (\c -> (c + 1, ())) + atomicModifyIORef' maxLatRef (\m -> (max m lat, ())) + when (readPeriodMicros > 0) $ Conc.threadDelay readPeriodMicros + go + +-- | Syncer: every 'syncPeriodSec', advance the tip and run the real sync. +runSyncer :: + IO Bool -> + Mempool IO TestBlock -> + StrictTVar IO (LedgerState TestBlock ValuesMK) -> + LedgerState TestBlock ValuesMK -> + IORef [DiffTime] -> + IO () +runSyncer expired mempool ledgerVar baseLedger syncDursRef = go 1 + where + go :: Int -> IO () + go n = do + Conc.threadDelay (round (1_000_000 * syncPeriodSec)) + done <- expired + if done + then pure () + else do + atomically $ writeTVar ledgerVar (advanceTip (fromIntegral n) baseLedger) + t0 <- getMonotonicTime + _ <- testSyncWithLedger mempool + t1 <- getMonotonicTime + atomicModifyIORef' syncDursRef (\ds -> (diffTime t1 t0 : ds, ())) + go (n + 1) + +-- * Latency-injecting ledger interface + +latencyLedgerInterface :: + StrictTVar IO (LedgerState TestBlock ValuesMK) -> + LedgerInterface IO TestBlock +latencyLedgerInterface ledgerVar = + LedgerInterface + { getCurrentLedgerState = do + st <- readTVar ledgerVar + pure $ + MempoolLedgerDBView + (forgetLedgerTables st) + ( pure $ + Right $ + ReadOnlyForker + { roforkerClose = pure () + , roforkerGetLedgerState = pure (forgetLedgerTables st) + , roforkerReadTables = \keys -> do + Conc.threadDelay (readBaseMicros + readPerKeyMicros * keysCount keys) + pure (projectLedgerTables st `restrictValues'` keys) + , roforkerReadStatistics = pure (Statistics 0) + , roforkerRangeReadTables = \_ -> pure (emptyLedgerTables, Nothing) + } + ) + } + +keysCount :: LedgerTables (LedgerState TestBlock) KeysMK -> Int +keysCount (LedgerTables (KeysMK s)) = Set.size s + +-- * Formatting + +showT :: DiffTime -> String +showT t + | s < 1e-3 = show (round (s * 1_000_000) :: Int) <> " us" + | s < 1 = show (round (s * 1_000) :: Int) <> " ms" + | otherwise = show (fromIntegral (round (s * 1000) :: Int) / 1000 :: Double) <> " s" + where + s = realToFrac t :: Double From 45be3105af6ee05e9448cba87e783c11f373cf5d Mon Sep 17 00:00:00 2001 From: Sebastian Nagel Date: Fri, 24 Jul 2026 06:51:32 +0200 Subject: [PATCH 2/5] Mempool-state-bench: model reapply << apply in the TestBlock The TestBlock's reapplyTx routed through applyTx, so full validation and reapplication cost the same and were both essentially free. That let the mempool-state-bench degenerate: with apply ~= reapply the sync-vs-ingest convergence assumption is vacuous. Give applyTx a configurable simulated CPU cost (MEMPOOL_APPLY_CPU_US, default 0 so ordinary tests and the criterion mempool-bench are unaffected) via a busy-wait, and run reapplyTx through the ledger transition directly with a much smaller MEMPOOL_REAPPLY_CPU_US so it no longer pays the validation cost. Setting e.g. 200us/20us reproduces the real-node relationship where a mempool sync (reapply-only) is markedly cheaper per tx than ingestion (full validation). --- .../Bench/Consensus/Mempool/TestBlock.hs | 63 ++++++++++++++++--- 1 file changed, 56 insertions(+), 7 deletions(-) diff --git a/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs b/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs index ade6029f54..70264ea2fe 100644 --- a/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs +++ b/ouroboros-consensus/bench/mempool-bench/Bench/Consensus/Mempool/TestBlock.hs @@ -40,6 +40,7 @@ import Data.Set (Set) import qualified Data.Set as Set import Data.TreeDiff (ToExpr) import Data.Word (Word64) +import GHC.Clock (getMonotonicTimeNSec) import GHC.Generics (Generic) import NoThunks.Class (NoThunks) import qualified Ouroboros.Consensus.Block as Block @@ -51,7 +52,10 @@ import Ouroboros.Consensus.Ledger.Tables import qualified Ouroboros.Consensus.Ledger.Tables.Diff as Diff import qualified Ouroboros.Consensus.Ledger.Tables.Utils as Ledger import Ouroboros.Consensus.Util.IndexedMemPack (IndexedMemPack (..)) +import System.Environment (lookupEnv) +import System.IO.Unsafe (unsafePerformIO) import Test.Util.TestBlock hiding (TestBlock) +import Text.Read (readMaybe) {------------------------------------------------------------------------------- MempoolTestBlock @@ -240,17 +244,62 @@ txSize (TestBlockGenTx tx) = fromIntegral $ 1 + length (consumed tx) + length (produced tx) +-- | Simulated CPU cost, in microseconds, of /fully/ validating a transaction +-- (the script and signature checks a real ledger performs in 'applyTx'). Read +-- once from @MEMPOOL_APPLY_CPU_US@; defaults to @0@ so ordinary tests and the +-- criterion @mempool-bench@ are unaffected. +-- +-- Together with 'reapplyCpuMicros' this lets a benchmark reproduce the +-- real-node relationship @reapply ≪ apply@: reapplication skips the expensive +-- checks, so a mempool sync (which only reapplies) is strictly cheaper per tx +-- than ingestion (which fully validates). The mempool-state-bench relies on +-- this for its sync-vs-ingest convergence to be faithful. +{-# NOINLINE applyCpuMicros #-} +applyCpuMicros :: Int +applyCpuMicros = envInt "MEMPOOL_APPLY_CPU_US" 0 + +-- | Simulated CPU cost, in microseconds, of /reapplying/ an already-validated +-- transaction. Read once from @MEMPOOL_REAPPLY_CPU_US@; defaults to @0@. Keep +-- this well below 'applyCpuMicros' to model @reapply ≪ apply@. +{-# NOINLINE reapplyCpuMicros #-} +reapplyCpuMicros :: Int +reapplyCpuMicros = envInt "MEMPOOL_REAPPLY_CPU_US" 0 + +{-# NOINLINE envInt #-} +envInt :: String -> Int -> Int +envInt k d = unsafePerformIO $ maybe d id . (>>= readMaybe) <$> lookupEnv k + +-- | Busy-wait (burning CPU, /not/ sleeping — validation contends for cores) +-- for @us@ microseconds, then return @tx@. The result is @tx@ itself and the +-- caller feeds it into the ledger transition, so the spin depends on the tx +-- and cannot be shared across calls or optimised away. +{-# NOINLINE burnCpuMicros #-} +burnCpuMicros :: Int -> Tx -> Tx +burnCpuMicros us tx + | us <= 0 = tx + | otherwise = unsafePerformIO $ do + let targetNs = fromIntegral us * 1000 :: Word64 + start <- getMonotonicTimeNSec + let go = do + now <- getMonotonicTimeNSec + if now - start >= targetNs then pure tx else go + go + instance Ledger.LedgerSupportsMempool TestBlock where applyTx _cfg _shouldIntervene _slot (TestBlockGenTx tx) tickedSt = except $ fmap ((,ValidatedGenTx (TestBlockGenTx tx)) . Ledger.trackingToDiffs) $ - applyDirectlyToPayloadDependentState tickedSt tx - - reapplyTx cfg slot (ValidatedGenTx genTx) tickedSt = - Ledger.applyDiffs tickedSt . fst - <$> Ledger.applyTx cfg Ledger.DoNotIntervene slot genTx tickedSt - - -- FIXME: it is ok to use 'DoNotIntervene' here? + -- Pay the (simulated) full-validation cost. 'burnCpuMicros' returns the + -- tx we then apply, so it is forced as part of producing the result. + applyDirectlyToPayloadDependentState tickedSt (burnCpuMicros applyCpuMicros tx) + + -- Reapplication does /not/ route through 'applyTx' (which would pay the full + -- validation cost); it runs the ledger transition directly, paying only the + -- much cheaper 'reapplyCpuMicros'. + reapplyTx _cfg _slot (ValidatedGenTx (TestBlockGenTx tx)) tickedSt = + except $ + Ledger.applyDiffs tickedSt . Ledger.trackingToDiffs + <$> applyDirectlyToPayloadDependentState tickedSt (burnCpuMicros reapplyCpuMicros tx) txForgetValidated (ValidatedGenTx tx) = tx From 009be940cd64764d6bcf7f2c075b123b93dafb9c Mon Sep 17 00:00:00 2001 From: Sebastian Nagel Date: Wed, 22 Jul 2026 13:39:39 +0200 Subject: [PATCH 3/5] Mempool QSM: scale down the parallel atomic linearizability test Drop the explicit withMaxSuccess (was 10000) on the atomic parallel test so it uses the default test count (100), overridable from the CLI via --quickcheck-tests, and reduce the reruns from 100 to 10. At the original size the QSM parallel-history linearizability search OOMs (which is why it was disabled); at this size it does not. Rationale for using the default count rather than a fixed one: this test's runtime is highly variable (seconds to minutes), because some generated parallel programs contain long transaction chains whose linearization search is expensive. Leaving the count at the CLI-overridable default lets it be tuned per environment without editing the source. --- .../consensus-test/Test/Consensus/Mempool/StateMachine.hs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/ouroboros-consensus/test/consensus-test/Test/Consensus/Mempool/StateMachine.hs b/ouroboros-consensus/test/consensus-test/Test/Consensus/Mempool/StateMachine.hs index 1b1ed31bc7..4008b901b0 100644 --- a/ouroboros-consensus/test/consensus-test/Test/Consensus/Mempool/StateMachine.hs +++ b/ouroboros-consensus/test/consensus-test/Test/Consensus/Mempool/StateMachine.hs @@ -777,7 +777,7 @@ prop_mempoolParallel :: MakeAtomic -> (Int -> LedgerState blk ValuesMK -> Gen [GenTx blk]) -> Property -prop_mempoolParallel cfg capacity initialState ma gTxs = forAllParallelCommandsNTimes sm0 Nothing 100 $ +prop_mempoolParallel cfg capacity initialState ma gTxs = forAllParallelCommandsNTimes sm0 Nothing 10 $ \cmds -> monadicIO $ do (sut, trcr) <- run $ mkSUT cfg initialState ior <- run $ newTVarIO sut @@ -805,9 +805,8 @@ tests = -- More commands require exponentially more memory to explore. localOption (QuickCheckMaxSize 40) $ testProperty "atomic" $ - withMaxSuccess 10000 $ - prop_mempoolParallel testLedgerConfigNoSizeLimits txMaxBytes' testInitLedger Atomic $ - \i -> fmap (fmap fst . fst) . genTxs i + prop_mempoolParallel testLedgerConfigNoSizeLimits txMaxBytes' testInitLedger Atomic $ + \i -> fmap (fmap fst . fst) . genTxs i , testProperty "non atomic" $ withMaxSuccess 10 $ prop_mempoolParallel testLedgerConfigNoSizeLimits txMaxBytes' testInitLedger NonAtomic $ From 532df7af072b010d8234da1f5615e367f2209e2d Mon Sep 17 00:00:00 2001 From: Sebastian Nagel Date: Wed, 22 Jul 2026 13:40:07 +0200 Subject: [PATCH 4/5] Mempool: double-buffer sync to decouple readers/adders from revalidation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mempool kept its InternalState in a single StrictTMVar used as both data cell and lock: adds, the full re-sync on tip change, removes, and every snapshot reader all contended on it. On a tip change the sync revalidated the entire mempool (reading all inputs from the LedgerDB) while holding the lock, stalling both intake and serving for the whole revalidation — cost growing with occupancy. Instead, the sync thread does its large LedgerDB read off the lock (against a non-emptying 'readTMVar' snapshot serving as the "off screen buffer") and only takes the lock for the reading and reappling the delta txs, so a reader blocks for that sub-second merge but not for the big read. If that merge grows too costly under higher load, the merge itself can later be moved off the lock behind an optimistic retry; the single-cell structure here is the clean baseline. --- .../Consensus/Mempool/Impl/Common.hs | 14 +- .../Ouroboros/Consensus/Mempool/Init.hs | 4 +- .../Ouroboros/Consensus/Mempool/Update.hs | 145 +++++++++--------- 3 files changed, 90 insertions(+), 73 deletions(-) diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs index 0b2119a205..b6843f64c4 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs @@ -264,6 +264,15 @@ data MempoolEnv m blk = MempoolEnv , mpEnvForker :: StrictMVar m (ReadOnlyForker m (LedgerState blk)) , mpEnvLedgerCfg :: LedgerConfig blk , mpEnvStateVar :: StrictTMVar m (InternalState blk) + -- ^ The single, authoritative internal state of the mempool, which doubles as + -- the /writer/ lock. Writers (adds, removes and the sync merge) 'takeTMVar' + -- it, do their work, and 'putTMVar' the new state; readers ('getSnapshot', + -- 'getCapacity', 'getSnapshotFor') 'readTMVar' it. Because it is a single + -- cell, the whole capacity accounting has one source of truth and cannot + -- diverge. A reader only ever blocks for the duration a writer holds the + -- lock; the sync keeps that short by doing its large LedgerDB read /before/ + -- taking the lock (see 'implSyncWithLedger'), so only the (sub-second) merge + -- is under it. , mpEnvAddTxsRemoteFifo :: StrictMVar m () , mpEnvAddTxsAllFifo :: StrictMVar m () , mpEnvTracer :: Tracer m (TraceEventMempool blk) @@ -293,9 +302,8 @@ initMempoolEnv ledgerInterface cfg capacityOverride mbTimeoutConfig tracer = do Right frk -> do frkMVar <- newMVar frk let (slot, st') = tickLedgerState cfg (ForgeInUnknownSlot st) - isVar <- - newTMVarIO $ - initInternalState capacityOverride TxSeq.zeroTicketNo cfg slot st' + is0 = initInternalState capacityOverride TxSeq.zeroTicketNo cfg slot st' + isVar <- newTMVarIO is0 addTxRemoteFifo <- newMVar () addTxAllFifo <- newMVar () return diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Init.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Init.hs index 6b696af015..ce900b771f 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Init.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Init.hs @@ -113,7 +113,9 @@ mkMempool mpEnv = Mempool { addTx = fmap runIdentity .: implAddTx mpEnv ProductionAddTx , removeTxsEvenIfValid = implRemoveTxsEvenIfValid mpEnv - , getSnapshot = snapshotFromIS <$> readTMVar istate + , -- Readers 'readTMVar' the single state cell; they only block while a + -- writer holds it (see 'MempoolEnv'). + getSnapshot = snapshotFromIS <$> readTMVar istate , getSnapshotFor = implGetSnapshotFor mpEnv , getSnapshotForNoCache = implGetSnapshotForNoCache mpEnv , getCapacity = isCapacity <$> readTMVar istate diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs index 2649fb11f0..3bce0f5843 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs @@ -27,7 +27,7 @@ import qualified Data.Text as T import Ouroboros.Consensus.HeaderValidation import Ouroboros.Consensus.Ledger.Abstract import Ouroboros.Consensus.Ledger.SupportsMempool -import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables) +import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables, unionValues) import Ouroboros.Consensus.Mempool.API import Ouroboros.Consensus.Mempool.Capacity import Ouroboros.Consensus.Mempool.Impl.Common @@ -259,7 +259,7 @@ doAddTx mpEnv caller wti tx = do -- in Conway. let txt = T.pack $ "MempoolTxTooSlow (" <> show dur <> ") " <> show (txId tx) in mkMempoolApplyTxError (isLedgerState is) txt - case mbX of + res <- case mbX of Nothing -> case (wti, mbTimeoutSoftTxErr) of (Intervene, Just txerr) -> do rejectBecauseOfTimeoutSoft txerr @@ -291,6 +291,7 @@ doAddTx mpEnv caller wti tx = do testDiffTime TransactionProcessingResult is' _ _ = outcome pure (Right outcome, fromMaybe is is') + pure res case (caller, eRes) of (ProductionAddTx, _) -> either (doAddTx' . Just) (pure . Identity) eRes (TestingAddTx _, Left _) -> pure Nothing @@ -531,74 +532,80 @@ implSyncWithLedger :: MempoolEnv m blk -> m r implSyncWithLedger projectResult mpEnv = - encloseTimedWith (TraceMempoolSynced >$< mpEnvTracer mpEnv) $ do - res <- - -- There could possibly be a race condition if we used there the state - -- that triggered the re-syncing in the background watcher, if a different - -- action acquired the state before the revalidation started. - -- - -- For that reason, we read the state again here in the same STM - -- transaction in which we acquire the internal state of the mempool. - -- - -- The following interleaving could happen: - -- - -- - [ChainSel thread] We adopt a new block B at the tip of our selection. - -- - -- - [Mempool sync thread] The Watcher wakes up, seeing that the tip has - -- changed to B, records it as the fingerprint, and invokes - -- implSyncWithLedger, but doesn't reach withTMVarAnd here. - -- - -- - [ChainSel thread] Adopt a new block C. - -- - -- - [Mempool thread] Execute withTMVarAnd here, obtaining the ledger - -- state for C and syncing the mempool with C. - -- - -- - [Mempool thread] The Watcher wakes up again, seeing that the tip has - -- changed from B to C, and invokes implSyncWithLedger. This time, - -- nothing needs to be done, resulting in TraceMempoolSyncNotNeeded. - -- - -- Just for performance reasons, we will avoid re-validating the mempool - -- if the state didn't change. - withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $ - \is (MempoolLedgerDBView ls meFrk) -> do - let (slot, ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls - if pointHash (isTip is) == castHash (getTipHash ls) && isSlotNo is == slot - then do - -- The tip didn't change, put the same state. - traceWith trcr $ TraceMempoolSyncNotNeeded (isTip is) - pure (Just (projectResult is), is) - else do - -- The tip changed, we have to revalidate - eFrk <- meFrk - case eFrk of - -- This case should happen only if the tip has moved again, this time - -- to a separate fork, since the background thread saw a change in the - -- tip, which should happen very rarely - Left{} -> do - traceWith trcr TraceMempoolTipMovedBetweenSTMBlocks - pure (Nothing, is) - Right frk -> do - modifyMVar_ - forkerMVar - ( \frkOld -> do - roforkerClose frkOld - pure frk - ) - tbs <- castLedgerTables <$> roforkerReadTables frk (castLedgerTables $ isTxKeys is) - let (is', mTrace) = - pureSyncWithLedger - capacityOverride - cfg - slot - ls' - tbs - is - whenJust mTrace (traceWith trcr) - pure (Just (projectResult is'), is') - case res of - Nothing -> implSyncWithLedger projectResult mpEnv - Just res' -> pure res' + encloseTimedWith (TraceMempoolSynced >$< mpEnvTracer mpEnv) go where + -- The expensive part — reading all of the mempool's tx inputs from the + -- LedgerDB — is done /off the lock/, against a snapshot @is0@ taken with a + -- non-emptying 'readTMVar', while adds keep appending. We then 'takeTMVar' + -- only briefly: read the (small) input values for the txs added in the + -- meantime (the "delta", by 'TicketNo'), union them with the off-lock values, + -- and run 'pureSyncWithLedger' over the current txs. Since the union covers + -- exactly the current txs' input keys, the resulting state is identical to a + -- single under-lock revalidation. Only that (sub-second) merge holds the + -- lock, so readers block for the merge but not for the big LedgerDB read. + go = do + MempoolLedgerDBView ls0 meFrk0 <- atomically $ getCurrentLedgerState ldgrInterface + is0 <- atomically $ readTMVar istate + let (slot0, _ls0') = tickLedgerState cfg $ ForgeInUnknownSlot ls0 + if pointHash (isTip is0) == castHash (getTipHash ls0) && isSlotNo is0 == slot0 + then do + -- Looks like a no-op. Confirm under the lock (re-reading the ledger) + -- and return the /current/ committed state, so a concurrent add is not + -- lost from the returned snapshot. + outcome <- + withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $ + \isNow (MempoolLedgerDBView ls _meFrk) -> do + let (slot, _ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls + if pointHash (isTip isNow) == castHash (getTipHash ls) && isSlotNo isNow == slot + then do + traceWith trcr $ TraceMempoolSyncNotNeeded (isTip isNow) + pure (Just (projectResult isNow), isNow) + else + -- The tip changed after our lock-free peek; retry via 'go', + -- which will now take the sync path. + pure (Nothing, isNow) + maybe go pure outcome + else do + eFrk <- meFrk0 + case eFrk of + -- Tip moved to a separate fork between reading it and getting the + -- forker; very rare. Retry. + Left{} -> do + traceWith trcr TraceMempoolTipMovedBetweenSTMBlocks + go + Right frk -> do + -- OFF-LOCK: big read of the snapshot's input values at the new tip. + values0 <- + castLedgerTables <$> roforkerReadTables frk (castLedgerTables $ isTxKeys is0) + outcome <- + withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $ + \isNow (MempoolLedgerDBView ls _meFrk) -> do + let (slot, ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls + if getTipHash ls /= getTipHash ls0 + then + -- Tip moved again during the off-lock read; retry. + pure (Nothing, isNow) + else do + -- Read only the delta's input keys (small), union with the + -- off-lock values, and revalidate /all/ current txs. + let (_, deltaSeq) = TxSeq.splitAfterTicketNo (isTxs isNow) (isLastTicketNo is0) + deltaKeys = + Foldable.foldMap' + (getTransactionKeySets . txForgetValidated . validatedTx . txTicketTx) + (TxSeq.toList deltaSeq) + valuesDelta <- + castLedgerTables <$> roforkerReadTables frk (castLedgerTables deltaKeys) + let allValues = ltliftA2 unionValues values0 valuesDelta + (isFinal, mTrace) = + pureSyncWithLedger capacityOverride cfg slot ls' allValues isNow + modifyMVar_ forkerMVar (\frkOld -> roforkerClose frkOld >> pure frk) + whenJust mTrace (traceWith trcr) + pure (Just (projectResult isFinal), isFinal) + case outcome of + -- Retry: the forker we opened was never installed, so close it. + Nothing -> roforkerClose frk >> go + Just r -> pure r + MempoolEnv { mpEnvStateVar = istate , mpEnvForker = forkerMVar From e0ea49bcea0aa3d4bf54c088cd14097043159d55 Mon Sep 17 00:00:00 2001 From: Sebastian Nagel Date: Fri, 24 Jul 2026 07:21:40 +0200 Subject: [PATCH 5/5] Mempool: bound the sync lock hold with an off-lock reapply loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The single-TMVar sync revalidated all current txs under the lock, so the lock hold — and hence how long snapshot readers (forging, tx serving) can block — grew with mempool occupancy (seconds at tens of thousands of txs). Do the revalidation off the lock and take the lock only for a small, bounded residual. 'implSyncWithLedger' now revalidates the snapshot's txs off the lock, then loops reading the txs added since (the delta, by TicketNo) and reapplying just those on top via the new 'extendReapply'. The delta shrinks each round (adds are serialised and pay full validation, reapplication is cheaper), so once it is at most 'syncDeltaCap' — or after 'syncMaxIters' as a safety valve — we take the lock and reapply only that bounded residual before swapping. The lock hold is thus ~constant (O(syncDeltaCap)) rather than O(occupancy). 'extendReapply' reapplies a delta on top of an already-revalidated state, seeded from its ledger state via 'applyMempoolDiffs', and assembles the result with the same 'buildRevalidatedIS' that 'revalidateTxsFor' uses, so it is byte-identical to a single revalidation of the concatenation (verified by the atomic QSM linearizability test). The candidate is committed via the TMVar, so unlike an optimistic swap it cannot starve. Measured (mempool-state-bench, apply 200us / reapply 20us): max reader stall at ~56k txs drops from ~2.1s to ~300ms, and no longer scales linearly with occupancy; throughput and sync count unchanged. --- .../Consensus/Mempool/Impl/Common.hs | 121 ++++++++++++-- .../Ouroboros/Consensus/Mempool/Update.hs | 153 ++++++++++-------- 2 files changed, 194 insertions(+), 80 deletions(-) diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs index b6843f64c4..dd9c7659ca 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Impl/Common.hs @@ -31,6 +31,7 @@ module Ouroboros.Consensus.Mempool.Impl.Common , RevalidateTxsResult (..) , computeSnapshot , revalidateTxsFor + , extendReapply , validateNewTransaction -- * Tracing @@ -423,19 +424,100 @@ revalidateTxsFor :: [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)] -> RevalidateTxsResult blk revalidateTxsFor capacityOverride cfg slot st values lastTicketNo txTickets = - let inputTxs = map wrap txTickets - inputKeys = Foldable.foldMap' (getTransactionKeySets . txForgetValidated . fst3) inputTxs + let inputTxs = map wrapTxTicket txTickets + inputKeys = Foldable.foldMap' (getTransactionKeySets . txForgetValidated . wtdTx) inputTxs ReapplyTxsResult err validTxs st' = reapplyTxs @blk @Collect cfg slot inputTxs $ applyMempoolDiffs values inputKeys st + in buildRevalidatedIS capacityOverride cfg slot st values lastTicketNo err validTxs st' - outputKeys = Foldable.foldMap' (getTransactionKeySets . txForgetValidated . fst3) validTxs - outputDiffs = Foldable.foldl' rawPrependDiffs (DiffMK mempty) $ map (getLedgerTables . snd3) validTxs +-- | Reapply an additional /delta/ of already-validated transactions on top of a +-- previously revalidated 'InternalState' @cand@, without reprocessing @cand@'s +-- transactions. +-- +-- @cand@ must be the result of revalidating some prefix (by 'TicketNo') of the +-- mempool against @st@ (the same ticked ledger state passed here), and +-- @deltaTxTickets@ the transactions added since, in ascending ticket order, +-- with @deltaValues@ their input values read at @st@'s tip. The result is +-- /byte-identical/ to @'revalidateTxsFor' \@ (cand's txs ++ delta)@: the delta +-- is reapplied against @cand@'s post-reapply ledger state (via +-- 'applyMempoolDiffs' on 'isLedgerState'), and the final 'InternalState' is +-- assembled by the /same/ 'buildRevalidatedIS' over the concatenated survivors. +-- This lets a mempool sync shrink its outstanding work off the lock and take +-- the lock only for a small, bounded final delta ('implSyncWithLedger'). +extendReapply :: + forall blk. + (LedgerSupportsMempool blk, HasTxId (GenTx blk)) => + MempoolCapacityBytesOverride -> + LedgerConfig blk -> + SlotNo -> + -- | The ticked ledger state @cand@ was revalidated against. + TickedLedgerState blk DiffMK -> + -- | The already-revalidated state to extend. + InternalState blk -> + -- | Input values for the delta txs, read at @st@'s tip. + LedgerTables (LedgerState blk) ValuesMK -> + -- | The new 'isLastTicketNo' (the mempool's current ticket counter). + TicketNo -> + -- | The delta txs, in ascending 'TicketNo' order. + [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)] -> + RevalidateTxsResult blk +extendReapply capacityOverride cfg slot st cand deltaValues lastTicketNo deltaTxTickets = + let candTxs = map wrapTxTicket (TxSeq.toList (isTxs cand)) + deltaTxs = map wrapTxTicket deltaTxTickets + deltaKeys = Foldable.foldMap' (getTransactionKeySets . txForgetValidated . wtdTx) deltaTxs + + -- Seed the delta reapplication from @cand@'s post-reapply ledger state, so + -- a delta tx spending one of @cand@'s outputs sees it. This is exactly the + -- state a full reapplication would be in after processing @cand@'s txs. + ReapplyTxsResult errDelta validDelta st' = + reapplyTxs @blk @Collect cfg slot deltaTxs $ + applyMempoolDiffs deltaValues deltaKeys (isLedgerState cand) + + -- @cand@'s txs remain valid (reapplying more txs after them cannot + -- invalidate earlier ones), so the combined survivors are just @cand@'s + -- followed by the delta's, and the values cover both. + allValues = ltliftA2 unionValues (isTxValues cand) deltaValues + in buildRevalidatedIS + capacityOverride + cfg + slot + st + allValues + lastTicketNo + errDelta + (candTxs ++ validDelta) + st' + +-- | Assemble an 'InternalState' from the result of a (re)application: the base +-- ticked ledger @st@, the input @values@, and the reapply's survivors and +-- resulting ledger state. Shared by 'revalidateTxsFor' and 'extendReapply' so +-- both produce identical states. +buildRevalidatedIS :: + forall blk. + (LedgerSupportsMempool blk, HasTxId (GenTx blk)) => + MempoolCapacityBytesOverride -> + LedgerConfig blk -> + SlotNo -> + TickedLedgerState blk DiffMK -> + LedgerTables (LedgerState blk) ValuesMK -> + TicketNo -> + [Invalidated blk] -> + [ ( Validated (GenTx blk) + , LedgerTables (TickedLedgerState blk) DiffMK + , (TicketNo, TxMeasureWithDiffTime blk) + ) + ] -> + TickedLedgerState blk EmptyMK -> + RevalidateTxsResult blk +buildRevalidatedIS capacityOverride cfg slot st values lastTicketNo err validTxs st' = + let outputKeys = Foldable.foldMap' (getTransactionKeySets . txForgetValidated . wtdTx) validTxs + outputDiffs = Foldable.foldl' rawPrependDiffs (DiffMK mempty) $ map (getLedgerTables . wtdDiffs) validTxs in RevalidateTxsResult ( IS - { isTxs = TxSeq.fromList $ map unwrap validTxs - , isTxIds = Set.fromList $ map (txId . txForgetValidated . fst3) validTxs + { isTxs = TxSeq.fromList $ map unwrapTxTicket validTxs + , isTxIds = Set.fromList $ map (txId . txForgetValidated . wtdTx) validTxs , isTxKeys = outputKeys , isTxValues = ltliftA2 restrictValuesMK values outputKeys , isLedgerState = @@ -448,11 +530,28 @@ revalidateTxsFor capacityOverride cfg slot st values lastTicketNo txTickets = } ) err - where - wrap = \(TxTicket (ValidatedTxWithDiffs tx df) tk tz) -> (tx, df, (tk, tz)) - unwrap = \(tx, df, (tk, tz)) -> TxTicket (ValidatedTxWithDiffs tx df) tk tz - fst3 (x, _, _) = x - snd3 (_, x, _) = x + +wrapTxTicket :: + TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk) -> + ( Validated (GenTx blk) + , LedgerTables (TickedLedgerState blk) DiffMK + , (TicketNo, TxMeasureWithDiffTime blk) + ) +wrapTxTicket (TxTicket (ValidatedTxWithDiffs tx df) tk tz) = (tx, df, (tk, tz)) + +unwrapTxTicket :: + ( Validated (GenTx blk) + , LedgerTables (TickedLedgerState blk) DiffMK + , (TicketNo, TxMeasureWithDiffTime blk) + ) -> + TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk) +unwrapTxTicket (tx, df, (tk, tz)) = TxTicket (ValidatedTxWithDiffs tx df) tk tz + +wtdTx :: (a, b, c) -> a +wtdTx (x, _, _) = x + +wtdDiffs :: (a, b, c) -> b +wtdDiffs (_, x, _) = x data RevalidateTxsResult blk = RevalidateTxsResult diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs index 3bce0f5843..566cd1803f 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs @@ -27,7 +27,7 @@ import qualified Data.Text as T import Ouroboros.Consensus.HeaderValidation import Ouroboros.Consensus.Ledger.Abstract import Ouroboros.Consensus.Ledger.SupportsMempool -import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables, unionValues) +import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables) import Ouroboros.Consensus.Mempool.API import Ouroboros.Consensus.Mempool.Capacity import Ouroboros.Consensus.Mempool.Impl.Common @@ -515,6 +515,21 @@ pureRemoveTxs capacityOverride lcfg slot lstate values tkt txs txIds = Sync with ledger -------------------------------------------------------------------------------} +-- | Maximum number of delta transactions 'implSyncWithLedger' will reapply +-- while holding the state lock. It shrinks its outstanding delta off the lock +-- until at most this many txs remain, so the final under-lock reapply — and +-- hence the time snapshot readers can be blocked — is bounded independently of +-- mempool occupancy. +syncDeltaCap :: Int +syncDeltaCap = 256 + +-- | Safety valve on the off-lock shrink loop: if the delta never falls below +-- 'syncDeltaCap' (e.g. reapplication is not actually cheaper than ingestion), +-- stop after this many rounds and finish anyway, degrading to a larger +-- under-lock reapply rather than looping unboundedly. +syncMaxIters :: Int +syncMaxIters = 8 + -- | See 'Ouroboros.Consensus.Mempool.API.testSyncWithLedger' and -- and 'Ouroboros.Consensus.Mempool.Init.forkSyncStateOnTipPointChange'. implSyncWithLedger :: @@ -534,15 +549,21 @@ implSyncWithLedger :: implSyncWithLedger projectResult mpEnv = encloseTimedWith (TraceMempoolSynced >$< mpEnvTracer mpEnv) go where - -- The expensive part — reading all of the mempool's tx inputs from the - -- LedgerDB — is done /off the lock/, against a snapshot @is0@ taken with a - -- non-emptying 'readTMVar', while adds keep appending. We then 'takeTMVar' - -- only briefly: read the (small) input values for the txs added in the - -- meantime (the "delta", by 'TicketNo'), union them with the off-lock values, - -- and run 'pureSyncWithLedger' over the current txs. Since the union covers - -- exactly the current txs' input keys, the resulting state is identical to a - -- single under-lock revalidation. Only that (sub-second) merge holds the - -- lock, so readers block for the merge but not for the big LedgerDB read. + -- Sync with a bounded lock hold. Everything expensive is done /off the lock/, + -- against a snapshot @is0@ taken with a non-emptying 'readTMVar' while adds + -- keep appending: the big LedgerDB read of @is0@'s inputs, the revalidation of + -- @is0@'s txs at the new tip (@cand0@), and then a converging loop that reads + -- the txs added in the meantime (the "delta", by 'TicketNo') and reapplies + -- them /on top/ via 'extendReapply' — never reprocessing what is already done. + -- + -- The delta shrinks each iteration: adds are serialised (the fifo 'MVar's) and + -- pay full validation, whereas the sync only reapplies, which is cheaper per + -- tx, so fewer txs arrive during a round than it processes. Once the delta is + -- at most 'syncDeltaCap' (or we hit 'syncMaxIters'), we 'takeTMVar' and + -- reapply just that bounded residual before swapping. So the lock is held for + -- a near-constant time — O('syncDeltaCap') — independent of mempool occupancy, + -- rather than for a full O(n) revalidation. The committed state is byte- + -- identical to a single revalidation of all current txs ('extendReapply'). go = do MempoolLedgerDBView ls0 meFrk0 <- atomically $ getCurrentLedgerState ldgrInterface is0 <- atomically $ readTMVar istate @@ -574,33 +595,61 @@ implSyncWithLedger projectResult mpEnv = traceWith trcr TraceMempoolTipMovedBetweenSTMBlocks go Right frk -> do - -- OFF-LOCK: big read of the snapshot's input values at the new tip. + let (slot, ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls0 + deltaKeysOf = + Foldable.foldMap' + (getTransactionKeySets . txForgetValidated . validatedTx . txTicketTx) + readDeltaValues ts = + castLedgerTables <$> roforkerReadTables frk (castLedgerTables $ deltaKeysOf ts) + deltaAfter cand is = + TxSeq.toList . snd $ TxSeq.splitAfterTicketNo (isTxs is) (isLastTicketNo cand) + traceRemoved removed sz = + if null removed + then Nothing + else + Just $ + TraceMempoolRemoveTxs (map (\x -> (getInvalidated x, getReason x)) removed) sz + -- Shrink the outstanding delta off the lock, then finish under it + -- with a bounded residual reapply. + shrinkThenCommit cand removedAcc iterN = do + isNow <- atomically $ readTMVar istate + let deltaTickets = deltaAfter cand isNow + if length deltaTickets > syncDeltaCap && iterN < syncMaxIters + then do + deltaValues <- readDeltaValues deltaTickets + let RevalidateTxsResult cand' removed' = + extendReapply capacityOverride cfg slot ls' cand deltaValues (isLastTicketNo isNow) deltaTickets + shrinkThenCommit cand' (removedAcc ++ removed') (iterN + 1) + else withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $ + \isLocked (MempoolLedgerDBView ls _meFrk) -> + if getTipHash ls /= getTipHash ls0 + then -- Tip moved while we worked; retry the whole sync. + pure (Nothing, isLocked) + else do + -- The lock is held, so no add can intervene: reapply + -- only the residual delta (bounded by the cap plus any + -- stragglers that landed while acquiring the lock). + let resTickets = deltaAfter cand isLocked + resValues <- readDeltaValues resTickets + let RevalidateTxsResult isFinal removedF = + extendReapply capacityOverride cfg slot ls' cand resValues (isLastTicketNo isLocked) resTickets + removed = removedAcc ++ removedF + modifyMVar_ forkerMVar (\frkOld -> roforkerClose frkOld >> pure frk) + whenJust (traceRemoved removed (isMempoolSize isFinal)) (traceWith trcr) + pure (Just (projectResult isFinal), isFinal) + -- OFF-LOCK: big read of the snapshot's inputs and initial revalidation. values0 <- castLedgerTables <$> roforkerReadTables frk (castLedgerTables $ isTxKeys is0) - outcome <- - withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $ - \isNow (MempoolLedgerDBView ls _meFrk) -> do - let (slot, ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls - if getTipHash ls /= getTipHash ls0 - then - -- Tip moved again during the off-lock read; retry. - pure (Nothing, isNow) - else do - -- Read only the delta's input keys (small), union with the - -- off-lock values, and revalidate /all/ current txs. - let (_, deltaSeq) = TxSeq.splitAfterTicketNo (isTxs isNow) (isLastTicketNo is0) - deltaKeys = - Foldable.foldMap' - (getTransactionKeySets . txForgetValidated . validatedTx . txTicketTx) - (TxSeq.toList deltaSeq) - valuesDelta <- - castLedgerTables <$> roforkerReadTables frk (castLedgerTables deltaKeys) - let allValues = ltliftA2 unionValues values0 valuesDelta - (isFinal, mTrace) = - pureSyncWithLedger capacityOverride cfg slot ls' allValues isNow - modifyMVar_ forkerMVar (\frkOld -> roforkerClose frkOld >> pure frk) - whenJust mTrace (traceWith trcr) - pure (Just (projectResult isFinal), isFinal) + let RevalidateTxsResult cand0 removed0 = + revalidateTxsFor + capacityOverride + cfg + slot + ls' + values0 + (isLastTicketNo is0) + (TxSeq.toList (isTxs is0)) + outcome <- shrinkThenCommit cand0 removed0 (0 :: Int) case outcome of -- Retry: the forker we opened was never installed, so close it. Nothing -> roforkerClose frk >> go @@ -614,37 +663,3 @@ implSyncWithLedger projectResult mpEnv = , mpEnvLedgerCfg = cfg , mpEnvCapacityOverride = capacityOverride } = mpEnv - --- | Create a 'SyncWithLedger' value representing the values that will need to --- be stored for committing this synchronization with the Ledger. --- --- See the documentation of 'runSyncWithLedger' for more context. -pureSyncWithLedger :: - (LedgerSupportsMempool blk, HasTxId (GenTx blk)) => - MempoolCapacityBytesOverride -> - LedgerConfig blk -> - SlotNo -> - TickedLedgerState blk DiffMK -> - LedgerTables (LedgerState blk) ValuesMK -> - InternalState blk -> - ( InternalState blk - , Maybe (TraceEventMempool blk) - ) -pureSyncWithLedger capacityOverride lcfg slot lstate values istate = - let RevalidateTxsResult is' removed = - revalidateTxsFor - capacityOverride - lcfg - slot - lstate - values - (isLastTicketNo istate) - (TxSeq.toList $ isTxs istate) - mTrace = - if null removed - then - Nothing - else - Just $ - TraceMempoolRemoveTxs (map (\x -> (getInvalidated x, getReason x)) removed) (isMempoolSize is') - in (is', mTrace)