diff --git a/ouroboros-consensus.cabal b/ouroboros-consensus.cabal index 3cf8155362..e293a858be 100644 --- a/ouroboros-consensus.cabal +++ b/ouroboros-consensus.cabal @@ -945,6 +945,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..70264ea2fe 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,8 @@ import Data.MemPack 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 @@ -48,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 @@ -86,15 +93,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 $ @@ -223,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 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..7ae6fbbf99 --- /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, modelling the proto-devnet baseline. +-- +-- We open the actual mempool ('openMempoolWithoutSyncThread') over a mocked +-- ledger interface whose forker reads inject a configurable latency to model +-- the on-disk (LSM) UTxO reads. Three roles run concurrently against it, as in +-- a node under tx-submission load: +-- +-- * __Adders__ (tx-submission clients): each submits an independent chain of +-- transactions via the real 'addTx', rate-limited to a target TPS (like the +-- tx-firehose). +-- +-- * __Readers__ (tx-submission servers / forging): tight loop of the real +-- 'getSnapshot' (@readTMVar istate@), measuring how long a read blocks. +-- +-- * __Syncer__ (the mempool sync thread): periodically advances the ledger tip +-- and runs the real 'testSyncWithLedger', which revalidates the whole mempool +-- through the latency-injected forker while holding the state lock. +-- +-- The goal of this first version is to reproduce the baseline: the mempool +-- keeps up with the submission rate, with periodic reader/adder stalls whose +-- size grows with occupancy — matching what we measure on the devnet. +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 exactly 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). For the +-- frozen 3-node baseline, node1 has 2 peers (node2, node3). +{-# NOINLINE numPeers #-} +numPeers :: Int +numPeers = envInt "PEERS" 2 + +-- | Number of local (N2C) clients — the tx-firehose(s). They add on behalf of a +-- local client (higher fifo priority). Baseline runs 1. +{-# 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 firehose(s) (N2C). +numAdders :: Int +numAdders = numPeers + numLocalClients + +-- | Total target submission rate across all adders (tx/s). Matches the +-- proto-devnet baseline firehose (TPS=100). @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. The devnet adopts a +-- block roughly every ~20 s; a shorter period here exercises the sync +-- contention more often. +{-# NOINLINE syncPeriodSec #-} +syncPeriodSec :: Double +syncPeriodSec = envDouble "SYNC_PERIOD" 5 + +-- | Fixed cost of a forker table read (models one LSM 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 LSM +-- 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. In the proto-devnet each downstream +-- peer pulled ~3–4 tx-body requests/s from node1 (node2 2.7/s, node3 4.0/s over +-- multi-hour runs), 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 the old 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' in a tight loop, 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 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..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 @@ -264,6 +265,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 +303,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 @@ -415,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 = @@ -440,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/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..566cd1803f 100644 --- a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs +++ b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Mempool/Update.hs @@ -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 @@ -514,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 :: @@ -531,74 +547,114 @@ 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 + -- 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 + 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 + 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) + 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 + Just r -> pure r + MempoolEnv { mpEnvStateVar = istate , mpEnvForker = forkerMVar @@ -607,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) 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 $