{-# LANGUAGE BangPatterns #-}
{-# 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 Data.Word (Word64)
import GHC.Clock (getMonotonicTimeNSec)
import Options.Applicative
  ( Parser
  , auto
  , execParser
  , fullDesc
  , help
  , helper
  , info
  , long
  , metavar
  , option
  , progDesc
  , showDefault
  , value
  , (<**>)
  )
import Ouroboros.Consensus.Ledger.Basics (LedgerState)
import Ouroboros.Consensus.Ledger.SupportsMempool (ByteSize32 (ByteSize32))
import Ouroboros.Consensus.Ledger.Tables
  ( KeysMK (KeysMK)
  , LedgerTables (LedgerTables)
  , ValuesMK
  , ltliftA2
  , projectLedgerTables
  )
import Ouroboros.Consensus.Ledger.Tables.Utils
  ( emptyLedgerTables
  , forgetLedgerTables
  , restrictValuesMK
  )
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 (setEnv)
import Test.Util.Orphans.IOLike ()

-- * Configuration (command-line options, each with a default)

-- | Benchmark parameters, parsed from the command line by 'configParser'.
data Config = Config
  { Config -> Double
cfgDurationSec :: !Double
  , Config -> Int
cfgNumPeers :: !Int
  -- ^ 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).
  , Config -> Int
cfgNumLocalClients :: !Int
  -- ^ Number of local (N2C) clients. They add on behalf of a local client
  -- (higher fifo priority).
  , Config -> Double
cfgTargetTpsTotal :: !Double
  -- ^ Total target submission rate across all adders (tx/s). @0@ means unbounded
  -- (adders submit as fast as they can), to measure the mempool's max sustained
  -- rate.
  , Config -> Double
cfgSyncPeriodSec :: !Double
  -- ^ 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.
  , Config -> Int
cfgReadBaseMicros :: !Int
  -- ^ Fixed cost of a forker table read (models one on-disk round-trip),
  -- microseconds.
  , Config -> Int
cfgReadPerKeyMicros :: !Int
  -- ^ 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.
  , Config -> Int
cfgReadPeriodMicros :: !Int
  -- ^ 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).
  , Config -> Int
cfgApplyCpuMicros :: !Int
  -- ^ Simulated CPU cost of fully validating a tx (@applyTx@), microseconds.
  -- Exported to the shared 'TestBlock' via @MEMPOOL_APPLY_CPU_US@ (see
  -- 'Bench.Consensus.Mempool.TestBlock.applyCpuMicros').
  , Config -> Int
cfgReapplyCpuMicros :: !Int
  -- ^ Simulated CPU cost of reapplying an already-validated tx (@reapplyTx@),
  -- microseconds. Kept well below 'cfgApplyCpuMicros' to model @reapply ≪ apply@.
  }

configParser :: Parser Config
configParser :: Parser Config
configParser =
  Double
-> Int
-> Int
-> Double
-> Double
-> Int
-> Int
-> Int
-> Int
-> Int
-> Config
Config
    (Double
 -> Int
 -> Int
 -> Double
 -> Double
 -> Int
 -> Int
 -> Int
 -> Int
 -> Int
 -> Config)
-> Parser Double
-> Parser
     (Int
      -> Int
      -> Double
      -> Double
      -> Int
      -> Int
      -> Int
      -> Int
      -> Int
      -> Config)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ReadM Double -> Mod OptionFields Double -> Parser Double
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Double
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"duration"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"SECONDS"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Double -> Mod OptionFields Double
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Double
20
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Double
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"Benchmark duration"
      )
    Parser
  (Int
   -> Int
   -> Double
   -> Double
   -> Int
   -> Int
   -> Int
   -> Int
   -> Int
   -> Config)
-> Parser Int
-> Parser
     (Int
      -> Double -> Double -> Int -> Int -> Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"peers"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"N"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
2
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"Number of N2N peers (each => one tx-submission server/reader and one remote adder)"
      )
    Parser
  (Int
   -> Double -> Double -> Int -> Int -> Int -> Int -> Int -> Config)
-> Parser Int
-> Parser
     (Double -> Double -> Int -> Int -> Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"local-clients"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"N"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
1
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"Number of local (N2C) clients (higher fifo priority)"
      )
    Parser
  (Double -> Double -> Int -> Int -> Int -> Int -> Int -> Config)
-> Parser Double
-> Parser (Double -> Int -> Int -> Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Double -> Mod OptionFields Double -> Parser Double
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Double
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"tps"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"TXS_PER_SEC"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Double -> Mod OptionFields Double
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Double
100
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Double
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"Total target submission rate across all adders; 0 = unbounded"
      )
    Parser (Double -> Int -> Int -> Int -> Int -> Int -> Config)
-> Parser Double
-> Parser (Int -> Int -> Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Double -> Mod OptionFields Double -> Parser Double
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Double
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"sync-period"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"SECONDS"
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Double -> Mod OptionFields Double
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Double
5
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Double
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Double
-> Mod OptionFields Double -> Mod OptionFields Double
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Double
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"How often the syncer advances the tip and revalidates"
      )
    Parser (Int -> Int -> Int -> Int -> Int -> Config)
-> Parser Int -> Parser (Int -> Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"read-base-us"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"US"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
0
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help
            [Char]
"Fixed per-read cost. 0 by default: the measured LSM read cost scales \
            \linearly with the number of txs (no batching — a round trip per tx), \
            \so it is charged per key below, not as a fixed component."
      )
    Parser (Int -> Int -> Int -> Int -> Config)
-> Parser Int -> Parser (Int -> Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"read-per-key-us"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"US"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
60
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help
            [Char]
"Per-key cost of a forker table read. Default 60us, midpoint of the \
            \~40-80us/tx measured on the LSM backend (see \
            \input-output-hk/ouroboros-leios#553); this bench's txs have ~1 input \
            \each, so per-key ~= per-tx here."
      )
    Parser (Int -> Int -> Int -> Config)
-> Parser Int -> Parser (Int -> Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"read-period-us"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"US"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
150000
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help [Char]
"Pause between successive getSnapshots per reader; 0 = tight loop"
      )
    Parser (Int -> Int -> Config)
-> Parser Int -> Parser (Int -> Config)
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"apply-us"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"US"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
128
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help
            [Char]
"Simulated CPU cost of fully validating a tx (applyTx). Default 128us, \
            \the measured median applyBlock cost (see \
            \input-output-hk/ouroboros-leios#553). Exported via MEMPOOL_APPLY_CPU_US."
      )
    Parser (Int -> Config) -> Parser Int -> Parser Config
forall a b. Parser (a -> b) -> Parser a -> Parser b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ReadM Int -> Mod OptionFields Int -> Parser Int
forall a. ReadM a -> Mod OptionFields a -> Parser a
option
      ReadM Int
forall a. Read a => ReadM a
auto
      ( [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasName f => [Char] -> Mod f a
long [Char]
"reapply-us"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. HasMetavar f => [Char] -> Mod f a
metavar [Char]
"US"
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Int -> Mod OptionFields Int
forall (f :: * -> *) a. HasValue f => a -> Mod f a
value Int
20
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> Mod OptionFields Int
forall a (f :: * -> *). Show a => Mod f a
showDefault
          Mod OptionFields Int
-> Mod OptionFields Int -> Mod OptionFields Int
forall a. Semigroup a => a -> a -> a
<> [Char] -> Mod OptionFields Int
forall (f :: * -> *) a. [Char] -> Mod f a
help
            [Char]
"Simulated CPU cost of reapplying an already-validated tx (reapplyTx). \
            \Default 20us, kept well below --apply-us. Exported via MEMPOOL_REAPPLY_CPU_US."
      )

-- | tx-submission servers = one per peer.
numReaders :: Config -> Int
numReaders :: Config -> Int
numReaders Config
cfg = Config -> Int
cfgNumPeers Config
cfg

-- | tx-submission clients = one per peer (N2N) + the local clients (N2C).
numAdders :: Config -> Int
numAdders :: Config -> Int
numAdders Config
cfg = Config -> Int
cfgNumPeers Config
cfg Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Config -> Int
cfgNumLocalClients Config
cfg

-- | Disjoint token namespace per adder so their chains never collide.
chainStride :: Int
chainStride :: Int
chainStride = Int
1_000_000_000

capacityOverride :: MempoolCapacityBytesOverride
capacityOverride :: MempoolCapacityBytesOverride
capacityOverride = ByteSize32 -> MempoolCapacityBytesOverride
MempoolCapacityBytesOverride (Word32 -> ByteSize32
ByteSize32 Word32
100_000_000)

-- * Main

main :: IO ()
IO ()
main = do
  cfg <-
    ParserInfo Config -> IO Config
forall a. ParserInfo a -> IO a
execParser (ParserInfo Config -> IO Config) -> ParserInfo Config -> IO Config
forall a b. (a -> b) -> a -> b
$
      Parser Config -> InfoMod Config -> ParserInfo Config
forall a. Parser a -> InfoMod a -> ParserInfo a
info
        (Parser Config
configParser Parser Config -> Parser (Config -> Config) -> Parser Config
forall (f :: * -> *) a b. Applicative f => f a -> f (a -> b) -> f b
<**> Parser (Config -> Config)
forall a. Parser (a -> a)
helper)
        ( InfoMod Config
forall a. InfoMod a
fullDesc
            InfoMod Config -> InfoMod Config -> InfoMod Config
forall a. Semigroup a => a -> a -> a
<> [Char] -> InfoMod Config
forall a. [Char] -> InfoMod a
progDesc
              [Char]
"Concurrent benchmark of the real mempool's shared-state access under Leios-scale load"
        )
  -- Export the apply/reapply CPU costs to the shared 'TestBlock' before anything
  -- forces its 'NOINLINE' CAFs (they read these env vars once via
  -- 'unsafePerformIO'). The 'TestBlock' defaults to 0 so the criterion
  -- 'mempool-bench' is unaffected; only this bench opts in.
  setEnv "MEMPOOL_APPLY_CPU_US" (show (cfgApplyCpuMicros cfg))
  setEnv "MEMPOOL_REAPPLY_CPU_US" (show (cfgReapplyCpuMicros cfg))
  putStr $
    unlines
      [ "Mempool shared-state concurrent benchmark (real mempool)"
      , "  duration      : " <> show (cfgDurationSec cfg) <> " s"
      , "  peers         : "
          <> show (cfgNumPeers cfg)
          <> " (=> "
          <> show (numReaders cfg)
          <> " servers/readers)"
      , "  clients       : "
          <> show (numAdders cfg)
          <> " ("
          <> show (cfgNumPeers cfg)
          <> " N2N + "
          <> show (cfgNumLocalClients cfg)
          <> " local, target "
          <> show (cfgTargetTpsTotal cfg)
          <> " tx/s total)"
      , "  sync period   : " <> show (cfgSyncPeriodSec cfg) <> " s"
      , "  forker read   : "
          <> show (cfgReadBaseMicros cfg)
          <> " us + "
          <> show (cfgReadPerKeyMicros cfg)
          <> " us/key"
      , "  tx cpu cost   : "
          <> show (cfgApplyCpuMicros cfg)
          <> " us apply / "
          <> show (cfgReapplyCpuMicros cfg)
          <> " us reapply"
      , ""
      ]
  let seeds = [Int -> Token
Token (Int
j Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
chainStride) | Int
j <- [Int
0 .. Config -> Int
numAdders Config
cfg Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1]]
      baseLedger = [Token] -> LedgerState TestBlock ValuesMK
mkInitialLedgerState [Token]
seeds
  ledgerVar <- newTVarIO baseLedger
  mempool <-
    openMempoolWithoutSyncThread
      (latencyLedgerInterface cfg 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 <- IO Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        pure (realToFrac (diffTime now start) >= cfgDurationSec cfg)

  syncer <- async (runSyncer cfg expired mempool ledgerVar baseLedger syncDursRef)
  readers <-
    forM [1 .. numReaders cfg] $ \Int
_ ->
      IO () -> IO (Async ())
forall a. IO a -> IO (Async a)
async (Config
-> IO Bool
-> Mempool IO TestBlock
-> IORef Int
-> IORef DiffTime
-> IO ()
runReader Config
cfg IO Bool
expired Mempool IO TestBlock
mempool IORef Int
readRef IORef DiffTime
maxReadLatRef)
  adders <-
    forM [0 .. numAdders cfg - 1] $ \Int
j -> do
      let onBehalf :: AddTxOnBehalfOf
onBehalf = if Int
j Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Config -> Int
cfgNumPeers Config
cfg then AddTxOnBehalfOf
AddTxForRemotePeer else AddTxOnBehalfOf
AddTxForLocalClient
      IO () -> IO (Async ())
forall a. IO a -> IO (Async a)
async (Config
-> IO Bool
-> Mempool IO TestBlock
-> AddTxOnBehalfOf
-> Int
-> IORef Int
-> IO ()
runAdder Config
cfg IO Bool
expired Mempool IO TestBlock
mempool AddTxOnBehalfOf
onBehalf Int
j IORef Int
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 = DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Time -> Time -> DiffTime
diffTime Time
end Time
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 ::
  Config -> IO Bool -> Mempool IO TestBlock -> AddTxOnBehalfOf -> Int -> IORef Int -> IO ()
runAdder :: Config
-> IO Bool
-> Mempool IO TestBlock
-> AddTxOnBehalfOf
-> Int
-> IORef Int
-> IO ()
runAdder Config
cfg IO Bool
expired Mempool IO TestBlock
mempool AddTxOnBehalfOf
onBehalf Int
j IORef Int
addedRef = Int -> IO ()
go Int
0
 where
  base :: Int
base = Int
j Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
chainStride

  intervalMicros :: Int
intervalMicros =
    if Config -> Double
cfgTargetTpsTotal Config
cfg Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
<= Double
0
      then Int
0
      else Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (Double
1_000_000 Double -> Double -> Double
forall a. Num a => a -> a -> a
* Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Config -> Int
numAdders Config
cfg) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Config -> Double
cfgTargetTpsTotal Config
cfg)

  go :: Int -> IO ()
go !Int
i = do
    done <- IO Bool
expired
    if done
      then pure ()
      else do
        let tx = [Token] -> [Token] -> GenTx TestBlock
mkTx [Int -> Token
Token (Int
base Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
i)] [Int -> Token
Token (Int
base Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
i Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)]
        r <- addTx mempool onBehalf tx
        when (isMempoolTxAdded r) $ atomicModifyIORef' addedRef (\Int
c -> (Int
c Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1, ()))
        when (intervalMicros > 0) $ Conc.threadDelay intervalMicros
        go (i + 1)

-- | Reader: real 'getSnapshot' on the configured per-reader cadence
-- ('cfgReadPeriodMicros'), recording max read latency.
runReader :: Config -> IO Bool -> Mempool IO TestBlock -> IORef Int -> IORef DiffTime -> IO ()
runReader :: Config
-> IO Bool
-> Mempool IO TestBlock
-> IORef Int
-> IORef DiffTime
-> IO ()
runReader Config
cfg IO Bool
expired Mempool IO TestBlock
mempool IORef Int
readRef IORef DiffTime
maxLatRef = IO ()
go
 where
  go :: IO ()
go = do
    done <- IO Bool
expired
    if done
      then pure ()
      else do
        t0 <- getMonotonicTime
        snap <- atomically (getSnapshot mempool)
        _ <- evaluate (length (snapshotTxs snap))
        t1 <- getMonotonicTime
        let lat = Time -> Time -> DiffTime
diffTime Time
t1 Time
t0
        atomicModifyIORef' readRef (\Int
c -> (Int
c Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1, ()))
        atomicModifyIORef' maxLatRef (\DiffTime
m -> (DiffTime -> DiffTime -> DiffTime
forall a. Ord a => a -> a -> a
max DiffTime
m DiffTime
lat, ()))
        when (cfgReadPeriodMicros cfg > 0) $ Conc.threadDelay (cfgReadPeriodMicros cfg)
        go

-- | Syncer: every 'cfgSyncPeriodSec', advance the tip and run the real sync.
runSyncer ::
  Config ->
  IO Bool ->
  Mempool IO TestBlock ->
  StrictTVar IO (LedgerState TestBlock ValuesMK) ->
  LedgerState TestBlock ValuesMK ->
  IORef [DiffTime] ->
  IO ()
runSyncer :: Config
-> IO Bool
-> Mempool IO TestBlock
-> StrictTVar IO (LedgerState TestBlock ValuesMK)
-> LedgerState TestBlock ValuesMK
-> IORef [DiffTime]
-> IO ()
runSyncer Config
cfg IO Bool
expired Mempool IO TestBlock
mempool StrictTVar IO (LedgerState TestBlock ValuesMK)
ledgerVar LedgerState TestBlock ValuesMK
baseLedger IORef [DiffTime]
syncDursRef = Word64 -> IO ()
go Word64
1
 where
  go :: Word64 -> IO ()
go !Word64
n = do
    Int -> IO ()
Conc.threadDelay (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (Double
1_000_000 Double -> Double -> Double
forall a. Num a => a -> a -> a
* Config -> Double
cfgSyncPeriodSec Config
cfg))
    done <- IO Bool
expired
    if done
      then pure ()
      else do
        atomically $ writeTVar ledgerVar (advanceTip n baseLedger)
        t0 <- getMonotonicTime
        _ <- testSyncWithLedger mempool
        t1 <- getMonotonicTime
        atomicModifyIORef' syncDursRef (\[DiffTime]
ds -> (Time -> Time -> DiffTime
diffTime Time
t1 Time
t0 DiffTime -> [DiffTime] -> [DiffTime]
forall a. a -> [a] -> [a]
: [DiffTime]
ds, ()))
        go (n + 1)

-- * Latency-injecting ledger interface

latencyLedgerInterface ::
  Config ->
  StrictTVar IO (LedgerState TestBlock ValuesMK) ->
  LedgerInterface IO TestBlock
latencyLedgerInterface :: Config
-> StrictTVar IO (LedgerState TestBlock ValuesMK)
-> LedgerInterface IO TestBlock
latencyLedgerInterface Config
cfg StrictTVar IO (LedgerState TestBlock ValuesMK)
ledgerVar =
  LedgerInterface
    { getCurrentLedgerState :: STM IO (MempoolLedgerDBView IO TestBlock)
getCurrentLedgerState = do
        st <- StrictTVar IO (LedgerState TestBlock ValuesMK)
-> STM IO (LedgerState TestBlock ValuesMK)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO (LedgerState TestBlock ValuesMK)
ledgerVar
        pure $
          MempoolLedgerDBView
            (forgetLedgerTables st)
            ( pure $
                Right $
                  ReadOnlyForker
                    { roforkerClose = pure ()
                    , roforkerGetLedgerState = pure (forgetLedgerTables st)
                    , roforkerReadTables = \LedgerTables TestBlock KeysMK
keys -> do
                        -- Busy-wait, not 'threadDelay': the read latency here is
                        -- often sub-millisecond, below 'threadDelay's timer
                        -- granularity, and it is on the critical path (held under
                        -- the mempool lock), so rounding it up to ~1ms would
                        -- badly distort the results.
                        Int -> IO ()
spinMicros (Config -> Int
cfgReadBaseMicros Config
cfg Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Config -> Int
cfgReadPerKeyMicros Config
cfg Int -> Int -> Int
forall a. Num a => a -> a -> a
* LedgerTables TestBlock KeysMK -> Int
keysCount LedgerTables TestBlock KeysMK
keys)
                        LedgerTables TestBlock ValuesMK
-> IO (LedgerTables TestBlock ValuesMK)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ((forall k v.
 LedgerTableConstraints' TestBlock k v =>
 ValuesMK k v -> KeysMK k v -> ValuesMK k v)
-> LedgerTables TestBlock ValuesMK
-> LedgerTables TestBlock KeysMK
-> LedgerTables TestBlock ValuesMK
forall l (mk1 :: MapKind) (mk2 :: MapKind) (mk3 :: MapKind).
LedgerTableConstraints l =>
(forall k v.
 LedgerTableConstraints' TestBlock k v =>
 mk1 k v -> mk2 k v -> mk3 k v)
-> LedgerTables l mk1 -> LedgerTables l mk2 -> LedgerTables l mk3
ltliftA2 ValuesMK k v -> KeysMK k v -> ValuesMK k v
forall k v. Ord k => ValuesMK k v -> KeysMK k v -> ValuesMK k v
forall k v.
LedgerTableConstraints' TestBlock k v =>
ValuesMK k v -> KeysMK k v -> ValuesMK k v
restrictValuesMK (LedgerState TestBlock ValuesMK -> LedgerTables TestBlock ValuesMK
forall (mk :: MapKind).
(CanMapMK mk, CanMapKeysMK mk, ZeroableMK mk) =>
LedgerState TestBlock mk -> LedgerTables TestBlock mk
forall (l :: StateKind) blk (mk :: MapKind).
(HasLedgerTables l blk, CanMapMK mk, CanMapKeysMK mk,
 ZeroableMK mk) =>
l blk mk -> LedgerTables blk mk
projectLedgerTables LedgerState TestBlock ValuesMK
st) LedgerTables TestBlock KeysMK
keys)
                    , roforkerReadStatistics = pure (Statistics 0)
                    , roforkerRangeReadTables = \RangeQueryPrevious TestBlock
_ -> (LedgerTables TestBlock ValuesMK, Maybe Token)
-> IO (LedgerTables TestBlock ValuesMK, Maybe Token)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (LedgerTables TestBlock ValuesMK
forall (mk :: MapKind) blk.
(ZeroableMK mk, LedgerTableConstraints blk) =>
LedgerTables blk mk
emptyLedgerTables, Maybe Token
forall a. Maybe a
Nothing)
                    }
            )
    }

keysCount :: LedgerTables TestBlock KeysMK -> Int
keysCount :: LedgerTables TestBlock KeysMK -> Int
keysCount (LedgerTables (KeysMK Set (TxIn TestBlock)
s)) = Set Token -> Int
forall a. Set a -> Int
Set.size Set (TxIn TestBlock)
Set Token
s

-- | Busy-wait for the given number of microseconds. Unlike 'Conc.threadDelay',
-- this honours sub-millisecond durations (which the RTS timer would round up),
-- at the cost of burning a core — appropriate for modelling a latency that sits
-- on the mempool's critical path.
spinMicros :: Int -> IO ()
spinMicros :: Int -> IO ()
spinMicros Int
us
  | Int
us Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
0 = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  | Bool
otherwise = do
      let target :: Word64
target = Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
us Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
* Word64
1_000 :: Word64
      start <- IO Word64
getMonotonicTimeNSec
      let go = do
            now <- IO Word64
getMonotonicTimeNSec
            when (now - start < target) go
      go

-- * Formatting

showT :: DiffTime -> String
showT :: DiffTime -> [Char]
showT DiffTime
t
  | Double
s Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
1e-3 = Int -> [Char]
forall a. Show a => a -> [Char]
show (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (Double
s Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
1_000_000) :: Int) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" us"
  | Double
s Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
1 = Int -> [Char]
forall a. Show a => a -> [Char]
show (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (Double
s Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
1_000) :: Int) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" ms"
  | Bool
otherwise = Double -> [Char]
forall a. Show a => a -> [Char]
show (Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
round (Double
s Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
1000) :: Int) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1000 :: Double) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" s"
 where
  s :: Double
s = DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
t :: Double