diff --git a/cardano-diffusion/changelog.d/20260216_123142_crocodile-dentist_fix_tx_submission.md b/cardano-diffusion/changelog.d/20260216_123142_crocodile-dentist_fix_tx_submission.md new file mode 100644 index 00000000000..4e2c0303140 --- /dev/null +++ b/cardano-diffusion/changelog.d/20260216_123142_crocodile-dentist_fix_tx_submission.md @@ -0,0 +1,23 @@ + + + +### Non-Breaking + +- Integrate TVar to collect duplicate tx's in the mempool writer + + diff --git a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet.hs b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet.hs index 2617359e0dc..7fe843bc14b 100644 --- a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet.hs +++ b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet.hs @@ -1090,10 +1090,14 @@ prop_check_inflight_ratio bi ds@(DiffusionScript simArgs _ _) = txDecisionPolicy = saTxDecisionPolicy simArgs in tabulate "Max observeed ratio of inflight multiplicity by the max stipulated by the policy" - (map (\m -> "has " ++ show m ++ " in flight - ratio: " - ++ show @(Ratio Int) (fromIntegral m / fromIntegral (txInflightMultiplicity txDecisionPolicy)) - ) - (Map.elems inflightTxsMap)) + (let maxAllowed = txInflightMultiplicity txDecisionPolicy + inflightCounts = map inFlightCount (Map.elems inflightTxsMap) + in map (\m -> "has " ++ show m ++ " in flight - ratio: " + ++ if maxAllowed > 0 + then show @(Ratio Int) (m % maxAllowed) + else "n/a" + ) + inflightCounts) True -- | This test coverage of InboundGovernor transitions. diff --git a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs index e12aea95372..6f56fa53972 100644 --- a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs +++ b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs @@ -276,6 +276,7 @@ applications :: forall block header s m. -> LimitsAndTimeouts header block -> AppArgs header block m -> (block -> header) + -> LazySTM.TVar m [TxId] -> Diffusion.Applications NtNAddr NtNVersion NtNVersionData NtCAddr NtCVersion NtCVersionData PeerTrustable m () @@ -298,7 +299,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , aaPeerMetrics , aaTxDecisionPolicy } - toHeader = + toHeader + duplicateTxVar = Diffusion.Applications { Diffusion.daApplicationInitiatorMode = simpleSingletonVersions UnversionedProtocol @@ -727,13 +729,13 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node aaTxDecisionPolicy sharedTxStateVar (getMempoolReader mempool) - (getMempoolWriter mempool) + (getMempoolWriter duplicateTxVar mempool) getTxSize them $ \api -> do let server = txSubmissionInboundV2 txSubmissionInboundTracer NoTxSubmissionInitDelay - (getMempoolWriter mempool) + (getMempoolWriter duplicateTxVar mempool) api labelThisThread "TxSubmissionServer" runPipelinedPeerWithLimits diff --git a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs index 40e57d97d40..fada44f69c3 100644 --- a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs +++ b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs @@ -38,6 +38,7 @@ module Test.Cardano.Network.Diffusion.Testnet.Simulation import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadMVar (MonadMVar) +import Control.Concurrent.Class.MonadSTM qualified as LazySTM import Control.Concurrent.Class.MonadSTM.Strict import Control.Monad (forM, when) import Control.Monad.Class.MonadAsync @@ -1171,6 +1172,7 @@ diffusionSimulationM churnModeVar <- newTVarIO ChurnModeNormal peerMetrics <- newPeerMetric PeerMetricsConfiguration { maxEntriesToTrack = 180 } policyStdGenVar <- newTVarIO (mkStdGen 12) + duplicateTxVar <- LazySTM.newTVarIO [] let readUseBootstrapPeers = stepScriptSTM' useBootstrapPeersScriptVar (bgaRng, rng) = Random.splitGen $ mkStdGen seed @@ -1340,6 +1342,7 @@ diffusionSimulationM limitsAndTimeouts appArgs blockHeader + duplicateTxVar where tracerTxSubmissionInbound = contramap DiffusionTxSubmissionInbound diff --git a/ouroboros-network/bench/Main.hs b/ouroboros-network/bench/Main.hs index 87ad6a34340..497e8a8dbee 100644 --- a/ouroboros-network/bench/Main.hs +++ b/ouroboros-network/bench/Main.hs @@ -1,16 +1,27 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE NumericUnderscores #-} +-- pPrint +{-# OPTIONS_GHC -Wno-unused-imports #-} + module Main (main) where import Control.DeepSeq import Control.Exception (evaluate) +import Control.Monad.Class.MonadTime.SI (Time) +import Data.Map.Strict qualified as Map +import Data.Set qualified as Set import Debug.Trace (traceMarkerIO) -import System.Random.SplitMix qualified as SM +import System.Mem (performMajorGC) import Test.Tasty.Bench +import Text.Pretty.Simple (pPrint) import Ouroboros.Network.TxSubmission.Inbound.V2.Decision qualified as Tx +import Ouroboros.Network.TxSubmission.Inbound.V2.Policy (TxDecisionPolicy) +import Ouroboros.Network.TxSubmission.Inbound.V2.State (SharedTxState (..)) import Test.Ouroboros.Network.TxSubmission.TxLogic qualified as TX - (mkDecisionContext) + (PeerAddr, mkDecisionContexts, printTxLogicBenchmarkContexts) +import Test.Ouroboros.Network.TxSubmission.Types (Tx, TxId) import Test.Ouroboros.Network.PeerSelection.PeerMetric (microbenchmark1GenerateInput, microbenchmark1ProcessInput) @@ -28,47 +39,108 @@ main = bench "100k" $ nfAppIO microbenchmark1ProcessInput i ] , bgroup "TxLogic" - [ env (do let a = TX.mkDecisionContext (SM.mkSMGen 131) 10 + [ env (do let a = TX.mkDecisionContexts 131 100 10 evaluate (rnf a) +#ifdef TXLOGIC_PRINT + TX.printTxLogicBenchmarkContexts a +#endif + -- pPrint a + performMajorGC traceMarkerIO "evaluated decision context" return a ) - (\a -> - bench "makeDecisions: 10" - $ nf (uncurry Tx.makeDecisions) a + (\as -> + bench "makeDecisions: 100 x 10" + $ let run (now, policy, state) = + Tx.makeDecisions now policy state (peerTxStates state) + in nf (map run) as ) - , env (do let a = TX.mkDecisionContext (SM.mkSMGen 131) 100 + , env (do let a = TX.mkDecisionContexts 131 100 100 evaluate (rnf a) +#ifdef TXLOGIC_PRINT + TX.printTxLogicBenchmarkContexts a +#endif + -- pPrint a + performMajorGC traceMarkerIO "evaluated decision context" return a ) - (\a -> - bench "makeDecisions: 100" - $ nf (uncurry Tx.makeDecisions) a + (\as -> + bench "makeDecisions: 100 x 100" + $ let run (now, policy, state) = + Tx.makeDecisions now policy state (peerTxStates state) + in nf (map run) as ) - , env (do let a = TX.mkDecisionContext (SM.mkSMGen 361) 1_000 + , env (do let a = TX.mkDecisionContexts 361 100 1_000 evaluate (rnf a) +#ifdef TXLOGIC_PRINT + TX.printTxLogicBenchmarkContexts a +#endif + -- pPrint a + performMajorGC traceMarkerIO "evaluated decision context" return a ) - (\a -> - bench "makeDecisions: 1000" - $ nf (uncurry Tx.makeDecisions) a + (\as -> + bench "makeDecisions: 100 x 1000" + $ let run (now, policy, state) = + Tx.makeDecisions now policy state (peerTxStates state) + in nf (map run) as ) -{- - , env (do - smGen <- SM.initSMGen - print smGen - let a = TX.mkDecisionContext smGen 1000 + , env (do let a = mkPendingContexts 361 100 1_000 0.0 evaluate (rnf a) - traceMarkerIO "evaluated decision context" + performMajorGC + traceMarkerIO "evaluated decision context (pending 0%)" + return a + ) + (\as -> + bench "makeDecisions+filterActivePeers: 1000/0% pending" + $ let run (now, policy, st) = + Tx.makeDecisions now policy st + (Tx.filterActivePeers now policy st) + in nf (map run) as + ) + , env (do let a = mkPendingContexts 361 100 1_000 0.5 + evaluate (rnf a) + performMajorGC + traceMarkerIO "evaluated decision context (pending 50%)" return a ) - (\a -> - bench "makeDecisions: random" - $ nf (uncurry Tx.makeDecisions) a + (\as -> + bench "makeDecisions+filterActivePeers: 1000/50% pending" + $ let run (now, policy, st) = + Tx.makeDecisions now policy st + (Tx.filterActivePeers now policy st) + in nf (map run) as + ) + , env (do let a = mkPendingContexts 361 100 1_000 0.9 + evaluate (rnf a) + performMajorGC + traceMarkerIO "evaluated decision context (pending 90%)" + return a + ) + (\as -> + bench "makeDecisions+filterActivePeers: 1000/90% pending" + $ let run (now, policy, st) = + Tx.makeDecisions now policy st + (Tx.filterActivePeers now policy st) + in nf (map run) as ) --} ] ] ] + +mkPendingContexts + :: Int + -> Int + -> Int + -> Double + -> [ (Time, TxDecisionPolicy, SharedTxState TX.PeerAddr TxId (Tx TxId)) ] +mkPendingContexts seed count size pendingRatio = + map applyPending (TX.mkDecisionContexts seed count size) + where + applyPending (now, policy, st) = + let peers = Map.keysSet (peerTxStates st) + pendingCount = floor (pendingRatio * fromIntegral (Set.size peers)) + pendingSet = Set.fromList (take pendingCount (Set.toList peers)) + in (now, policy, st { pendingDecisions = pendingSet }) diff --git a/ouroboros-network/changelog.d/20260216_100847_crocodile-dentist_fix_tx_submission.md b/ouroboros-network/changelog.d/20260216_100847_crocodile-dentist_fix_tx_submission.md new file mode 100644 index 00000000000..343e391134b --- /dev/null +++ b/ouroboros-network/changelog.d/20260216_100847_crocodile-dentist_fix_tx_submission.md @@ -0,0 +1,27 @@ + + + + +### Non-Breaking + +- tx-submission: Ensure all eligible downloaded tx's will be submitted to the mempool +- tx-submission: Enforce that no transaction is enqueued to the mempool more than once by the same peer +- tx-submission: Improve testcase generation and inflight test +- tx-submission: Remove global size limit for inflight tx's + + diff --git a/ouroboros-network/changelog.d/20260226_075458_karl.fb.knutsson_fix_tx_submission.md b/ouroboros-network/changelog.d/20260226_075458_karl.fb.knutsson_fix_tx_submission.md new file mode 100644 index 00000000000..ebba0faa48f --- /dev/null +++ b/ouroboros-network/changelog.d/20260226_075458_karl.fb.knutsson_fix_tx_submission.md @@ -0,0 +1,6 @@ +### Non-Breaking + +- tx-submission v2: improve TxLogic benchmark by running it 100 times with different contexts. +- tx-submission v2: space out requests for the same TX by 200ms to reduce load. +- tx-submission v2: Avoid making decisions for peers that already have pending desisions. + diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2.hs index cf25071361e..5c8d93eae45 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2.hs @@ -60,7 +60,7 @@ txSubmissionInboundV2 readTxDecision, handleReceivedTxIds, handleReceivedTxs, - submitTxToMempool + submitTxsToMempool } = TxSubmissionServerPipelined $ do @@ -82,12 +82,12 @@ txSubmissionInboundV2 -- Only attempt to add TXs if we have some work to do when (collected > 0) $ do - -- submitTxToMempool traces: + -- submitTxsToMempool traces: -- * `TraceTxSubmissionProcessed`, -- * `TraceTxInboundAddedToMempool`, and -- * `TraceTxInboundRejectedFromMempool` -- events. - mapM_ (uncurry $ submitTxToMempool tracer) listOfTxsToMempool + submitTxsToMempool tracer listOfTxsToMempool -- TODO: -- We can update the state so that other `tx-submission` servers will diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Decision.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Decision.hs index 4bed1fc31bc..f940688c537 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Decision.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Decision.hs @@ -18,6 +18,7 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Decision import Control.Arrow ((>>>)) import Control.Exception (assert) +import Control.Monad.Class.MonadTime.SI (addTime, Time) import Data.Bifunctor (second) import Data.Hashable @@ -46,7 +47,9 @@ makeDecisions , Ord txid , Hashable peeraddr ) - => TxDecisionPolicy + => Time + -- ^ current time + -> TxDecisionPolicy -- ^ decision policy -> SharedTxState peeraddr txid tx -- ^ decision context @@ -60,11 +63,11 @@ makeDecisions -> ( SharedTxState peeraddr txid tx , Map peeraddr (TxDecision txid tx) ) -makeDecisions policy st = +makeDecisions now policy st = let (salt, rng') = random (peerRng st) st' = st { peerRng = rng' } in fn - . pickTxsToDownload policy st' + . pickTxsToDownload now policy st' . orderByRejections salt where fn :: forall a. @@ -93,19 +96,15 @@ orderByRejections salt = -- | Internal state of `pickTxsToDownload` computation. -- data St peeraddr txid tx = - St { stInflightSize :: !SizeInBytes, - -- ^ size of all `tx`s in-flight. - - stInflight :: !(Map txid Int), + St { stInflight :: !(Map txid InFlightState), -- ^ `txid`s in-flight. stAcknowledged :: !(Map txid Int), -- ^ acknowledged `txid` with multiplicities. It is used to update -- `referenceCounts`. - stInSubmissionToMempoolTxs :: !(Set txid) - -- ^ TXs on their way to the mempool. Used to prevent issueing new - -- fetch requests for them. + stNewInSubmissionToMempoolTxs :: !(Set txid) + -- ^ TXs newly selected for mempool submission in this decision round. } @@ -126,7 +125,9 @@ pickTxsToDownload ( Ord peeraddr , Ord txid ) - => TxDecisionPolicy + => Time + -- ^ current time + -> TxDecisionPolicy -- ^ decision policy -> SharedTxState peeraddr txid tx -- ^ shared state @@ -136,12 +137,11 @@ pickTxsToDownload , [(peeraddr, TxDecision txid tx)] ) -pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, - maxTxsSizeInflight, - txInflightMultiplicity } +pickTxsToDownload now policy@TxDecisionPolicy { txsSizeInflightPerPeer, + txInflightMultiplicity, + interTxSpace } sharedState@SharedTxState { peerTxStates, inflightTxs, - inflightTxsSize, bufferedTxs, inSubmissionToMempoolTxs, referenceCounts } = @@ -150,9 +150,8 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, accumFn -- initial state St { stInflight = inflightTxs, - stInflightSize = inflightTxsSize, stAcknowledged = Map.empty, - stInSubmissionToMempoolTxs = Map.keysSet inSubmissionToMempoolTxs } + stNewInSubmissionToMempoolTxs = Set.empty } >>> gn @@ -166,9 +165,8 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, ) accumFn st@St { stInflight, - stInflightSize, stAcknowledged, - stInSubmissionToMempoolTxs } + stNewInSubmissionToMempoolTxs } ( peeraddr , peerTxState@PeerTxState { availableTxIds, unknownTxs, @@ -177,156 +175,116 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, } ) = - let sizeInflightAll :: SizeInBytes - sizeInflightOther :: SizeInBytes - - sizeInflightAll = stInflightSize - sizeInflightOther = sizeInflightAll - requestedTxsInflightSize - - in if sizeInflightAll >= maxTxsSizeInflight - then let ( numTxIdsToAck - , numTxIdsToReq - , txsToMempool@TxsToMempool { listOfTxsToMempool } - , RefCountDiff { txIdsToAck } - , peerTxState' - ) = acknowledgeTxIds policy sharedState peerTxState - - stAcknowledged' = Map.unionWith (+) stAcknowledged txIdsToAck - stInSubmissionToMempoolTxs' = stInSubmissionToMempoolTxs - <> Set.fromList (map fst listOfTxsToMempool) - in - if requestedTxIdsInflight peerTxState' > 0 - then - -- we have txids to request - ( st { stAcknowledged = stAcknowledged' - , stInSubmissionToMempoolTxs = stInSubmissionToMempoolTxs' } - , ( (peeraddr, peerTxState') - , TxDecision { txdTxIdsToAcknowledge = numTxIdsToAck, - txdTxIdsToRequest = numTxIdsToReq, - txdPipelineTxIds = not - . StrictSeq.null - . unacknowledgedTxIds - $ peerTxState', - txdTxsToRequest = Map.empty, - txdTxsToMempool = txsToMempool - } - ) - ) - else - -- there are no `txid`s to request, nor we can request `tx`s due - -- to in-flight size limits - ( st - , ( (peeraddr, peerTxState') - , emptyTxDecision - ) - ) - else - let requestedTxsInflightSize' :: SizeInBytes - txsToRequestMap :: Map txid SizeInBytes - - (requestedTxsInflightSize', txsToRequestMap) = - -- inner fold: fold available `txid`s - -- - -- Note: although `Map.foldrWithKey` could be used here, it - -- does not allow to short circuit the fold, unlike - -- `foldWithState`. - foldWithState - (\(txid, (txSize, inflightMultiplicity)) sizeInflight -> - if -- note that we pick `txid`'s as long the `s` is - -- smaller or equal to `txsSizeInflightPerPeer`. - sizeInflight <= txsSizeInflightPerPeer - -- overall `tx`'s in-flight must be smaller than - -- `maxTxsSizeInflight` - && sizeInflight + sizeInflightOther <= maxTxsSizeInflight - -- the transaction must not be downloaded from more - -- than `txInflightMultiplicity` peers simultaneously - && inflightMultiplicity < txInflightMultiplicity - -- TODO: we must validate that `txSize` is smaller than - -- maximum txs size - then Just (sizeInflight + txSize, (txid, txSize)) - else Nothing - ) - (Map.assocs $ - -- merge `availableTxIds` with `stInflight`, so we don't - -- need to lookup into `stInflight` on every `txid` which - -- is in `availableTxIds`. - Map.merge (Map.mapMaybeMissing \_txid -> Just . (,0)) - Map.dropMissing - (Map.zipWithMatched \_txid -> (,)) - - availableTxIds - stInflight - -- remove `tx`s which were already downloaded by some - -- other peer or are in-flight or unknown by this peer. - `Map.withoutKeys` ( - Map.keysSet bufferedTxs - <> requestedTxsInflight - <> unknownTxs - <> stInSubmissionToMempoolTxs - ) - ) - requestedTxsInflightSize - -- pick from `txid`'s which are available from that given - -- peer. Since we are folding a dictionary each `txid` - -- will be selected only once from a given peer (at least - -- in each round). - - txsToRequest = Map.keysSet txsToRequestMap - peerTxState' = peerTxState { - requestedTxsInflightSize = requestedTxsInflightSize', - requestedTxsInflight = requestedTxsInflight - <> txsToRequest - } - - ( numTxIdsToAck - , numTxIdsToReq - , txsToMempool@TxsToMempool { listOfTxsToMempool } - , RefCountDiff { txIdsToAck } - , peerTxState'' - ) = acknowledgeTxIds policy sharedState peerTxState' - - stAcknowledged' = Map.unionWith (+) stAcknowledged txIdsToAck - - stInflightDelta :: Map txid Int - stInflightDelta = Map.fromSet (\_ -> 1) txsToRequest - -- note: this is right since every `txid` - -- could be picked at most once - - stInflight' :: Map txid Int - stInflight' = Map.unionWith (+) stInflightDelta stInflight - - stInSubmissionToMempoolTxs' = stInSubmissionToMempoolTxs - <> Set.fromList (map fst listOfTxsToMempool) - in - if requestedTxIdsInflight peerTxState'' > 0 - then - -- we can request `txid`s & `tx`s - ( St { stInflight = stInflight', - stInflightSize = sizeInflightOther + requestedTxsInflightSize', - stAcknowledged = stAcknowledged', - stInSubmissionToMempoolTxs = stInSubmissionToMempoolTxs' } - , ( (peeraddr, peerTxState'') - , TxDecision { txdTxIdsToAcknowledge = numTxIdsToAck, - txdPipelineTxIds = not - . StrictSeq.null - . unacknowledgedTxIds - $ peerTxState'', - txdTxIdsToRequest = numTxIdsToReq, - txdTxsToRequest = txsToRequestMap, - txdTxsToMempool = txsToMempool - } - ) - ) - else - -- there are no `txid`s to request, only `tx`s. - ( st { stInflight = stInflight', - stInflightSize = sizeInflightOther + requestedTxsInflightSize', - stInSubmissionToMempoolTxs = stInSubmissionToMempoolTxs' - } - , ( (peeraddr, peerTxState'') - , emptyTxDecision { txdTxsToRequest = txsToRequestMap } - ) - ) + let requestedTxsInflightSize' :: SizeInBytes + txsToRequestMap :: Map txid SizeInBytes + + blockedTxid :: txid -> Bool + blockedTxid txid = + Set.member txid requestedTxsInflight + || Set.member txid unknownTxs + || Map.member txid bufferedTxs + || Map.member txid inSubmissionToMempoolTxs + || Set.member txid stNewInSubmissionToMempoolTxs + + (requestedTxsInflightSize', txsToRequestMap) = + let candidates :: [(txid, (SizeInBytes, InFlightState))] + candidates = + Map.foldrWithKey + (\txid txSize acc -> + if blockedTxid txid + then acc + else (txid, (txSize, Map.findWithDefault mempty txid stInflight)) : acc) + [] + availableTxIds + in + -- inner fold: fold available `txid`s + -- + -- Note: although `Map.foldrWithKey` could be used here, it + -- does not allow to short circuit the fold, unlike + -- `foldWithState`. + foldWithState + (\(txid, (txSize, inflightSt)) sizeInflight -> + let inflightMultiplicity = inFlightCount inflightSt + inflightSpace = inFlightNextReq inflightSt + in + if -- note that we pick `txid`'s as long the `s` is + -- smaller or equal to `txsSizeInflightPerPeer`. + sizeInflight <= txsSizeInflightPerPeer + -- the transaction must not be downloaded from more + -- than `txInflightMultiplicity` peers simultaneously + && inflightMultiplicity < txInflightMultiplicity + -- we issue new requests for the same TX with + -- a minimum time between them + && inflightSpace <= now + -- TODO: we must validate that `txSize` is smaller than + -- maximum txs size + then Just (sizeInflight + txSize, (txid, txSize)) + else Nothing + ) + candidates + requestedTxsInflightSize + -- pick from `txid`'s which are available from that given + -- peer. Since we are folding a dictionary each `txid` + -- will be selected only once from a given peer (at least + -- in each round). + + txsToRequest = Map.keysSet txsToRequestMap + peerTxState' = peerTxState { + requestedTxsInflightSize = requestedTxsInflightSize', + requestedTxsInflight = requestedTxsInflight + <> txsToRequest + } + + ( numTxIdsToAck + , numTxIdsToReq + , txsToMempool@TxsToMempool { listOfTxsToMempool } + , RefCountDiff { txIdsToAck } + , peerTxState'' + ) = acknowledgeTxIds policy sharedState peerTxState' + + stAcknowledged' = Map.unionWith (+) stAcknowledged txIdsToAck + + stInflightDelta :: Map txid InFlightState + stInflightDelta = + Map.map (const $ InFlightState 1 $ addTime interTxSpace now) txsToRequestMap + -- note: this is right since every `txid` + -- could be picked at most once + + stInflight' :: Map txid InFlightState + stInflight' = Map.unionWith (<>) stInflightDelta stInflight + + stNewInSubmissionToMempoolTxs' = + List.foldl' (\acc (txid, _) -> Set.insert txid acc) + stNewInSubmissionToMempoolTxs + listOfTxsToMempool + in + if requestedTxIdsInflight peerTxState'' > 0 + then + -- we can request `txid`s & `tx`s + ( St { stInflight = stInflight', + stAcknowledged = stAcknowledged', + stNewInSubmissionToMempoolTxs = stNewInSubmissionToMempoolTxs' } + , ( (peeraddr, peerTxState'') + , TxDecision { txdTxIdsToAcknowledge = numTxIdsToAck, + txdPipelineTxIds = not + . StrictSeq.null + . unacknowledgedTxIds + $ peerTxState'', + txdTxIdsToRequest = numTxIdsToReq, + txdTxsToRequest = txsToRequestMap, + txdTxsToMempool = txsToMempool + } + ) + ) + else + -- there are no `txid`s to request, only `tx`s. + ( st { stInflight = stInflight', + stNewInSubmissionToMempoolTxs = stNewInSubmissionToMempoolTxs' + } + , ( (peeraddr, peerTxState'') + , emptyTxDecision { txdTxsToRequest = txsToRequestMap } + ) + ) gn :: ( St peeraddr txid tx , [((peeraddr, PeerTxState txid tx), TxDecision txid tx)] @@ -336,7 +294,6 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, ) gn ( St { stInflight, - stInflightSize, stAcknowledged } , as ) @@ -364,7 +321,6 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, in ( sharedState { peerTxStates = peerTxStates', inflightTxs = stInflight, - inflightTxsSize = stInflightSize, bufferedTxs = bufferedTxs', referenceCounts = referenceCounts', inSubmissionToMempoolTxs = inSubmissionToMempoolTxs'} @@ -402,48 +358,41 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer, -- filterActivePeers :: forall peeraddr txid tx. - Ord txid + ( Ord txid + , Ord peeraddr + ) => HasCallStack - => TxDecisionPolicy + => Time + -> TxDecisionPolicy -> SharedTxState peeraddr txid tx -> Map peeraddr (PeerTxState txid tx) filterActivePeers + now policy@TxDecisionPolicy { maxUnacknowledgedTxIds, txsSizeInflightPerPeer, - maxTxsSizeInflight, txInflightMultiplicity } sharedTxState@SharedTxState { peerTxStates, bufferedTxs, inflightTxs, - inflightTxsSize, - inSubmissionToMempoolTxs + inSubmissionToMempoolTxs, + pendingDecisions } - | inflightTxsSize > maxTxsSizeInflight - -- we might be able to request txids, we cannot download txs - = Map.filter fn peerTxStates - | otherwise - -- we might be able to request txids or txs. - = Map.filter gn peerTxStates + = Map.filterWithKey (\peer ps -> peer `Set.notMember` pendingDecisions && gn ps) + peerTxStates where - unrequestable = Map.keysSet (Map.filter (>= txInflightMultiplicity) inflightTxs) - <> Map.keysSet bufferedTxs - - fn :: PeerTxState txid tx -> Bool - fn peerTxState@PeerTxState { - requestedTxIdsInflight - } = - requestedTxIdsInflight == 0 - -- if a peer has txids in-flight, we cannot request more txids or txs. - && requestedTxIdsInflight + numOfUnacked <= maxUnacknowledgedTxIds - && txIdsToRequest > 0 - where - -- Split `unacknowledgedTxIds'` into the longest prefix of `txid`s which - -- can be acknowledged and the unacknowledged `txid`s. - (txIdsToRequest, _, unackedTxIds) = splitAcknowledgedTxIds policy sharedTxState peerTxState - numOfUnacked = fromIntegral (StrictSeq.length unackedTxIds) + unrequestableFilter :: InFlightState -> Bool + unrequestableFilter InFlightState{inFlightCount, inFlightNextReq} = + inFlightCount >= txInflightMultiplicity || inFlightNextReq > now + + isUnrequestable :: txid -> Bool + isUnrequestable txid = + Map.member txid bufferedTxs + || case Map.lookup txid inflightTxs of + Just inflightSt -> unrequestableFilter inflightSt + Nothing -> False gn :: PeerTxState txid tx -> Bool gn peerTxState@PeerTxState { unacknowledgedTxIds, @@ -457,15 +406,21 @@ filterActivePeers && requestedTxIdsInflight + numOfUnacked <= maxUnacknowledgedTxIds && txIdsToRequest > 0 ) - || (underSizeLimit && not (Map.null downloadable)) + || (underSizeLimit && downloadable) where numOfUnacked = fromIntegral (StrictSeq.length unacknowledgedTxIds) underSizeLimit = requestedTxsInflightSize <= txsSizeInflightPerPeer - downloadable = availableTxIds - `Map.withoutKeys` requestedTxsInflight - `Map.withoutKeys` unknownTxs - `Map.withoutKeys` unrequestable - `Map.withoutKeys` Map.keysSet inSubmissionToMempoolTxs + downloadable = + Map.foldrWithKey + (\txid _ acc -> + acc + || ( Set.notMember txid requestedTxsInflight + && Set.notMember txid unknownTxs + && not (isUnrequestable txid) + && Map.notMember txid inSubmissionToMempoolTxs + )) + False + availableTxIds -- Split `unacknowledgedTxIds'` into the longest prefix of `txid`s which -- can be acknowledged and the unacknowledged `txid`s. diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs index 2c03a6958f9..5dcf6eae33b 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs @@ -59,9 +59,12 @@ data TxDecisionPolicy = TxDecisionPolicy { scoreRate :: !Double, -- ^ rate at which "rejected" TXs drain. Unit: TX/seconds. - scoreMax :: !Double + scoreMax :: !Double, -- ^ Maximum number of "rejections". Unit: seconds + interTxSpace :: !DiffTime + -- ^ space between requests for the same TX. + } deriving Show @@ -71,12 +74,13 @@ instance NFData TxDecisionPolicy where defaultTxDecisionPolicy :: TxDecisionPolicy defaultTxDecisionPolicy = TxDecisionPolicy { - maxNumTxIdsToRequest = 3, + maxNumTxIdsToRequest = 10, maxUnacknowledgedTxIds = 10, -- must be the same as txSubmissionMaxUnacked txsSizeInflightPerPeer = max_TX_SIZE * 6, maxTxsSizeInflight = max_TX_SIZE * 20, txInflightMultiplicity = 2, bufferedTxsMinLifetime = 2, scoreRate = 0.1, - scoreMax = 15 * 60 + scoreMax = 15 * 60, + interTxSpace = 0.2 } diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs index 209fa129f1f..bc4ff75f628 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs @@ -18,8 +18,10 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Registry ) where import Control.Concurrent.Class.MonadMVar.Strict +import Control.Concurrent.Class.MonadSTM qualified as Lazy import Control.Concurrent.Class.MonadSTM.Strict import Control.Concurrent.Class.MonadSTM.TSem +import Control.Monad (when) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadThrow @@ -29,13 +31,16 @@ import Control.Tracer (Tracer, traceWith) import Data.Foldable as Foldable (foldl', traverse_) import Data.Hashable +import Data.List qualified as List import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Maybe (fromMaybe) import Data.Sequence.Strict (StrictSeq) import Data.Sequence.Strict qualified as StrictSeq +import Data.Set (Set) import Data.Set qualified as Set import Data.Typeable (Typeable) +import Data.Word (Word64) import Data.Void (Void) import Ouroboros.Network.Protocol.TxSubmission2.Type @@ -62,6 +67,9 @@ newtype TxMempoolSem m = TxMempoolSem (TSem m) newTxMempoolSem :: MonadSTM m => m (TxMempoolSem m) newTxMempoolSem = TxMempoolSem <$> atomically (newTSem 1) +bumpGeneration :: SharedTxState peeraddr txid tx -> SharedTxState peeraddr txid tx +bumpGeneration st@SharedTxState { generation } = st { generation = generation + 1 } + -- | API to access `PeerTxState` inside `PeerTxStateVar`. -- data PeerTxAPI m txid tx = PeerTxAPI { @@ -83,9 +91,10 @@ data PeerTxAPI m txid tx = PeerTxAPI { -> m (Maybe TxSubmissionProtocolError), -- ^ handle received txs - submitTxToMempool :: Tracer m (TraceTxSubmissionInbound txid tx) - -> txid -> tx -> m () - -- ^ submit the given (txid, tx) to the mempool. + submitTxsToMempool :: Tracer m (TraceTxSubmissionInbound txid tx) + -> [(txid, tx)] + -> m () + -- ^ submit the given txs to the mempool. } @@ -143,19 +152,24 @@ withPeer tracer txChannelMap return ( TxChannels { txChannelMap = txChannelMap' } - , PeerTxAPI { readTxDecision = takeMVar chann', + , PeerTxAPI { readTxDecision = do + d <- takeMVar chann' + atomically $ modifyTVar sharedStateVar $ \st -> + st { pendingDecisions = + Set.delete peeraddr (pendingDecisions st) } + return d, handleReceivedTxIds, handleReceivedTxs, - submitTxToMempool } + submitTxsToMempool } ) - atomically $ modifyTVar sharedStateVar registerPeer + atomically $ modifyTVar sharedStateVar (bumpGeneration . registerPeer) return peerTxAPI ) -- the handler is a short blocking operation, thus we need to use -- `uninterruptibleMask_` (\_ -> uninterruptibleMask_ do - atomically $ modifyTVar sharedStateVar unregisterPeer + atomically $ modifyTVar sharedStateVar (bumpGeneration . unregisterPeer) modifyMVar_ channelsVar \ TxChannels { txChannelMap } -> return TxChannels { txChannelMap = Map.delete peeraddr txChannelMap } @@ -190,18 +204,17 @@ withPeer tracer bufferedTxs, referenceCounts, inflightTxs, - inflightTxsSize, - inSubmissionToMempoolTxs } = + inSubmissionToMempoolTxs, + pendingDecisions } = st { peerTxStates = peerTxStates', bufferedTxs = bufferedTxs', referenceCounts = referenceCounts', inflightTxs = inflightTxs', - inflightTxsSize = inflightTxsSize', - inSubmissionToMempoolTxs = inSubmissionToMempoolTxs' } + inSubmissionToMempoolTxs = inSubmissionToMempoolTxs', + pendingDecisions = Set.delete peeraddr pendingDecisions } where (PeerTxState { unacknowledgedTxIds, requestedTxsInflight, - requestedTxsInflightSize, toMempoolTxs } , peerTxStates') = @@ -229,7 +242,6 @@ withPeer tracer liveSet inflightTxs' = Foldable.foldl' purgeInflightTxs inflightTxs requestedTxsInflight - inflightTxsSize' = inflightTxsSize - requestedTxsInflightSize -- When we unregister a peer, we need to subtract all txs in the -- `toMempoolTxs`, as they will not be submitted to the mempool. @@ -244,72 +256,83 @@ withPeer tracer purgeInflightTxs m txid = Map.alter fn txid m where - fn (Just n) | n > 1 = Just $! pred n - fn _ = Nothing + fn (Just a ) | inFlightCount a > 1 = Just $! a { inFlightCount = inFlightCount a - 1 } + fn _ = Nothing -- -- PeerTxAPI -- - submitTxToMempool :: Tracer m (TraceTxSubmissionInbound txid tx) -> txid -> tx -> m () - submitTxToMempool txTracer txid tx = + submitTxsToMempool :: Tracer m (TraceTxSubmissionInbound txid tx) + -> [(txid, tx)] + -> m () + submitTxsToMempool _ [] = return () + submitTxsToMempool txTracer txs = bracket_ (atomically $ waitTSem mempoolSem) (atomically $ signalTSem mempoolSem) $ do start <- getMonotonicTime - res <- addTx + mpSnapshot <- atomically mempoolGetSnapshot + + -- Note that checking if the mempool contains a TX before + -- spending several ms attempting to add it to the pool has + -- been judged immoral. + let toSubmit = + [ (txid, tx) + | (txid, tx) <- txs + , not (mempoolHasTx mpSnapshot txid) + ] + toSubmitTxs = map snd toSubmit + + (acceptedTxIds, _) <- + if null toSubmitTxs + then return ([], []) + else mempoolAddTxs toSubmitTxs + end <- getMonotonicTime - atomically $ modifyTVar sharedStateVar (updateBufferedTx end res) let duration = end `diffTime` start - case res of - TxAccepted -> traceWith txTracer (TraceTxInboundAddedToMempool [txid] duration) - TxRejected -> traceWith txTracer (TraceTxInboundRejectedFromMempool [txid] duration) + acceptedSet = Set.fromList acceptedTxIds + isAccepted txid = txid `Set.member` acceptedSet + rejectedTxIds = [ txid | (txid, _) <- txs, not (isAccepted txid) ] + acceptedCount = length acceptedTxIds + rejectedCount = length rejectedTxIds - where - -- add the tx to the mempool - addTx :: m TxMempoolResult - addTx = do - mpSnapshot <- atomically mempoolGetSnapshot - - -- Note that checking if the mempool contains a TX before - -- spending several ms attempting to add it to the pool has - -- been judged immoral. - if mempoolHasTx mpSnapshot txid - then do - !now <- getMonotonicTime - !s <- countRejectedTxs now 1 - traceWith txTracer $ TraceTxSubmissionProcessed ProcessedTxCount { - ptxcAccepted = 0 - , ptxcRejected = 1 - , ptxcScore = s - } - return TxRejected - else do - (acceptedTxs, _) <- mempoolAddTxs [tx] - end <- getMonotonicTime - case acceptedTxs of - [] -> do - !s <- countRejectedTxs end 1 - traceWith txTracer $ TraceTxSubmissionProcessed ProcessedTxCount { - ptxcAccepted = 0 - , ptxcRejected = 1 - , ptxcScore = s - } - return TxRejected - (_:_) -> do - !s <- countRejectedTxs end 0 - traceWith txTracer $ TraceTxSubmissionProcessed ProcessedTxCount { - ptxcAccepted = 1 - , ptxcRejected = 0 - , ptxcScore = s - } - return TxAccepted + !s <- atomically $ do + modifyTVar sharedStateVar $ \st -> + bumpGeneration $ Foldable.foldl' (updateBufferedTx end acceptedSet) st txs + countRejectedTxs end (fromIntegral rejectedCount) + + traceWith txTracer $ TraceTxSubmissionProcessed ProcessedTxCount { + ptxcAccepted = acceptedCount + , ptxcRejected = rejectedCount + , ptxcScore = s + } + + when (not $ List.null acceptedTxIds) $ + traceWith txTracer (TraceTxInboundAddedToMempool acceptedTxIds duration) + when (not $ List.null rejectedTxIds) $ + traceWith txTracer (TraceTxInboundRejectedFromMempool rejectedTxIds duration) + + where updateBufferedTx :: Time - -> TxMempoolResult + -> Set txid -> SharedTxState peeraddr txid tx + -> (txid, tx) -> SharedTxState peeraddr txid tx - updateBufferedTx _ TxRejected st@SharedTxState { peerTxStates + updateBufferedTx now acceptedSet st (txid, tx) = + let res = if txid `Set.member` acceptedSet + then TxAccepted + else TxRejected + in updateBufferedTxResult now res txid tx st + + updateBufferedTxResult :: Time + -> TxMempoolResult + -> txid + -> tx + -> SharedTxState peeraddr txid tx + -> SharedTxState peeraddr txid tx + updateBufferedTxResult _ TxRejected txid _ st@SharedTxState { peerTxStates , inSubmissionToMempoolTxs } = st { peerTxStates = peerTxStates' , inSubmissionToMempoolTxs = inSubmissionToMempoolTxs' } @@ -322,7 +345,7 @@ withPeer tracer where fn ps = Just $! ps { toMempoolTxs = Map.delete txid (toMempoolTxs ps)} - updateBufferedTx now TxAccepted + updateBufferedTxResult now TxAccepted txid tx st@SharedTxState { peerTxStates , bufferedTxs , referenceCounts @@ -385,12 +408,12 @@ withPeer tracer -- PRECONDITION: the `Double` argument is non-negative. countRejectedTxs :: Time -> Double - -> m Double + -> STM m Double countRejectedTxs _ n | n < 0 = error ("TxSubmission.countRejectedTxs: invariant violation for peer " ++ show peeraddr) - countRejectedTxs now n = atomically $ stateTVar sharedStateVar $ \st -> - let (result, peerTxStates') = Map.alterF fn peeraddr (peerTxStates st) - in (result, st { peerTxStates = peerTxStates' }) + countRejectedTxs now n = stateTVar sharedStateVar $ \st -> + let (result, peerTxStates') = Map.alterF fn peeraddr (peerTxStates st) in + (result, st { peerTxStates = peerTxStates' }) where fn :: Maybe (PeerTxState txid tx) -> (Double, Maybe (PeerTxState txid tx)) fn Nothing = error ("TxSubmission.withPeer: invariant violation for peer " ++ show peeraddr) @@ -423,10 +446,11 @@ drainRejectionThread , Ord txid ) => Tracer m (TraceTxLogic peeraddr txid tx) + -> Tracer m TxSubmissionCounters -> TxDecisionPolicy -> SharedTxStateVar m peeraddr txid tx -> m Void -drainRejectionThread tracer policy sharedStateVar = do +drainRejectionThread tracer counterTracer policy sharedStateVar = do labelThisThread "tx-rejection-drain" now <- getMonotonicTime go $ addTime drainInterval now @@ -439,68 +463,158 @@ drainRejectionThread tracer policy sharedStateVar = do threadDelay 1 !now <- getMonotonicTime - st'' <- atomically $ do + st''' <- atomically $ do st <- readTVar sharedStateVar let ptss = if now > nextDrain then Map.map (updateRejects policy now 0) (peerTxStates st) else peerTxStates st st' = tickTimedTxs now st { peerTxStates = ptss } - writeTVar sharedStateVar st' - return st' - traceWith tracer (TraceSharedTxState "drainRejectionThread" st'') + st'' = st' { inflightTxs = Map.filter (filterStaleReq now) (inflightTxs st')} + writeTVar sharedStateVar (bumpGeneration st'') + return st'' + traceWith tracer (TraceSharedTxState "drainRejectionThread" st''') + traceWith counterTracer (mkTxSubmissionCounters st''') if now > nextDrain then go $ addTime drainInterval now else go nextDrain + filterStaleReq :: Time -> InFlightState -> Bool + filterStaleReq now e = inFlightCount e > 0 || inFlightNextReq e > now + decisionLogicThread :: forall m peeraddr txid tx. ( MonadDelay m , MonadMVar m - , MonadSTM m , MonadMask m , MonadFork m + , MonadTimer m , Ord peeraddr , Ord txid , Hashable peeraddr ) => Tracer m (TraceTxLogic peeraddr txid tx) - -> Tracer m TxSubmissionCounters -> TxDecisionPolicy -> TxChannelsVar m peeraddr txid tx -> SharedTxStateVar m peeraddr txid tx -> m Void -decisionLogicThread tracer counterTracer policy txChannelsVar sharedStateVar = do +decisionLogicThread tracer policy txChannelsVar sharedStateVar = do labelThisThread "tx-decision" - go + initialGeneration <- atomically $ generation <$> readTVar sharedStateVar + go initialGeneration where - go :: m Void - go = do - -- We rate limit the decision making process, it could overwhelm the CPU - -- if there are too many inbound connections. - threadDelay _DECISION_LOOP_DELAY - - (decisions, st) <- atomically do + maxCoalesce :: DiffTime + maxCoalesce = 0.050 + + debounceTime :: DiffTime + debounceTime = 0.01 + + debounce :: Lazy.TVar m Bool -> Time -> Word64 -> m () + debounce txTimer deadline lastGen = do + now <- getMonotonicTime + let remaining = diffTime deadline now + if remaining <= 0 + then return () + else do + let waitFor = min debounceTime remaining + debounceTimer <- registerDelay waitFor + res <- atomically $ do + st <- readTVar sharedStateVar + expiredTx <- Lazy.readTVar txTimer + expiredDebounce <- Lazy.readTVar debounceTimer + let gen = generation st + case (expiredTx, gen > lastGen, expiredDebounce) of + (True , _, _) -> return $ Right gen -- txTimer wins, stop coalescing + (False, True, _) -> return $ Left gen -- new change, restart debounce + (False, False, True) -> return $ Right gen -- queit period elapsed + (False, False, False) -> retry + case res of + Left newGen -> debounce txTimer deadline newGen + Right _ -> return () + + go :: Word64 -> m Void + go lastGen = do + now <- getMonotonicTime + nextDelay <- atomically $ do + sharedTxState <- readTVar sharedStateVar + return $ nextDecisionDelay now sharedTxState + delayVar <- registerDelay nextDelay + -- Wait until there is potentially some work to do or the timer expires. + wake_res <- atomically do + sharedTxState <- readTVar sharedStateVar + let gen = generation sharedTxState + timerExpired <- Lazy.readTVar delayVar + + -- block until state changes or the timer expires + if timerExpired + then return Nothing + else if gen > lastGen + then return $ Just gen + else retry + + -- Unless a tx timer expired run debouncer + -- in order to coalesce events together. + case wake_res of + Nothing -> return () + Just gen -> do + now' <- getMonotonicTime + let deadline = addTime maxCoalesce now' + debounce delayVar deadline gen + + -- Use a fresh timestamp for decisions + now' <- getMonotonicTime + res_m <- atomically do sharedTxState <- readTVar sharedStateVar - let activePeers = filterActivePeers policy sharedTxState - - -- block until at least one peer is active - check (not (Map.null activePeers)) - - let (sharedState, decisions) = makeDecisions policy sharedTxState activePeers - writeTVar sharedStateVar sharedState - return (decisions, sharedState) - traceWith tracer (TraceSharedTxState "decisionLogicThread" st) - traceWith tracer (TraceTxDecisions decisions) - TxChannels { txChannelMap } <- readMVar txChannelsVar - traverse_ - (\(mvar, d) -> modifyMVarWithDefault_ mvar d (\d' -> pure (d' <> d))) - (Map.intersectionWith (,) - txChannelMap - decisions) - traceWith counterTracer (mkTxSubmissionCounters st) - go + let activePeers = filterActivePeers now' policy sharedTxState + + if Map.null activePeers + then return (Left (generation sharedTxState)) + else do + let (sharedState, decisions) = makeDecisions now' policy sharedTxState activePeers + writeTVar sharedStateVar sharedState + return $ Right (decisions, sharedState) + + case res_m of + Left gen -> go gen + Right (decisions, st) -> do + traceWith tracer (TraceSharedTxState "decisionLogicThread" st) + traceWith tracer (TraceTxDecisions decisions) + TxChannels { txChannelMap } <- readMVar txChannelsVar + let deliverable = Map.intersection txChannelMap decisions + atomically $ modifyTVar sharedStateVar $ \st' -> + st' { pendingDecisions = pendingDecisions st' <> Map.keysSet deliverable } + traverse_ + (\(mvar, d) -> modifyMVarWithDefault_ mvar d (\d' -> pure (d' <> d))) + (Map.intersectionWith (,) + txChannelMap + decisions) + go (generation st) + + nextDecisionDelay + :: Time + -> SharedTxState peeraddr txid tx + -> DiffTime + nextDecisionDelay now SharedTxState { inflightTxs } = + fromMaybe maxDelay (diffTimeNow <$> nextWake) + where + -- If there are no outstanding TXs we wait for a long time + -- or until an STM value changes. + maxDelay :: DiffTime + maxDelay = 120 + + nextWake :: Maybe Time + nextWake = + Foldable.foldl' step Nothing inflightTxs + + step :: Maybe Time -> InFlightState -> Maybe Time + step acc InFlightState { inFlightNextReq } = + if inFlightNextReq <= now + then acc + else Just $ maybe inFlightNextReq (min inFlightNextReq) acc + + diffTimeNow :: Time -> DiffTime + diffTimeNow t = t `diffTime` now -- Variant of modifyMVar_ that puts a default value if the MVar is empty. modifyMVarWithDefault_ :: StrictMVar m a -> a -> (a -> m a) -> m () @@ -523,6 +637,7 @@ decisionLogicThreads , MonadMask m , MonadAsync m , MonadFork m + , MonadTimer m , Ord peeraddr , Ord txid , Hashable peeraddr @@ -535,11 +650,9 @@ decisionLogicThreads -> m Void decisionLogicThreads tracer counterTracer policy txChannelsVar sharedStateVar = uncurry (<>) <$> - drainRejectionThread tracer policy sharedStateVar + drainRejectionThread tracer counterTracer policy sharedStateVar `concurrently` - decisionLogicThread tracer counterTracer policy txChannelsVar sharedStateVar + decisionLogicThread tracer policy txChannelsVar sharedStateVar --- `5ms` delay -_DECISION_LOOP_DELAY :: DiffTime -_DECISION_LOOP_DELAY = 0.005 +-- No fixed delay: decision loop blocks on STM/timers and uses debounce. diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs index 672db8a911e..0258a0d4f54 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE BangPatterns #-} {-# LANGUAGE BlockArguments #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} @@ -34,7 +35,7 @@ import Data.Functor (($>)) import Data.Map.Merge.Strict qualified as Map import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map -import Data.Maybe (fromJust, maybeToList) +import Data.Maybe (fromJust) import Data.Sequence.Strict (StrictSeq) import Data.Sequence.Strict qualified as StrictSeq import Data.Set qualified as Set @@ -72,7 +73,7 @@ acknowledgeTxIds acknowledgeTxIds policy - sharedTxState + sharedTxState@SharedTxState { bufferedTxs } ps@PeerTxState { availableTxIds, unknownTxs, requestedTxIdsInflight, @@ -111,30 +112,25 @@ acknowledgeTxIds (txIdsToRequest, acknowledgedTxIds, unacknowledgedTxIds') = splitAcknowledgedTxIds policy sharedTxState ps - txsToMempool = [ (txid, tx) - | txid <- toList toMempoolTxIds - , txid `Map.notMember` bufferedTxs sharedTxState - , tx <- maybeToList $ txid `Map.lookup` downloadedTxs - ] - -- Select downloaded txs from the prefix of `acknowledgedTxIds`, ignoring - -- unknown and buffered txs. - toMempoolTxIds = - StrictSeq.filter (`Map.member` downloadedTxs) acknowledgedTxIds - - txsToMempoolMap = Map.fromList txsToMempool + -- foldl' reverses the order, so we need to undo it here + txsToMempool = reverse txsToMempoolRev toMempoolTxs' = toMempoolTxs <> txsToMempoolMap - (downloadedTxs', ackedDownloadedTxs) = Map.partitionWithKey (\txid _ -> txid `Set.member` liveSet) downloadedTxs - -- latexTxs: transactions which were downloaded by another peer before we - -- downloaded them; it relies on that `txToMempool` filters out - -- `bufferedTxs`. - lateTxs = Map.filterWithKey (\txid _ -> txid `Map.notMember` txsToMempoolMap) ackedDownloadedTxs - score' = score + fromIntegral (Map.size lateTxs) + (txsToMempoolRev, txsToMempoolMap, refCountDiffMap) = + Foldable.foldl' accAcknowledgedTxId ([], Map.empty, Map.empty) acknowledgedTxIds -- the set of live `txids` liveSet = Set.fromList (toList unacknowledgedTxIds') + (downloadedTxs', lateTxCount) = + Map.foldlWithKey' keepLiveDownloadedTx (Map.empty, 0) downloadedTxs + + -- lateTxCount: transactions which were downloaded by another peer before we + -- downloaded them; it relies on that `txToMempool` filters out + -- `bufferedTxs`. + score' = score + fromIntegral lateTxCount + availableTxIds' = availableTxIds `Map.restrictKeys` liveSet @@ -145,13 +141,44 @@ acknowledgeTxIds -- above). unknownTxs' = unknownTxs `Set.intersection` liveSet - refCountDiff = RefCountDiff - $ foldr (Map.alter fn) - Map.empty acknowledgedTxIds - where - fn :: Maybe Int -> Maybe Int - fn Nothing = Just 1 - fn (Just n) = Just $! n + 1 + refCountDiff = RefCountDiff refCountDiffMap + + bumpRefCount :: Maybe Int -> Maybe Int + bumpRefCount Nothing = Just 1 + bumpRefCount (Just n) = Just $! n + 1 + + -- Fold one acknowledged txid into the mempool queue and refcount diff. + accAcknowledgedTxId + :: ([(txid, tx)], Map txid tx, Map txid Int) + -> txid + -> ([(txid, tx)], Map txid tx, Map txid Int) + accAcknowledgedTxId (!txsRev, !txsMap, !refDiffMap) txid = + let !refDiffMap' = Map.alter bumpRefCount txid refDiffMap + in case Map.lookup txid downloadedTxs of + Just tx + | txid `Map.notMember` bufferedTxs + -- without the guard below we could potentially enqueue + -- the same tx into the mempool multiple times over + -- several decision loop iterations before the tx + -- is finally in the mempool, or rejected. + , txid `Map.notMember` toMempoolTxs + -> ( (txid, tx) : txsRev + , Map.insert txid tx txsMap + , refDiffMap' + ) + _ -> (txsRev, txsMap, refDiffMap') + + -- Rebuild the live downloaded set and count as late only non-live txs that + -- are already buffered globally. + keepLiveDownloadedTx + :: (Map txid tx, Int) + -> txid + -> tx + -> (Map txid tx, Int) + keepLiveDownloadedTx (!downloadedTxsAcc, !lateTxCountAcc) txid tx + | txid `Set.member` liveSet = (Map.insert txid tx downloadedTxsAcc, lateTxCountAcc) + | txid `Map.member` bufferedTxs = (downloadedTxsAcc, lateTxCountAcc + 1) + | otherwise = (downloadedTxsAcc, lateTxCountAcc) txIdsToAcknowledge :: NumTxIdsToAck txIdsToAcknowledge = fromIntegral $ StrictSeq.length acknowledgedTxIds @@ -320,37 +347,18 @@ receivedTxIdsImpl unacknowledgedTxIds } = (st', ps') where + (availableTxIds', bufferedTxs') = + Map.foldlWithKey' accumulateReceivedTxId + (availableTxIds, bufferedTxs) + txidsMap + -- -- Handle new `txid`s -- - -- Divide the new txids in two: those that are already in the mempool - -- and those that are not. We'll request some txs from the latter. - (ignoredTxIds, availableTxIdsMap) = - Map.partitionWithKey - (\txid _ -> mempoolHasTx txid) - txidsMap - - -- Add all `txids` from `availableTxIdsMap` which are not - -- unacknowledged or already buffered. Unacknowledged txids must have - -- already been added to `availableTxIds` map before. - availableTxIds' = - Map.foldlWithKey - (\m txid sizeInBytes -> Map.insert txid sizeInBytes m) - availableTxIds - (Map.filterWithKey - (\txid _ -> txid `notElem` unacknowledgedTxIds - && txid `Map.notMember` bufferedTxs) - availableTxIdsMap) - -- Add received txids to `unacknowledgedTxIds`. unacknowledgedTxIds' = unacknowledgedTxIds <> txidsSeq - -- Add ignored `txs` to buffered ones. - -- Note: we prefer to keep the `tx` if it's already in `bufferedTxs`. - bufferedTxs' = bufferedTxs - <> Map.map (const Nothing) ignoredTxIds - referenceCounts' = Foldable.foldl' (flip $ Map.alter (\case @@ -366,6 +374,26 @@ receivedTxIdsImpl unacknowledgedTxIds = unacknowledgedTxIds', requestedTxIdsInflight = requestedTxIdsInflight - reqNo } + -- Fold one received txid into the available and buffered tx maps. + accumulateReceivedTxId + :: (Map txid SizeInBytes, Map txid (Maybe tx)) + -> txid + -> SizeInBytes + -> (Map txid SizeInBytes, Map txid (Maybe tx)) + accumulateReceivedTxId (!availableTxIdsAcc, !bufferedTxsAcc) txid sizeInBytes + | mempoolHasTx txid + = (availableTxIdsAcc, Map.alter keepBufferedTx txid bufferedTxsAcc) + | txid `elem` unacknowledgedTxIds || txid `Map.member` bufferedTxs + = (availableTxIdsAcc, bufferedTxsAcc) + | otherwise + = (Map.insert txid sizeInBytes availableTxIdsAcc, bufferedTxsAcc) + + -- Insert a placeholder only if absent; never overwrite a buffered tx body. + keepBufferedTx :: Maybe (Maybe tx) + -> Maybe (Maybe tx) + keepBufferedTx Nothing = Just Nothing + keepBufferedTx existing = existing + -- | We check advertised sizes up in a fuzzy way. The advertised and received -- sizes need to agree up to `const_MAX_TX_SIZE_DISCREPANCY`. -- @@ -459,20 +487,14 @@ collectTxsImpl txSize peeraddr requestedTxIdsMap receivedTxs Map.merge (Map.mapMaybeMissing \_ x -> Just x) (Map.mapMaybeMissing \_ _ -> assert False Nothing) - (Map.zipWithMaybeMatched \_ x y -> assert (x >= y) - let z = x - y in - if z > 0 - then Just z - else Nothing) + (Map.zipWithMaybeMatched \_ x y -> + assert (inFlightCount x >= y) + let cnt' = inFlightCount x - y in + Just $ x { inFlightCount = cnt' }) (inflightTxs st) (Map.fromSet (const 1) requestedTxIds) - inflightTxsSize'' = assert (inflightTxsSize st >= requestedSize) $ - inflightTxsSize st - requestedSize - - st' = st { inflightTxs = inflightTxs'', - inflightTxsSize = inflightTxsSize'' - } + st' = st { inflightTxs = inflightTxs'' } -- -- Update PeerTxState @@ -511,18 +533,22 @@ collectTxsImpl txSize peeraddr requestedTxIdsMap receivedTxs type SharedTxStateVar m peeraddr txid tx = StrictTVar m (SharedTxState peeraddr txid tx) +bumpGeneration :: SharedTxState peeraddr txid tx -> SharedTxState peeraddr txid tx +bumpGeneration st@SharedTxState { generation } = st { generation = generation + 1 } + newSharedTxStateVar :: MonadSTM m => StdGen -> m (SharedTxStateVar m peeraddr txid tx) newSharedTxStateVar rng = newTVarIO SharedTxState { peerTxStates = Map.empty, inflightTxs = Map.empty, - inflightTxsSize = 0, bufferedTxs = Map.empty, referenceCounts = Map.empty, timedTxs = Map.empty, inSubmissionToMempoolTxs = Map.empty, - peerRng = rng + pendingDecisions = Set.empty, + peerRng = rng, + generation = 0 } @@ -547,7 +573,10 @@ receivedTxIds receivedTxIds tracer sharedVar getMempoolSnapshot peeraddr reqNo txidsSeq txidsMap = do st <- atomically $ do MempoolSnapshot{mempoolHasTx} <- getMempoolSnapshot - stateTVar sharedVar ((\a -> (a,a)) . receivedTxIdsImpl mempoolHasTx peeraddr reqNo txidsSeq txidsMap) + stateTVar sharedVar (\st -> + let st' = receivedTxIdsImpl mempoolHasTx peeraddr reqNo txidsSeq txidsMap st + st'' = bumpGeneration st' + in (st'', st'')) traceWith tracer (TraceSharedTxState "receivedTxIds" st) @@ -571,7 +600,7 @@ collectTxs tracer txSize sharedVar peeraddr txidsRequested txsMap = do r <- atomically $ do st <- readTVar sharedVar case collectTxsImpl txSize peeraddr txidsRequested txsMap st of - r@(Right st') -> writeTVar sharedVar st' + r@(Right st') -> writeTVar sharedVar (bumpGeneration st') $> r r@Left {} -> pure r case r of diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs index 68ecb40462f..18e00c75cd4 100644 --- a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs +++ b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs @@ -15,6 +15,8 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Types PeerTxState (..) -- * SharedTxState , SharedTxState (..) + -- * InFlightState + , InFlightState (..) -- * Decisions , TxsToMempool (..) , TxDecision (..) @@ -37,7 +39,7 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Types ) where import Control.DeepSeq -import Control.Exception (Exception (..)) +import Control.Exception (Exception (..), assert) import Control.Monad.Class.MonadTime.SI import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map @@ -46,6 +48,7 @@ import Data.Sequence.Strict (StrictSeq) import Data.Set (Set) import Data.Set qualified as Set import Data.Typeable (Typeable, eqT, (:~:) (Refl)) +import Data.Word (Word64) import GHC.Generics (Generic) import System.Random (StdGen) @@ -140,6 +143,21 @@ instance ( NoThunks txid , NoThunks tx ) => NoThunks (PeerTxState txid tx) +data InFlightState = InFlightState { + inFlightCount :: !Int + , inFlightNextReq :: !Time + } + deriving (Eq, Ord, Show, Generic, NFData) + +instance Semigroup InFlightState where + (<>) (InFlightState c0 t0) (InFlightState c1 t1) = InFlightState (c0 + c1) (max t0 t1) + +instance Monoid InFlightState where + mempty = InFlightState 0 (Time 0) + mappend = (<>) + +instance NoThunks InFlightState + -- | Shared state of all `TxSubmission` clients. -- @@ -176,11 +194,7 @@ data SharedTxState peeraddr txid tx = SharedTxState { -- -- This set can intersect with `availableTxIds`. -- - inflightTxs :: !(Map txid Int), - - -- | Overall size of all `tx`s in-flight. - -- - inflightTxsSize :: !SizeInBytes, + inflightTxs :: !(Map txid InFlightState), -- | Map of `tx` which: -- @@ -241,6 +255,16 @@ data SharedTxState peeraddr txid tx = SharedTxState { -- inSubmissionToMempoolTxs :: !(Map txid Int), + -- | Peers with a pending decision in their channel. + -- + -- We use this to avoid recomputing decisions for peers that haven't yet + -- consumed their previous decision (e.g. blocked on mempool submission). + pendingDecisions :: !(Set peeraddr), + + -- | Monotonic generation bumped on external state changes. + -- Used by decision logic to sleep until work is available. + generation :: !Word64, + -- | Rng used to randomly order peers peerRng :: !StdGen } @@ -318,7 +342,11 @@ instance Ord txid => Semigroup (TxDecision txid tx) where txdTxIdsToRequest = txdTxIdsToRequest + txdTxIdsToRequest', txdPipelineTxIds = txdPipelineTxIds', txdTxsToRequest = txdTxsToRequest <> txdTxsToRequest', - txdTxsToMempool = txdTxsToMempool <> txdTxsToMempool' + txdTxsToMempool = + let left = Set.fromList . fmap fst $ listOfTxsToMempool txdTxsToMempool + right = Set.fromList . fmap fst $ listOfTxsToMempool txdTxsToMempool' + shared = Set.intersection left right + in assert (Set.null shared) $ txdTxsToMempool <> txdTxsToMempool' } -- | A no-op decision. @@ -431,7 +459,7 @@ mkTxSubmissionCounters Set.\\ Map.keysSet inSubmissionToMempoolTxs, numOfBufferedTxs = Map.size bufferedTxs, numOfInSubmissionToMempoolTxs = Map.size inSubmissionToMempoolTxs, - numOfTxIdsInflight = getSum $ foldMap Sum inflightTxs + numOfTxIdsInflight = getSum $ foldMap (Sum . inFlightCount) inflightTxs } diff --git a/ouroboros-network/ouroboros-network.cabal b/ouroboros-network/ouroboros-network.cabal index a3c84c2d2fa..78dc824ed55 100644 --- a/ouroboros-network/ouroboros-network.cabal +++ b/ouroboros-network/ouroboros-network.cabal @@ -31,6 +31,11 @@ flag nightly manual: False default: False +flag txlogic-print + description: Enable TxLogic benchmark context printing + manual: True + default: False + source-repository head type: git location: https://github.com/intersectmbo/ouroboros-network @@ -1052,11 +1057,17 @@ benchmark sim-benchmarks main-is: Main.hs build-depends: base, + containers, deepseq, + io-classes:{io-classes, si-timers}, ouroboros-network:{ouroboros-network, ouroboros-network-tests-lib}, + pretty-simple, + random, splitmix, tasty-bench >=0.3.5, + if flag(txlogic-print) + cpp-options: -DTXLOGIC_PRINT -- We use `-fproc-alignemtn` option to avoid skewed results due to changes in cache-line -- alignment. See https://github.com/Bodigrim/tasty-bench#comparison-against-baseline -- We use threaded RTS, because of @@ -1065,8 +1076,5 @@ benchmark sim-benchmarks -fno-ignore-asserts -threaded -rtsopts - -with-rtsopts=-A32m + "-with-rtsopts=-A32m -T" -fproc-alignment=64 - +RTS - -T - -RTS diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToClient/Version.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToClient/Version.hs deleted file mode 100644 index dc5ce6832a3..00000000000 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToClient/Version.hs +++ /dev/null @@ -1,37 +0,0 @@ -{-# LANGUAGE NamedFieldPuns #-} -{-# OPTIONS_GHC -Wno-orphans #-} - -module Test.Ouroboros.Network.NodeToClient.Version (tests) where - -import Cardano.Network.NodeToClient.Version - -import Ouroboros.Network.CodecCBORTerm -import Ouroboros.Network.Magic - -import Test.QuickCheck -import Test.Tasty (TestTree, testGroup) -import Test.Tasty.QuickCheck (testProperty) - - -tests :: TestTree -tests = testGroup "Ouroboros.Network.NodeToClient.Version" - [ testProperty "nodeToClientCodecCBORTerm" prop_nodeToClientCodec - ] - -data VersionAndVersionData = - VersionAndVersionData NodeToClientVersion NodeToClientVersionData - deriving Show - -instance Arbitrary VersionAndVersionData where - arbitrary = - VersionAndVersionData - <$> elements [ minBound .. maxBound] - <*> (NodeToClientVersionData . NetworkMagic <$> arbitrary <*> arbitrary) - -prop_nodeToClientCodec :: VersionAndVersionData -> Bool -prop_nodeToClientCodec (VersionAndVersionData vNumber vData) = - case decodeTerm (encodeTerm vData) of - Right vData' -> networkMagic vData' == networkMagic vData - Left {} -> False - where - CodecCBORTerm { encodeTerm, decodeTerm } = nodeToClientCodecCBORTerm vNumber diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToNode/Version.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToNode/Version.hs deleted file mode 100644 index 919ca345b2d..00000000000 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/NodeToNode/Version.hs +++ /dev/null @@ -1,48 +0,0 @@ -{-# LANGUAGE NamedFieldPuns #-} -{-# OPTIONS_GHC -Wno-orphans #-} - -module Test.Ouroboros.Network.NodeToNode.Version (tests) where - -import Ouroboros.Network.CodecCBORTerm -import Ouroboros.Network.Magic - -import Cardano.Network.NodeToNode.Version - -import Ouroboros.Network.PeerSelection.PeerSharing (PeerSharing (..)) -import Test.QuickCheck -import Test.Tasty (TestTree, testGroup) -import Test.Tasty.QuickCheck (testProperty) - - -tests :: TestTree -tests = testGroup "Ouroboros.Network.NodeToNode.Version" - [ testProperty "nodeToNodeCodecCBORTerm" prop_nodeToNodeCodec - ] - -instance Arbitrary NodeToNodeVersion where - arbitrary = arbitraryBoundedEnum - - shrink v - | v == minBound = [] - | otherwise = [pred v] - -instance Arbitrary NodeToNodeVersionData where - arbitrary = - NodeToNodeVersionData - <$> (NetworkMagic <$> arbitrary) - <*> oneof [ pure InitiatorOnlyDiffusionMode - , pure InitiatorAndResponderDiffusionMode - ] - <*> elements [ PeerSharingDisabled - , PeerSharingEnabled - ] - <*> arbitrary - -prop_nodeToNodeCodec :: NodeToNodeVersion -> NodeToNodeVersionData -> Bool -prop_nodeToNodeCodec ntnVersion ntnData = - case decodeTerm (encodeTerm ntnData) of - Right ntnData' -> networkMagic ntnData' == networkMagic ntnData - && diffusionMode ntnData' == diffusionMode ntnData - Left {} -> False - where - CodecCBORTerm { encodeTerm, decodeTerm } = nodeToNodeCodecCBORTerm ntnVersion diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV1.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV1.hs index 15a28691603..f751101343e 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV1.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV1.hs @@ -94,6 +94,7 @@ txSubmissionSimulation tracer maxUnacked outboundTxs inboundDelay outboundDelay = do inboundMempool <- emptyMempool + duplicateTxIdsVar <- newTVarIO [] outboundMempool <- newMempool outboundTxs (outboundChannel, inboundChannel) <- createConnectedChannels outboundAsync <- @@ -112,7 +113,7 @@ txSubmissionSimulation tracer maxUnacked outboundTxs (byteLimitsTxSubmission2 (fromIntegral . BSL.length)) timeLimitsTxSubmission2 (maybe id delayChannel inboundDelay inboundChannel) - (txSubmissionServerPeerPipelined (inboundPeer inboundMempool)) + (txSubmissionServerPeerPipelined (inboundPeer duplicateTxIdsVar inboundMempool)) _ <- waitAnyCancel [ outboundAsync, inboundAsync ] @@ -130,14 +131,16 @@ txSubmissionSimulation tracer maxUnacked outboundTxs (maxBound :: TestVersion) controlMessageSTM - inboundPeer :: Mempool m txid (Tx txid) -> TxSubmissionServerPipelined txid (Tx txid) m () - inboundPeer inboundMempool = + inboundPeer :: TVar m [txid] + -> Mempool m txid (Tx txid) + -> TxSubmissionServerPipelined txid (Tx txid) m () + inboundPeer duplicateTxIdsVar inboundMempool = txSubmissionInbound (("INBOUND",) `contramap` verboseTracer) NoTxSubmissionInitDelay maxUnacked (getMempoolReader inboundMempool) - (getMempoolWriter inboundMempool) + (getMempoolWriter duplicateTxIdsVar inboundMempool) (maxBound :: TestVersion) prop_txSubmission :: Positive Word16 diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV2.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV2.hs index ecee359ccc3..ec8698802ce 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV2.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/AppV2.hs @@ -20,6 +20,7 @@ import Prelude hiding (seq) import NoThunks.Class import Control.Concurrent.Class.MonadMVar.Strict +import Control.Concurrent.Class.MonadSTM qualified as Lazy import Control.Concurrent.Class.MonadSTM.Strict import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork @@ -31,13 +32,14 @@ import Control.Monad.Class.MonadTimer.SI import Control.Monad.IOSim import Control.Tracer (Tracer (..), contramap) -import Data.Bifoldable import Data.ByteString.Lazy qualified as BSL import Data.Foldable (traverse_) import Data.Function (on) import Data.Hashable import Data.List (nubBy) import Data.List qualified as List +import Data.List.Trace qualified as Trace +import Data.Map.Merge.Strict import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Maybe (fromMaybe, isJust) @@ -104,7 +106,7 @@ instance Arbitrary TxSubmissionState where -- NOTE: using sortOn would forces tx-decision logic to download txs in the -- order of unacknowledgedTxIds. This could be useful to get better -- properties when wrongly sized txs are present. - txs <- divvy txsN . nubBy (on (==) getTxId) {- . List.sortOn getTxId -} <$> vectorOf (peersN * txsN) arbitrary + txs <- fmap (nubBy (on (==) getTxId)) . divvy txsN {- . List.sortOn getTxId -} <$> vectorOf (peersN * txsN) arbitrary peers <- vectorOf peersN arbitrary peersState <- zipWith (curry (\(a, (b, c)) -> (a, b, c))) txs <$> vectorOf peersN arbitrary @@ -177,9 +179,14 @@ runTxSubmission tracer tracerTxLogic st0 txDecisionPolicy = do ) st0 inboundMempool <- emptyMempool let txRng = mkStdGen 42 -- TODO + txMap = Map.fromList [ (getTxId tx, tx) + | (txs, _, _, _) <- Map.elems st0 + , tx <- txs] + txChannelsVar <- newMVar (TxChannels Map.empty) txMempoolSem <- newTxMempoolSem + duplicateTxIdsVar <- Lazy.newTVarIO [] sharedTxStateVar <- newSharedTxStateVar txRng traceTVarIO sharedTxStateVar \_ -> return . TraceDynamic . TxStateTrace labelTVarIO sharedTxStateVar "shared-tx-state" @@ -212,15 +219,18 @@ runTxSubmission tracer tracerTxLogic st0 txDecisionPolicy = do txDecisionPolicy sharedTxStateVar (getMempoolReader inboundMempool) - (getMempoolWriter inboundMempool) + (getMempoolWriter duplicateTxIdsVar inboundMempool) getTxSize addr $ \api -> do - let server = txSubmissionInboundV2 verboseTracer - NoTxSubmissionInitDelay - (getMempoolWriter inboundMempool) - api + let server = + txSubmissionInboundV2 sayTracer --verboseTracer + NoTxSubmissionInitDelay + (getMempoolWriter duplicateTxIdsVar + inboundMempool) + api runPipelinedPeerWithLimits - (("INBOUND " ++ show addr,) `contramap` verboseTracer) + + (("INBOUND " ++ show addr,) `contramap` sayTracer) txSubmissionCodec2 (byteLimitsTxSubmission2 (fromIntegral . BSL.length)) timeLimitsTxSubmission2 @@ -235,10 +245,12 @@ runTxSubmission tracer tracerTxLogic st0 txDecisionPolicy = do cancel a inmp <- readMempool inboundMempool + dupTxIds <- Lazy.readTVarIO duplicateTxIdsVar let outmp = map (\(txs, _, _, _) -> txs) $ Map.elems st0 + dupTxs = [ txMap Map.! txid | txid <- dupTxIds] - return (inmp, outmp) + return (inmp <> dupTxs, outmp) where waitAllServers :: [(Async m x, Async m x)] -> m [Either SomeException x] waitAllServers [] = return [] @@ -299,9 +311,7 @@ txSubmissionSimulation (TxSubmissionState state txDecisionPolicy) = do atomically (traverse_ (`writeTVar` Terminate) controlMessageVars) ) \_ -> do let tracer :: forall a. (Show a, Typeable a) => Tracer (IOSim s) a - tracer = verboseTracer - <> debugTracer - <> Tracer traceM + tracer = dynamicTracer <> sayTracer -- <> verboseTracer <> debugTracer runTxSubmission tracer tracer state'' txDecisionPolicy filterValidTxs :: [Tx txid] -> [Tx txid] @@ -344,6 +354,12 @@ prop_txSubmission st@(TxSubmissionState peers _) = counterexample (ppTrace tr) $ conjoin (validate inmp `map` outmps) where + checkMempools :: [Tx Int] -> [Tx Int] -> Bool + checkMempools consumer producer = + let producer' = Set.fromList $ getTxId <$> producer + consumer' = Set.fromList $ getTxId <$> consumer + in producer' `Set.isSubsetOf` consumer' + validate :: [Tx Int] -- the inbound mempool -> [Tx Int] -- one of the outbound mempools -> Property @@ -361,7 +377,7 @@ prop_txSubmission st@(TxSubmissionState peers _) = counterexample (show x) . counterexample (show inmp) . counterexample (show outmp) - $ checkMempools inmp (take (length inmp) outValidTxs) + $ checkMempools inmp outValidTxs x@(True, False) | Nothing <- List.find (\tx -> getTxAdvSize tx /= getTxSize tx) outmp -> -- If we are presented with a stream of unique txids then we should have @@ -370,7 +386,7 @@ prop_txSubmission st@(TxSubmissionState peers _) = . counterexample (show inmp) . counterexample (show outValidTxs) - $ checkMempools inmp (take (length inmp) outValidTxs) + $ checkMempools inmp outValidTxs | otherwise -> -- If there's one tx with an invalid size, we will download only -- some of them, but we don't guarantee how many we will download. @@ -385,9 +401,8 @@ prop_txSubmission st@(TxSubmissionState peers _) = counterexample (show x) . counterexample (show inmp) . counterexample (show outmp) - $ checkMempools (map getTxId inmp) - (take (length inmp) - (getTxId <$> filterValidTxs outUniqueTxIds)) + $ checkMempools inmp + (filterValidTxs outUniqueTxIds) (False, False) -> -- If we are presented with a stream of valid and invalid Txs with @@ -399,14 +414,11 @@ prop_txSubmission st@(TxSubmissionState peers _) = -- | This test checks that all txs are downloaded from all available peers if -- available. -- --- This test takes advantage of the fact that the mempool implementation --- allows duplicates. --- --- TODO: do we generated enough outbound mempools which intersect in interesting +-- TODO: have we generated enough outbound mempools which interact in interesting -- ways? prop_txSubmission_inflight :: TxSubmissionState -> Property -prop_txSubmission_inflight st@(TxSubmissionState state _) = - let maxRepeatedValidTxs = Map.foldr (\(txs, _, _) r -> foldr (fn r) r txs) +prop_txSubmission_inflight st@(TxSubmissionState state policy) = + let maxRepeatedValidTxs = Map.foldr (\(txs, _, _) r -> foldr fn r txs) Map.empty state hasInvalidSize = @@ -416,88 +428,100 @@ prop_txSubmission_inflight st@(TxSubmissionState state _) = ) state trace = runSimTrace (txSubmissionSimulation st) + pTrace = List.intercalate "\n" $ map (\(Time t, ev) -> show t <> " " <> ev) $ + selectTraceEventsSayWithTime' trace in case traceResult True trace of - Left err -> counterexample (ppTrace trace) + Left err -> counterexample pTrace --(ppTrace trace) $ counterexample (show err) $ property False Right (inmp, _) -> let resultRepeatedValidTxs = - foldr (fn Map.empty) Map.empty inmp + foldr fn Map.empty inmp in label (if hasInvalidSize then "has wrongly sized tx" else "has no wrongly sized tx") - . counterexample (ppTrace trace) - . counterexample (show resultRepeatedValidTxs) - . counterexample (show maxRepeatedValidTxs) - $ if hasInvalidSize - then resultRepeatedValidTxs `Map.isSubmapOf` maxRepeatedValidTxs - else resultRepeatedValidTxs == maxRepeatedValidTxs + . counterexample pTrace --(ppTrace trace) + . counterexample ("hasInvalidSize: " <> show hasInvalidSize) + . counterexample ("Result valid [(txid, repeated)]:\n" <> show resultRepeatedValidTxs) + . counterexample ("Testcase max valid [(txid, repeated)]:\n" <> show maxRepeatedValidTxs) + . conjoin . Map.elems $ if hasInvalidSize + then merge (mapMissing \_txid _left -> error "impossible") + (mapMissing \_txid _right -> True) + (zipWithMatched \_txid left right -> + left <= right `min` txInflightMultiplicity policy + ) + resultRepeatedValidTxs + maxRepeatedValidTxs + else merge (mapMissing \_txid _left -> error "impossible") + (mapMissing \_txid _right -> False) + (zipWithMatched \_txid left right -> + if txInflightMultiplicity policy >= right + then left <= right + else left <= txInflightMultiplicity policy + ) + resultRepeatedValidTxs + maxRepeatedValidTxs where - fn empty tx rr | getTxAdvSize tx /= getTxSize tx - = empty - | Map.member tx rr - , getTxValid tx - = Map.update (Just . succ @Int) tx rr - | getTxValid tx - = Map.insert tx 1 rr - | otherwise - = rr + -- we work with txid's because a repeated tx may have different advertised/actual + -- byte size by different peers in this test, but otherwise multiplicity + -- should be determined by txid. + fn :: Tx TxId -> Map TxId Int -> Map TxId Int + fn tx r' -- | getTxAdvSize tx /= getTxSize tx + -- = empty + -- ^ that is too severe + | getTxValid tx + = Map.alter (Just . maybe 1 succ) (getTxId tx) r' + | otherwise + = r' prop_sharedTxStateInvariant :: TxSubmissionState -> Property prop_sharedTxStateInvariant initialState@(TxSubmissionState st0 _) = let tr = runSimTrace (() <$ txSubmissionSimulation initialState) + pTrace = List.intercalate "\n" $ map (\(Time t, ev) -> show t <> " " <> ev) $ + selectTraceEventsSayWithTime' tr in case traceResult True tr of - Left err -> counterexample (ppTrace tr) + Left err -> counterexample pTrace --(ppTrace tr) . counterexample (show err) $ False Right _ -> - let tr' :: Trace (SimResult ()) TxStateTraceType - tr' = traceSelectTraceEventsDynamic tr - in case - bifoldMap (\_ -> (Every (property True), Sum 0)) - (\case - (TxStateTrace st)-> ( Every $ counterexample (show st) - $ sharedTxStateInvariant WeakInvariant st - , Sum 1 - ) + let lookBack, tr' :: [TxStateTraceType] + lookBack = Trace.toList $ traceSelectTraceEventsDynamic tr + tr' = drop 1 lookBack + in counterexample pTrace case + foldMap (\case + (TxStateTrace stBack, TxStateTrace st)-> + (Every . counterexample (show st) $ + sharedTxStateInvariant WeakInvariant st + .&&. let inflight = Map.keysSet $ inflightTxs st + buffered = Map.keysSet $ bufferedTxs st + inflightBack = Map.keysSet $ inflightTxs stBack + in + -- here we account for a very slow peer from whom we requested + -- a transaction, but it didn't arrive until we have also requested + -- it from another peer, received it, and placed it into the mempool, + -- and so it ended up in bufferedTxs when the first one is still + -- in flight. It is an error when the opposite happens. + null $ (inflight Set.\\ inflightBack) `Set.intersection` buffered + , Sum 1 + ) ) - tr' - of (p, Sum c) -> - label ("number of txs: " - ++ - renderRanges 10 - ( Set.size - . foldMap (Set.fromList . (\(txs, _, _) -> getTxId <$> txs)) - $ Map.elems st0 - )) - . label ("number of evaluated states: " - ++ renderRanges 100 c) - $ p + (zip lookBack tr') + of (p, Sum c) -> + label ("number of txs: " + ++ + renderRanges 10 + ( Set.size + . foldMap (Set.fromList . (\(txs, _, _) -> getTxId <$> txs)) + $ Map.elems st0 + )) + . label ("number of evaluated states: " + ++ renderRanges 100 c) + $ p -- -- Utils -- --- | Check that the inbound mempool contains all outbound `tx`s as a proper --- subsequence. It might contain more `tx`s from other peers. --- -checkMempools :: Eq tx - => [tx] -- inbound mempool - -> [tx] -- outbound mempool - -> Bool -checkMempools _ [] = True -- all outbound `tx` were found in the inbound - -- mempool -checkMempools [] (_:_) = False -- outbound mempool contains `tx`s which were - -- not transferred to the inbound mempool -checkMempools (i : is') os@(o : os') - | i == o - = checkMempools is' os' - - | otherwise - -- `_i` is not present in the outbound mempool, we can skip it. - = checkMempools is' os - - -- | Split a list into sub list of at most `n` elements. -- divvy :: Int -> [a] -> [[a]] diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs index d8179022c5e..b05a970710d 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs @@ -23,13 +23,14 @@ module Test.Ouroboros.Network.TxSubmission.TxLogic , sharedTxStateInvariant , InvariantStrength (..) -- * Utils - , mkDecisionContext + , mkDecisionContexts + , printTxLogicBenchmarkContexts ) where import Prelude hiding (seq) import Control.Exception (assert) -import Control.Monad.Class.MonadTime.SI (Time (..)) +import Control.Monad.Class.MonadTime.SI (Time (..), addTime) import Data.Foldable as Foldable (fold, foldl', toList) import Data.List (intercalate, isPrefixOf, isSuffixOf, mapAccumR, nub, @@ -43,8 +44,9 @@ import Data.Sequence.Strict qualified as StrictSeq import Data.Set (Set) import Data.Set qualified as Set import Data.Typeable -import System.Random (StdGen, mkStdGen) -import System.Random.SplitMix (SMGen) +import Data.Word (Word32) +import System.Random (StdGen, mkStdGen, randomR, randoms, splitGen) +import System.Random.SplitMix qualified as SM import NoThunks.Class @@ -69,6 +71,7 @@ import "quickcheck-monoids" Test.QuickCheck.Monoids import Test.Tasty (TestTree, testGroup) import Test.Tasty.QuickCheck (testProperty) import Text.Pretty.Simple +import Text.Printf (printf) tests :: TestTree @@ -135,6 +138,7 @@ data InvariantStrength = WeakInvariant sharedTxStateInvariant :: forall peeraddr txid tx. ( Ord txid + , Ord peeraddr , Show txid , Show tx ) @@ -145,39 +149,36 @@ sharedTxStateInvariant invariantStrength SharedTxState { peerTxStates, inflightTxs, - inflightTxsSize, bufferedTxs, referenceCounts, - timedTxs + timedTxs, + pendingDecisions } = - - -- `inflightTxs` and `bufferedTxs` are disjoint - counterexample "inflightTxs not disjoint with bufferedTxs" - (null (Map.keysSet inflightTxs `Set.intersection` bufferedTxsSet)) - .&&. counterexample "bufferedTxs txid not a subset of unacknowledged txids" - let unacknowledgedSet = - foldr (\PeerTxState { unacknowledgedTxIds } r -> - r <> Set.fromList (toList unacknowledgedTxIds)) - Set.empty txStates - timedSet = foldMap Set.fromList timedTxs - in case invariantStrength of - WeakInvariant -> - -- `submitTxToMempool` caches buffered `txs`, we check here that - -- they do not leak - counterexample ("unacknowledgedSet: " ++ show unacknowledgedSet) $ - counterexample ("bufferedTxsSet: " ++ show bufferedTxsSet) $ - counterexample ("timedTxsSet: " ++ show timedSet) $ - (bufferedTxsSet Set.\\ unacknowledgedSet) - `Set.isSubsetOf` - timedSet - - StrongInvariant -> property $ - -- the set of buffered txids must be a subset of sum of the sets of - -- unacknowledged txids - bufferedTxsSet - `Set.isSubsetOf` - unacknowledgedSet + ( + let unacknowledgedSet = + foldr (\PeerTxState { unacknowledgedTxIds } r -> + r <> Set.fromList (toList unacknowledgedTxIds)) + Set.empty txStates + timedSet = foldMap Set.fromList timedTxs + in case invariantStrength of + WeakInvariant -> + -- `submitTxToMempool` caches buffered `txs`, we check here that + -- they do not leak + counterexample ("unacknowledgedSet: " ++ show unacknowledgedSet) $ + counterexample ("bufferedTxsSet: " ++ show bufferedTxsSet) $ + counterexample ("timedTxsSet: " ++ show timedSet) $ + (bufferedTxsSet Set.\\ unacknowledgedSet) + `Set.isSubsetOf` + timedSet + + StrongInvariant -> property $ + -- the set of buffered txids must be a subset of sum of the sets of + -- unacknowledged txids + bufferedTxsSet + `Set.isSubsetOf` + unacknowledgedSet + ) .&&. counterexample "referenceCounts invariant violation" ( @@ -214,13 +215,18 @@ sharedTxStateInvariant invariantStrength ++ show (Map.keysSet bufferedTxs `Set.difference` liveSet)) (Map.keysSet bufferedTxs `Set.isSubsetOf` liveSet) - .&&. counterexample "inflightTxs must be a sum of requestedTxInflight sets" - (inflightTxs + -- Note: inflightTxs can keep zero-count entries to track inter TX spacing + -- (inFlightNextReq); the invariant only applies to positive-count entries. + .&&. counterexample "inflightTxs (count>0) must be a sum of requestedTxInflight sets" + (Map.filter (> 0) (Map.map TXS.inFlightCount inflightTxs) === - foldr (\PeerTxState { requestedTxsInflight } m -> - Map.unionWith (+) (Map.fromSet (\_ -> 1) requestedTxsInflight) m) + Map.map TXS.inFlightCount (foldr (\PeerTxState { requestedTxsInflight } m -> + Map.unionWith (<>) (Map.fromSet (\_ -> TXS.InFlightState 1 (Time 0)) requestedTxsInflight) m) Map.empty - peerTxStates) + peerTxStates)) + + .&&. counterexample "pendingDecisions must be a subset of peers" + (pendingDecisions `Set.isSubsetOf` Map.keysSet peerTxStates) -- PeerTxState invariants .&&. counterexample "PeerTxState invariant violation" @@ -231,11 +237,6 @@ sharedTxStateInvariant invariantStrength ) peerTxStates) - .&&. counterexample "inflightTxsSize invariant violation" - (inflightTxsSize === foldMap requestedTxsInflightSize peerTxStates) - - - where peerTxStateInvariant :: PeerTxState txid tx -> Property peerTxStateInvariant PeerTxState { availableTxIds, @@ -341,6 +342,68 @@ instance Arbitrary tx => Arbitrary (TxMask tx) where -- `mkArbPeerTxState` and shrinking the unacknowledged txs & mask map. +genTxMaskWith :: Gen (Tx txid) -> Gen (TxMask (Tx txid)) +genTxMaskWith genTx = + oneof [ TxAvailable + <$> genTx + <*> arbitrary + , TxBuffered <$> genTx + ] + +genTxStatusBench :: Gen TxStatus +genTxStatusBench = + frequency + [ (6, pure Inflight) + , (3, pure Available) + , (1, pure Unknown) + ] + + +genTxMaskBenchWith :: Gen (Tx txid) -> Gen (TxMask (Tx txid)) +genTxMaskBenchWith genTx = + frequency + [ (9, TxAvailable <$> genTx <*> genTxStatusBench) + , (1, TxBuffered <$> genTx) + ] + + +genArbPeerTxStateWith + :: forall txid. + ( Arbitrary txid + , Ord txid + ) + => Gen (Tx txid) + -> Fun txid Bool + -> Int -- ^ max txids inflight + -> Gen (ArbPeerTxState txid (Tx txid)) +genArbPeerTxStateWith genTx mempoolHasTxFun maxTxIdsInflight = do + -- unacknowledged sequence + unacked <- arbitrary + -- generate `Map txid (TxMask tx)` + txIdsInflight <- choose (0, maxTxIdsInflight) + txMap <- Map.fromList + <$> traverse (\txid -> (\a -> (txid, fixupTxMask txid a)) <$> genTxMaskWith genTx) + (nub unacked) + return $ mkArbPeerTxState mempoolHasTxFun txIdsInflight unacked txMap + +genArbPeerTxStateBench + :: forall txid. + ( Arbitrary txid + , Ord txid + ) + => Gen (Tx txid) + -> Fun txid Bool + -> Int -- ^ max txids inflight + -> Gen (ArbPeerTxState txid (Tx txid)) +genArbPeerTxStateBench genTx mempoolHasTxFun maxTxIdsInflight = do + unacked <- sized $ \n -> resize (max 30 (min 200 n)) arbitrary + txIdsInflight <- choose (max 1 (maxTxIdsInflight `div` 2), max 1 maxTxIdsInflight) + txMap <- Map.fromList + <$> traverse (\txid -> (\a -> (txid, fixupTxMask txid a)) <$> genTxMaskBenchWith genTx) + (nub unacked) + return $ mkArbPeerTxState mempoolHasTxFun txIdsInflight unacked txMap + + -- | Smart constructor for `ArbPeerTxState`. -- mkArbPeerTxState :: Ord txid @@ -404,42 +467,63 @@ mkArbPeerTxState mempoolHasTxFun txIdsInflight unacked txMaskMap = ] -genArbPeerTxState +genSharedTxState :: forall txid. ( Arbitrary txid , Ord txid + , Function txid + , CoArbitrary txid ) - => Fun txid Bool + => Int -- ^ max txids inflight + -> Gen ( Fun txid Bool + , (PeerAddr, PeerTxState txid (Tx txid)) + , SharedTxState PeerAddr txid (Tx txid) + , Map PeerAddr (ArbPeerTxState txid (Tx txid)) + ) +genSharedTxState maxTxIdsInflight = + genSharedTxStateWith (arbitrary :: Gen (Tx txid)) maxTxIdsInflight + + +genSharedTxStateWith + :: forall txid. + ( Arbitrary txid + , Ord txid + , Function txid + , CoArbitrary txid + ) + => Gen (Tx txid) -> Int -- ^ max txids inflight - -> Gen (ArbPeerTxState txid (Tx txid)) -genArbPeerTxState mempoolHasTxFun maxTxIdsInflight = do - -- unacknowledged sequence - unacked <- arbitrary - -- generate `Map txid (TxMask tx)` - txIdsInflight <- choose (0, maxTxIdsInflight) - txMap <- Map.fromList - <$> traverse (\txid -> (\a -> (txid, fixupTxMask txid a)) <$> arbitrary) - (nub unacked) - return $ mkArbPeerTxState mempoolHasTxFun txIdsInflight unacked txMap + -> Gen ( Fun txid Bool + , (PeerAddr, PeerTxState txid (Tx txid)) + , SharedTxState PeerAddr txid (Tx txid) + , Map PeerAddr (ArbPeerTxState txid (Tx txid)) + ) +genSharedTxStateWith genTx maxTxIdsInflight = do + genSharedTxStateWithPeerGen + (\mempoolHasTxFun maxTxIdsInflight' -> + genArbPeerTxStateWith genTx mempoolHasTxFun maxTxIdsInflight' + ) + maxTxIdsInflight -genSharedTxState +genSharedTxStateWithPeerGen :: forall txid. ( Arbitrary txid , Ord txid , Function txid , CoArbitrary txid ) - => Int -- ^ max txids inflight + => (Fun txid Bool -> Int -> Gen (ArbPeerTxState txid (Tx txid))) + -> Int -- ^ max txids inflight -> Gen ( Fun txid Bool , (PeerAddr, PeerTxState txid (Tx txid)) , SharedTxState PeerAddr txid (Tx txid) , Map PeerAddr (ArbPeerTxState txid (Tx txid)) ) -genSharedTxState maxTxIdsInflight = do +genSharedTxStateWithPeerGen genPeerState maxTxIdsInflight = do _mempoolHasTxFun@(Fun (_, _, x) _) <- arbitrary :: Gen (Fun Bool Bool) let mempoolHasTxFun = Fun (function (const False), False, x) (const False) - pss <- listOf1 (genArbPeerTxState mempoolHasTxFun maxTxIdsInflight) + pss <- listOf1 (genPeerState mempoolHasTxFun maxTxIdsInflight) seed <- arbitrary let pss' :: [(PeerAddr, ArbPeerTxState txid (Tx txid))] @@ -456,12 +540,11 @@ genSharedTxState maxTxIdsInflight = do | (peeraddr, ArbPeerTxState { arbPeerTxState }) <- pss' ], - inflightTxs = Foldable.foldl' (Map.unionWith (+)) Map.empty - [ Map.fromSet (const 1) (Set.map getTxId arbInflightSet) + inflightTxs = Foldable.foldl' (Map.unionWith (<>)) Map.empty + [ Map.fromSet (const $ TXS.InFlightState 1 (Time 0)) (Set.map getTxId arbInflightSet) | ArbPeerTxState { arbInflightSet } <- pss ], - inflightTxsSize = 0, -- It is set by fixupSharedTxState bufferedTxs = fold [ arbBufferedMap | ArbPeerTxState { arbBufferedMap } @@ -471,7 +554,10 @@ genSharedTxState maxTxIdsInflight = do timedTxs = Map.empty, inSubmissionToMempoolTxs = Map.empty, - peerRng = mkStdGen seed + pendingDecisions + = Set.empty, + peerRng = mkStdGen seed, + generation = 0 } return ( mempoolHasTxFun @@ -481,6 +567,28 @@ genSharedTxState maxTxIdsInflight = do ) +genSharedTxStateBenchWith + :: forall txid. + ( Arbitrary txid + , Ord txid + , Function txid + , CoArbitrary txid + ) + => Gen (Tx txid) + -> Int -- ^ max txids inflight + -> Gen ( Fun txid Bool + , (PeerAddr, PeerTxState txid (Tx txid)) + , SharedTxState PeerAddr txid (Tx txid) + , Map PeerAddr (ArbPeerTxState txid (Tx txid)) + ) +genSharedTxStateBenchWith genTx maxTxIdsInflight = + genSharedTxStateWithPeerGen + (\mempoolHasTxFun maxTxIdsInflight' -> + genArbPeerTxStateBench genTx mempoolHasTxFun maxTxIdsInflight' + ) + maxTxIdsInflight + + -- | Make sure `SharedTxState` is well formed. -- fixupSharedTxState @@ -491,7 +599,6 @@ fixupSharedTxState fixupSharedTxState _mempoolHasTx st@SharedTxState { peerTxStates } = st { peerTxStates = peerTxStates', inflightTxs = inflightTxs', - inflightTxsSize = foldMap requestedTxsInflightSize peerTxStates', bufferedTxs = bufferedTxs', referenceCounts = referenceCounts' } @@ -513,8 +620,8 @@ fixupSharedTxState _mempoolHasTx st@SharedTxState { peerTxStates } = peerTxStates inflightTxs' = foldr (\PeerTxState { requestedTxsInflight } m -> - Map.unionWith (+) - (Map.fromSet (const 1) requestedTxsInflight) + Map.unionWith (<>) + (Map.fromSet (const $ TXS.InFlightState 1 (Time 0)) requestedTxsInflight) m ) Map.empty @@ -890,8 +997,6 @@ prop_collectTxs_generator (ArbCollectTxs _ requestedTxIds receivedTxs peeraddr st) = counterexample "size of requested txs must not be larger than requestedTxsInflightSize" (requestedSize <= requestedTxsInflightSize) - .&&. counterexample "inflightTxsSize must be greater than requestedSize" - (inflightTxsSize st >= requestedSize) .&&. counterexample ("receivedTxs must be a subset of requestedTxIds " ++ show (Map.keysSet receivedTxs Set.\\ requestedTxIdsSet)) (Map.keysSet receivedTxs `Set.isSubsetOf` requestedTxIdsSet) @@ -1017,7 +1122,8 @@ instance Arbitrary ArbTxDecisionPolicy where <*> (getSmall . getPositive <$> arbitrary) <*> (realToFrac <$> choose (0 :: Double, 2)) <*> choose (0, 1) - <*> choose (0, 1800)) + <*> choose (0, 1800) + <*> (realToFrac <$> choose (0 :: Double, 1))) shrink (ArbTxDecisionPolicy a@TxDecisionPolicy { maxNumTxIdsToRequest, @@ -1090,15 +1196,18 @@ data ArbDecisionContexts txid = ArbDecisionContexts { arbSharedState :: SharedTxState PeerAddr txid (Tx txid), - arbMempoolHasTx :: Fun txid Bool + arbMempoolHasTx :: Fun txid Bool, -- ^ needed just for shrinking + + arbTime :: Time } instance Show txid => Show (ArbDecisionContexts txid) where show ArbDecisionContexts { arbDecisionPolicy, arbSharedState = st, - arbMempoolHasTx + arbMempoolHasTx, + arbTime } = intercalate "\n\t" @@ -1106,6 +1215,7 @@ instance Show txid => Show (ArbDecisionContexts txid) where , show arbDecisionPolicy , show st , show arbMempoolHasTx + , show arbTime ] -- | Fix-up `PeerTxState` according to `TxDecisionPolicy`. @@ -1154,11 +1264,13 @@ fixupSharedTxStateForPolicy :: forall peeraddr txid tx. Ord txid => (txid -> Bool) -- ^ mempoolHasTx + -> Time -> TxDecisionPolicy -> SharedTxState peeraddr txid tx -> SharedTxState peeraddr txid tx fixupSharedTxStateForPolicy mempoolHasTx + now policy@TxDecisionPolicy { txsSizeInflightPerPeer, maxTxsSizeInflight, @@ -1166,9 +1278,12 @@ fixupSharedTxStateForPolicy } st@SharedTxState { peerTxStates } = - fixupSharedTxState - mempoolHasTx - st { peerTxStates = snd . mapAccumR fn (0, Map.empty) $ peerTxStates } + let st' = + fixupSharedTxState + mempoolHasTx + st { peerTxStates = snd . mapAccumR fn (0, Map.empty) $ peerTxStates } + (rng', inflightTxs') = assignInflightTimes now (peerRng st') (inflightTxs st') + in st' { inflightTxs = inflightTxs', peerRng = rng' } where -- fixup `PeerTxState` and accumulate size of all `tx`'s in-flight across -- all peers. @@ -1207,6 +1322,20 @@ fixupSharedTxStateForPolicy (0, Set.empty, inflightMap) (availableTxIds ps' `Map.restrictKeys` requestedTxsInflight ps') + assignInflightTimes + :: Time + -> StdGen + -> Map txid TXS.InFlightState + -> (StdGen, Map txid TXS.InFlightState) + assignInflightTimes now' = + Map.mapAccum + (\g inflightSt -> + let (delta, g') = randomR (-10.0, 10.0 :: Double) g + nextReq = addTime (realToFrac delta) now' + in (g', inflightSt { TXS.inFlightNextReq = nextReq }) + ) + + instance (Arbitrary txid, Ord txid, Function txid, CoArbitrary txid) => Arbitrary (ArbDecisionContexts txid) where @@ -1214,46 +1343,224 @@ instance (Arbitrary txid, Ord txid, Function txid, CoArbitrary txid) ArbTxDecisionPolicy policy <- arbitrary (mempoolHasTx, _ps, st, _) <- genSharedTxState (fromIntegral $ maxNumTxIdsToRequest policy) + arbTime <- Time . realToFrac <$> choose (1 :: Double, 1000) let st' = fixupSharedTxStateForPolicy - (apply mempoolHasTx) policy st + (apply mempoolHasTx) arbTime policy st return $ ArbDecisionContexts { arbDecisionPolicy = policy, arbMempoolHasTx = mempoolHasTx, - arbSharedState = st' + arbSharedState = st', + arbTime = arbTime } shrink a@ArbDecisionContexts { arbDecisionPolicy = policy, arbMempoolHasTx = mempoolHasTx, - arbSharedState = sharedState + arbSharedState = sharedState, + arbTime = now } = -- shrink shared state [ a { arbSharedState = sharedState'' } | sharedState' <- shrinkSharedTxState (apply mempoolHasTx) sharedState , let sharedState'' = fixupSharedTxStateForPolicy - (apply mempoolHasTx) policy sharedState' + (apply mempoolHasTx) now policy sharedState' , sharedState'' /= sharedState ] --- | Construct decision context in a deterministic way. For micro benchmarks. --- --- It is based on QuickCheck's `arbitrary` instance for `ArbDecisionContexts. --- -mkDecisionContext :: SMGen - -- ^ pseudo random generator - -> Int - -- ^ size - -> (TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId)) -mkDecisionContext stdgen size = - case unGen gen (QCGen stdgen) size of - ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = sharedState - } -> (policy, sharedState) +genTxSizeRealistic :: Gen SizeInBytes +genTxSizeRealistic = + -- Distribution targets (approx): + -- min 55, p10 268, p25 402, median 879, p75 1400, p90 4500, p95 9600, max 16384, mean ~1600. + -- Distribution taken from all TXs on mainnet during 2024 - 2025. + frequency + [ (10, biasedRange (SizeInBytes 55) (SizeInBytes 268)) + , (15, biasedRange (SizeInBytes 269) (SizeInBytes 402)) + , (25, biasedRange (SizeInBytes 403) (SizeInBytes 879)) + , (25, biasedRange (SizeInBytes 880) (SizeInBytes 1400)) + , (15, biasedRange (SizeInBytes 1401) (SizeInBytes 4500)) + , (5, biasedRange (SizeInBytes 4501) (SizeInBytes 9600)) + , (5, biasedRange (SizeInBytes 9601) (SizeInBytes 16384)) + ] + where + + -- NOTE: This is intentionally not `choose lo hi` (uniform). Squaring `u` + -- biases toward smaller values to match the target quantiles above. + biasedRange :: SizeInBytes -> SizeInBytes -> Gen SizeInBytes + biasedRange (SizeInBytes lo) (SizeInBytes hi) = do + u <- choose (0.0, 1.0 :: Double) + let range = fromIntegral (hi - lo) :: Double + offset :: Word32 + offset = floor (u * u * range) + pure (SizeInBytes (lo + offset)) + + +genTxRealistic :: Arbitrary txid => Gen (Tx txid) +genTxRealistic = do + size <- genTxSizeRealistic + valid <- frequency [ (3, pure True) + , (1, pure False) + ] + Tx <$> arbitrary + <*> pure size + <*> pure size -- We assume matching TX sizes + <*> pure valid + + +--- | Construct decision context in a deterministic way. For micro benchmarks. +--- +-- It uses default policy values as a base and realistic tx sizes. +mkDecisionContexts :: Int + -- ^ seed for deriving per-context generators + -> Int + -- ^ number of contexts to generate + -> Int + -- ^ size + -> [ (Time, TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId)) ] +mkDecisionContexts seed count size = + map mkContext subSeeds where - gen :: Gen (ArbDecisionContexts TxId) - gen = arbitrary + subSeeds :: [Int] + subSeeds = take count (randoms (mkStdGen seed) :: [Int]) + + mkContext :: Int -> (Time, TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId)) + mkContext subSeed = + let smGen = SM.mkSMGen (fromIntegral subSeed) + (smGenState, smGenTimePick) = splitGen smGen + qcGenState = QCGen smGenState + qcGenPick = QCGen smGenTimePick + (policy, st, mempoolHasTx) = unGen gen qcGenState size + now = unGen (Time . realToFrac <$> choose (1 :: Double, 1000)) qcGenPick size + st' = fixupSharedTxStateForPolicy + (apply mempoolHasTx) now policy st + in (now, policy, st') + + gen :: Gen (TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId), Fun TxId Bool) + gen = do + let policy = defaultTxDecisionPolicy + (mempoolHasTx, _ps, st, _) <- + genSharedTxStateBenchWith genTxRealistic (fromIntegral $ maxNumTxIdsToRequest policy) + return (policy, st, mempoolHasTx) + + +summarizeTxLogicBenchmarkContext + :: [ (Time, TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId)) ] + -> String +summarizeTxLogicBenchmarkContext contexts = + case contexts of + [] -> + unlines + [ "TxLogic benchmark context" + , "empty context list" + ] + _ -> + unlines + [ "TxLogic benchmark context (mean over " ++ show count ++ " contexts)" + , "policy: " ++ policySummary + , "state: " ++ stateSummary + ] + where + count = length contexts + + mean :: [Double] -> Double + mean xs = + if count <= 0 + then 0 + else sum xs / fromIntegral count + + policySummary = + intercalate " " + [ "maxReq=" ++ fmtMean0 (mean [ fromIntegral (getNumTxIdsToReq $ maxNumTxIdsToRequest p) + | (_, p, _) <- contexts + ]) + , "maxUnacked=" ++ fmtMean0 (mean [ fromIntegral (getNumTxIdsToReq $ maxUnacknowledgedTxIds p) + | (_, p, _) <- contexts + ]) + , "inflightPerPeer=" ++ fmtBytes (mean [ fromIntegral (getSizeInBytes $ txsSizeInflightPerPeer p) + | (_, p, _) <- contexts + ]) + , "inflightTotal=" ++ fmtBytes (mean [ fromIntegral (getSizeInBytes $ maxTxsSizeInflight p) + | (_, p, _) <- contexts + ]) + , "inflightMult=" ++ fmtMean2 (mean [ fromIntegral (txInflightMultiplicity p) + | (_, p, _) <- contexts + ]) + , "bufferedMin=" ++ fmtSeconds (mean [ realToFrac (bufferedTxsMinLifetime p) :: Double + | (_, p, _) <- contexts + ]) + , "scoreRate=" ++ printf "%.3f" (mean [ scoreRate p | (_, p, _) <- contexts ]) + , "scoreMax=" ++ fmtSeconds (mean [ scoreMax p | (_, p, _) <- contexts ]) + ] + + stateMetrics = + [ let peerStates = Map.elems (peerTxStates st) + peers = length peerStates + bufferedCount = Map.size (bufferedTxs st) + inflightUnique = Map.size (inflightTxs st) + inflightTotal = Map.foldl (\s e -> s + TXS.inFlightCount e) 0 (inflightTxs st) + totalInflightTxs = + sum [ Set.size (requestedTxsInflight ps) | ps <- peerStates ] + totalInflightBytes = + sum [ fromIntegral (getSizeInBytes $ requestedTxsInflightSize ps) + | ps <- peerStates + ] :: Double + totalAvailable = + sum [ fromIntegral (Map.size (availableTxIds ps)) | ps <- peerStates ] + totalUnacked = + sum [ fromIntegral (StrictSeq.length (unacknowledgedTxIds ps)) | ps <- peerStates ] + totalUnknown = + sum [ fromIntegral (Set.size (unknownTxs ps)) | ps <- peerStates ] + perPeer n = + if peers <= 0 + then 0 + else n / fromIntegral peers + in ( fromIntegral peers + , fromIntegral bufferedCount + , fromIntegral inflightUnique + , fromIntegral inflightTotal + , perPeer (fromIntegral totalInflightTxs) + , perPeer totalInflightBytes + , perPeer totalAvailable + , perPeer totalUnacked + , perPeer totalUnknown + ) + | (_, _, st) <- contexts + ] + + meanMetric f = mean (map f stateMetrics) + + stateSummary = + intercalate " " + [ "peers=" ++ fmtMean0 (meanMetric (\(a, _, _, _, _, _, _, _, _) -> a)) + , "bufferedTxs=" ++ fmtMean0 (meanMetric (\(_, b, _, _, _, _, _, _, _) -> b)) + , "inflightTxIds=" ++ fmtMean0 (meanMetric (\(_, _, c, _, _, _, _, _, _) -> c)) + , "inflightTxsTotal=" ++ fmtMean0 (meanMetric (\(_, _, _, d, _, _, _, _, _) -> d)) + , "meanInflightTxs/peer=" ++ fmtMean2 (meanMetric (\(_, _, _, _, e, _, _, _, _) -> e)) + , "meanInflightBytes/peer=" ++ fmtBytes (meanMetric (\(_, _, _, _, _, f, _, _, _) -> f)) + , "meanAvailableTxIds/peer=" ++ fmtMean2 (meanMetric (\(_, _, _, _, _, _, g, _, _) -> g)) + , "meanUnackedTxIds/peer=" ++ fmtMean2 (meanMetric (\(_, _, _, _, _, _, _, h, _) -> h)) + , "meanUnknownTxIds/peer=" ++ fmtMean2 (meanMetric (\(_, _, _, _, _, _, _, _, i) -> i)) + ] + + fmtMean0 :: Double -> String + fmtMean0 = printf "%.0f" + + fmtMean2 :: Double -> String + fmtMean2 = printf "%.2f" + + fmtBytes :: Double -> String + fmtBytes b = printf "%.0fB" b + + fmtSeconds :: Double -> String + fmtSeconds seconds = printf "%.0fs" seconds + + +printTxLogicBenchmarkContexts + :: [ (Time, TxDecisionPolicy, SharedTxState PeerAddr TxId (Tx TxId)) ] + -> IO () +printTxLogicBenchmarkContexts contexts = + putStrLn (summarizeTxLogicBenchmarkContext contexts) prop_ArbDecisionContexts_generator @@ -1288,8 +1595,9 @@ prop_makeDecisions_sharedstate -> Property prop_makeDecisions_sharedstate ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = sharedTxState } = - let (sharedState, decisions) = TXS.makeDecisions policy sharedTxState (peerTxStates sharedTxState) + arbSharedState = sharedTxState, + arbTime = now } = + let (sharedState, decisions) = TXS.makeDecisions now policy sharedTxState (peerTxStates sharedTxState) in counterexample (show sharedState) $ counterexample (show decisions) $ sharedTxStateInvariant StrongInvariant sharedState @@ -1308,10 +1616,11 @@ prop_makeDecisions_inflight prop_makeDecisions_inflight ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = sharedTxState + arbSharedState = sharedTxState, + arbTime = now } = - let (sharedState', decisions) = TXS.makeDecisions policy sharedTxState (peerTxStates sharedTxState) + let (sharedState', decisions) = TXS.makeDecisions now policy sharedTxState (peerTxStates sharedTxState) inflightSet :: Set TxId inflightSet = foldMap (Map.keysSet . txdTxsToRequest) decisions @@ -1405,27 +1714,14 @@ prop_makeDecisions_policy -> Property prop_makeDecisions_policy ArbDecisionContexts { - arbDecisionPolicy = policy@TxDecisionPolicy { maxTxsSizeInflight, - txsSizeInflightPerPeer, + arbDecisionPolicy = policy@TxDecisionPolicy { txsSizeInflightPerPeer, txInflightMultiplicity }, - arbSharedState = sharedTxState + arbSharedState = sharedTxState, + arbTime = now } = - let (sharedState', _decisions) = TXS.makeDecisions policy sharedTxState (peerTxStates sharedTxState) - maxTxsSizeInflightEff = maxTxsSizeInflight + maxTxSize + let (sharedState', _decisions) = TXS.makeDecisions now policy sharedTxState (peerTxStates sharedTxState) txsSizeInflightPerPeerEff = txsSizeInflightPerPeer + maxTxSize - - sizeInflight = - foldMap (\PeerTxState { availableTxIds, requestedTxsInflight } -> - fold (availableTxIds `Map.restrictKeys` requestedTxsInflight)) - (peerTxStates sharedState') - - in counterexample (show sharedState') $ - - -- size of txs inflight cannot exceed `maxTxsSizeInflight` by more - -- than maximal tx size. - counterexample ("txs inflight exceed limit " ++ show (sizeInflight, maxTxsSizeInflightEff)) - (sizeInflight <= maxTxsSizeInflightEff) - .&&. + in -- size in flight for each peer cannot exceed `txsSizeInflightPerPeer` counterexample "size in flight per peer vaiolation" ( foldMap @@ -1446,7 +1742,7 @@ prop_makeDecisions_policy let inflight = inflightTxs sharedState' in counterexample ("multiplicities violation: " ++ show inflight) - . foldMap (Every . (<= txInflightMultiplicity)) + . foldMap (Every . (\e -> TXS.inFlightCount e <= txInflightMultiplicity)) $ inflight ) @@ -1458,10 +1754,11 @@ prop_makeDecisions_acknowledged -> Property prop_makeDecisions_acknowledged ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = sharedTxState + arbSharedState = sharedTxState, + arbTime = now } = whenFail (pPrintOpt CheckColorTty defaultOutputOptionsDarkBg { outputOptionsCompact = True } sharedTxState) $ - let (_, decisions) = TXS.makeDecisions policy sharedTxState (peerTxStates sharedTxState) + let (_, decisions) = TXS.makeDecisions now policy sharedTxState (peerTxStates sharedTxState) ackFromDecisions :: Map PeerAddr NumTxIdsToAck ackFromDecisions = Map.fromList @@ -1502,15 +1799,18 @@ prop_makeDecisions_exhaustive prop_makeDecisions_exhaustive ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = sharedTxState + arbSharedState = sharedTxState, + arbTime = now } = let (sharedTxState', decisions') - = TXS.makeDecisions policy + = TXS.makeDecisions now + policy sharedTxState (peerTxStates sharedTxState) (sharedTxState'', decisions'') - = TXS.makeDecisions policy + = TXS.makeDecisions now + policy sharedTxState' (peerTxStates sharedTxState') in counterexample ("decisions': " ++ show decisions') @@ -1543,9 +1843,10 @@ instance Arbitrary ArbDecisionContextWithReceivedTxIds where st <- arbitrary + now <- Time . realToFrac <$> choose (1 :: Double, 1000) let st' = fixupSharedTxStateForPolicy (apply mempoolHasTx) - policy st + now policy st ps' = fixupPeerTxStateWithPolicy policy ps txIdsToAck' = take (fromIntegral (TXS.requestedTxIdsInflight $ peerTxStates st' Map.! peeraddr)) txIdsToAck @@ -1591,7 +1892,8 @@ instance Arbitrary ArbDecisionContextWithReceivedTxIds where <- shrink ArbDecisionContexts { arbDecisionPolicy = policy, arbSharedState = st, - arbMempoolHasTx = mempoolHasTx + arbMempoolHasTx = mempoolHasTx, + arbTime = arbTimeFromState st } , peeraddr `Map.member` peerTxStates st' , let txIdsToAck' = take ( fromIntegral @@ -1600,6 +1902,11 @@ instance Arbitrary ArbDecisionContextWithReceivedTxIds where ) txIdsToAck ] + where + arbTimeFromState st' = + case Map.elems (inflightTxs st') of + [] -> Time 0 + inflights -> minimum (map TXS.inFlightNextReq inflights) -- | `filterActivePeers` should not change decisions made by `makeDecisions` @@ -1610,7 +1917,8 @@ prop_filterActivePeers_not_limitting_decisions prop_filterActivePeers_not_limitting_decisions ArbDecisionContexts { arbDecisionPolicy = policy, - arbSharedState = st + arbSharedState = st, + arbTime = now } = counterexample (unlines @@ -1624,12 +1932,12 @@ prop_filterActivePeers_not_limitting_decisions ) (Map.keysSet decisionsOfActivePeers `Set.isSubsetOf` Map.keysSet decisions) where - activePeersMap = TXS.filterActivePeers policy st + activePeersMap = TXS.filterActivePeers now policy st activePeers = Map.keysSet activePeersMap (_, decisionsOfActivePeers) - = TXS.makeDecisions policy st activePeersMap + = TXS.makeDecisions now policy st activePeersMap - (_, decisions) = TXS.makeDecisions policy st (peerTxStates st) + (_, decisions) = TXS.makeDecisions now policy st (peerTxStates st) decisionPeers = Map.keysSet decisions diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxSubmissionV2.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxSubmissionV2.hs deleted file mode 100644 index 2d6085ea250..00000000000 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxSubmissionV2.hs +++ /dev/null @@ -1,399 +0,0 @@ -{-# LANGUAGE BlockArguments #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TupleSections #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE TypeOperators #-} - -{-# OPTIONS_GHC -Wno-orphans #-} - -module Test.Ouroboros.Network.TxSubmission.TxSubmissionV2 (tests) where - -import Prelude hiding (seq) - -import NoThunks.Class - -import Control.Concurrent.Class.MonadMVar.Strict -import Control.Concurrent.Class.MonadSTM.Strict -import Control.Monad (forM) -import Control.Monad.Class.MonadAsync -import Control.Monad.Class.MonadFork -import Control.Monad.Class.MonadSay -import Control.Monad.Class.MonadST -import Control.Monad.Class.MonadThrow -import Control.Monad.Class.MonadTime.SI -import Control.Monad.Class.MonadTimer.SI -import Control.Monad.IOSim hiding (SimResult) -import Control.Tracer (Tracer (..), contramap) - - -import Data.ByteString.Lazy (ByteString) -import Data.ByteString.Lazy qualified as BSL -import Data.Foldable (traverse_) -import Data.Function (on) -import Data.List (nubBy) -import Data.Map.Strict (Map) -import Data.Map.Strict qualified as Map -import Data.Maybe (fromMaybe) -import Data.Void (Void) - -import Cardano.Network.NodeToNode (NodeToNodeVersion (..)) - -import Ouroboros.Network.Channel -import Ouroboros.Network.ControlMessage (ControlMessage (..), ControlMessageSTM) -import Ouroboros.Network.Driver -import Ouroboros.Network.Protocol.TxSubmission2.Client -import Ouroboros.Network.Protocol.TxSubmission2.Codec -import Ouroboros.Network.Protocol.TxSubmission2.Server -import Ouroboros.Network.Protocol.TxSubmission2.Type -import Ouroboros.Network.TxSubmission.Inbound.Policy -import Ouroboros.Network.TxSubmission.Inbound.Registry -import Ouroboros.Network.TxSubmission.Inbound.Server (txSubmissionInboundV2) -import Ouroboros.Network.TxSubmission.Inbound.State -import Ouroboros.Network.TxSubmission.Outbound -import Ouroboros.Network.Util.ShowProxy - -import Test.Ouroboros.Network.TxSubmission.Common hiding (tests) -import Test.Ouroboros.Network.Utils hiding (debugTracer) - -import Test.QuickCheck -import Test.Tasty (TestTree, testGroup) -import Test.Tasty.QuickCheck (testProperty) - - -tests :: TestTree -tests = testGroup "Ouroboros.Network.TxSubmission.TxSubmissionV2" - [ testProperty "txSubmission" prop_txSubmission - , testProperty "txSubmission inflight" prop_txSubmission_inflight - ] - -data TxSubmissionState = - TxSubmissionState { - peerMap :: Map Int ( [Tx Int] - , Maybe (Positive SmallDelay) - , Maybe (Positive SmallDelay) - -- ^ The delay must be smaller (<) than 5s, so that overall - -- delay is less than 10s, otherwise 'smallDelay' in - -- 'timeLimitsTxSubmission2' will kick in. - ) - , decisionPolicy :: TxDecisionPolicy - } deriving (Show) - -instance Arbitrary TxSubmissionState where - arbitrary = do - ArbTxDecisionPolicy decisionPolicy <- arbitrary - peersN <- choose (1, 10) - txsN <- choose (1, 10) - txs <- divvy txsN . nubBy (on (==) getTxId) <$> vectorOf (peersN * txsN) arbitrary - peers <- vectorOf peersN arbitrary - peersState <- map (\(a, (b, c)) -> (a, b, c)) - . zip txs - <$> vectorOf peersN arbitrary - return TxSubmissionState { peerMap = Map.fromList (zip peers peersState), - decisionPolicy - } - shrink TxSubmissionState { peerMap, decisionPolicy } = - TxSubmissionState <$> shrinkMap1 peerMap - <*> [ policy - | ArbTxDecisionPolicy policy <- shrink (ArbTxDecisionPolicy decisionPolicy) - ] - where - shrinkMap1 :: (Ord k, Arbitrary k, Arbitrary v) => Map k v -> [Map k v] - shrinkMap1 m - | Map.size m <= 1 = [m] - | otherwise = [Map.delete k m | k <- Map.keys m] ++ singletonMaps - where - singletonMaps = [Map.singleton k v | (k, v) <- Map.toList m] - -runTxSubmission - :: forall m peeraddr txid. - ( MonadAsync m - , MonadDelay m - , MonadFork m - , MonadMask m - , MonadMVar m - , MonadSay m - , MonadST m - , MonadLabelledSTM m - , MonadTimer m - , MonadThrow m - , MonadThrow (STM m) - , MonadMonotonicTime m - , Ord txid - , Eq txid - , ShowProxy txid - , NoThunks (Tx txid) - , Show peeraddr - , Ord peeraddr - - , txid ~ Int - ) - => Tracer m (String, TraceSendRecv (TxSubmission2 txid (Tx txid))) - -> Tracer m (DebugSharedTxState peeraddr txid (Tx txid)) - -> Tracer m (DebugTxLogic peeraddr txid (Tx txid)) - -> Map peeraddr ( [Tx txid] - , ControlMessageSTM m - , Maybe DiffTime - , Maybe DiffTime - ) - -> TxDecisionPolicy - -> m ([Tx txid], [[Tx txid]]) -runTxSubmission tracer tracerDST tracerTxLogic state txDecisionPolicy = do - - state' <- traverse (\(b, c, d, e) -> do - mempool <- newMempool b - (outChannel, inChannel) <- createConnectedChannels - return (mempool, c, d, e, outChannel, inChannel) - ) state - - inboundMempool <- emptyMempool - - txChannelsMVar <- newMVar (TxChannels Map.empty) - sharedTxStateVar <- newSharedTxStateVar - labelTVarIO sharedTxStateVar "shared-tx-state" - - run state' - txChannelsMVar - sharedTxStateVar - inboundMempool - (\(a, as) -> do - _ <- waitAnyCancel as - cancel a - - inmp <- readMempool inboundMempool - outmp <- forM (Map.elems state') - (\(outMempool, _, _, _, _, _) -> readMempool outMempool) - return (inmp, outmp) - ) - - where - run :: Map peeraddr ( Mempool m txid -- ^ Outbound mempool - , ControlMessageSTM m - , Maybe DiffTime -- ^ Outbound delay - , Maybe DiffTime -- ^ Inbound delay - , Channel m ByteString -- ^ Outbound channel - , Channel m ByteString -- ^ Inbound channel - ) - -> TxChannelsVar m peeraddr txid (Tx txid) - -> SharedTxStateVar m peeraddr txid (Tx txid) - -> Mempool m txid -- ^ Inbound mempool - -> ((Async m Void, [Async m ((), Maybe ByteString)]) -> m b) - -> m b - run st txChannelsVar sharedTxStateVar - inboundMempool k = - withAsync (decisionLogicThread tracerTxLogic txDecisionPolicy txChannelsVar sharedTxStateVar) $ \a -> do - -- Construct txSubmission outbound client - let clients = (\(addr, (mempool, ctrlMsgSTM, outDelay, _, outChannel, _)) -> do - let client = txSubmissionOutbound (Tracer $ say . show) - (NumTxIdsToAck $ getNumTxIdsToReq - $ maxUnacknowledgedTxIds - $ txDecisionPolicy) - (getMempoolReader mempool) - (maxBound :: NodeToNodeVersion) - ctrlMsgSTM - runPeerWithLimits (("OUTBOUND " ++ show addr,) `contramap` tracer) - txSubmissionCodec2 - (byteLimitsTxSubmission2 (fromIntegral . BSL.length)) - timeLimitsTxSubmission2 - (maybe id delayChannel outDelay outChannel) - (txSubmissionClientPeer client) - ) - <$> Map.assocs st - - -- Construct txSubmission inbound server - servers = (\(addr, (_, _, _, inDelay, _, inChannel)) -> - withPeer tracerDST - txChannelsVar - sharedTxStateVar - (getMempoolReader inboundMempool) - addr $ \api -> do - let server = txSubmissionInboundV2 verboseTracer - (getMempoolWriter inboundMempool) - api - runPipelinedPeerWithLimits - (("INBOUND " ++ show addr,) `contramap` verboseTracer) - txSubmissionCodec2 - (byteLimitsTxSubmission2 (fromIntegral . BSL.length)) - timeLimitsTxSubmission2 - (maybe id delayChannel inDelay inChannel) - (txSubmissionServerPeerPipelined server) - ) <$> Map.assocs st - - -- Run clients and servers - withAsyncAll (clients ++ servers) (\asyncs -> k (a, asyncs)) - - withAsyncAll :: MonadAsync m => [m a] -> ([Async m a] -> m b) -> m b - withAsyncAll xs0 action = go [] xs0 - where - go as [] = action (reverse as) - go as (x:xs) = withAsync x (\a -> go (a:as) xs) - -txSubmissionSimulation :: forall s . TxSubmissionState -> IOSim s ([Tx Int], [[Tx Int]]) -txSubmissionSimulation (TxSubmissionState state txDecisionPolicy) = do - state' <- traverse (\(txs, mbOutDelay, mbInDelay) -> do - let mbOutDelayTime = getSmallDelay . getPositive <$> mbOutDelay - mbInDelayTime = getSmallDelay . getPositive <$> mbInDelay - controlMessageVar <- newTVarIO Continue - return ( txs - , controlMessageVar - , mbOutDelayTime - , mbInDelayTime - ) - ) - state - - state'' <- traverse (\(txs, var, mbOutDelay, mbInDelay) -> do - return ( txs - , readTVar var - , mbOutDelay - , mbInDelay - ) - ) - state' - - let simDelayTime = Map.foldl' (\m (txs, _, mbInDelay, mbOutDelay) -> - max m ( fromMaybe 1 (max <$> mbInDelay <*> mbOutDelay) - * realToFrac (length txs `div` 4) - ) - ) - 0 - $ state'' - controlMessageVars = (\(_, x, _, _) -> x) - <$> Map.elems state' - - _ <- async do - threadDelay (simDelayTime + 1000) - atomically (traverse_ (`writeTVar` Terminate) controlMessageVars) - - let tracer :: forall a. Show a => Tracer (IOSim s) a - tracer = verboseTracer <> debugTracer - runTxSubmission tracer tracer tracer state'' txDecisionPolicy - --- | Tests overall tx submission semantics. The properties checked in this --- property test are the same as for tx submission v1. We need this to know we --- didn't regress. --- -prop_txSubmission :: TxSubmissionState -> Property -prop_txSubmission st = - let tr = runSimTrace (txSubmissionSimulation st) in - case traceResult True tr of - Left e -> - counterexample (show e) - . counterexample (ppTrace tr) - $ False - Right (inmp, outmps) -> - counterexample (ppTrace tr) - $ conjoin (validate inmp `map` outmps) - where - validate :: [Tx Int] -- the inbound mempool - -> [Tx Int] -- one of the outbound mempools - -> Property - validate inmp outmp = - let outUniqueTxIds = nubBy (on (==) getTxId) outmp - outValidTxs = filter getTxValid outmp - in - case ( length outUniqueTxIds == length outmp - , length outValidTxs == length outmp - ) of - x@(True, True) -> - -- If we are presented with a stream of unique txids for valid - -- transactions the inbound transactions should match the outbound - -- transactions exactly. - counterexample (show x) - . counterexample (show inmp) - . counterexample (show outmp) - $ checkMempools inmp (take (length inmp) outValidTxs) - - x@(True, False) -> - -- If we are presented with a stream of unique txids then we should have - -- fetched all valid transactions. - counterexample (show x) - . counterexample (show inmp) - . counterexample (show outmp) - $ checkMempools inmp (take (length inmp) outValidTxs) - - x@(False, True) -> - -- If we are presented with a stream of valid txids then we should have - -- fetched some version of those transactions. - counterexample (show x) - . counterexample (show inmp) - . counterexample (show outmp) - $ checkMempools (map getTxId inmp) - (take (length inmp) - (map getTxId $ filter getTxValid outUniqueTxIds)) - - (False, False) -> - -- If we are presented with a stream of valid and invalid Txs with - -- duplicate txids we're content with completing the protocol - -- without error. - property True - --- | This test checks that all txs are downloaded from all available peers if --- available. --- --- This test takes advantage of the fact that the mempool implementation --- allows duplicates. --- -prop_txSubmission_inflight :: TxSubmissionState -> Property -prop_txSubmission_inflight st@(TxSubmissionState state _) = - let trace = runSimTrace (txSubmissionSimulation st) - maxRepeatedValidTxs = Map.foldr (\(txs, _, _) r -> - foldr (\tx rr -> - if Map.member tx rr && getTxValid tx - then Map.update (Just . succ @Int) tx rr - else if getTxValid tx - then Map.insert tx 1 rr - else rr - ) - r - txs - ) - Map.empty - state - - in case traceResult True trace of - Left err -> counterexample (ppTrace trace) - $ counterexample (show err) - $ property False - Right (inmp, _) -> - let resultRepeatedValidTxs = - foldr (\tx rr -> - if Map.member tx rr && getTxValid tx - then Map.update (Just . succ @Int) tx rr - else if getTxValid tx - then Map.insert tx 1 rr - else rr - ) - Map.empty - inmp - in resultRepeatedValidTxs === maxRepeatedValidTxs - - --- | Check that the inbound mempool contains all outbound `tx`s as a proper --- subsequence. It might contain more `tx`s from other peers. --- -checkMempools :: Eq tx - => [tx] -- inbound mempool - -> [tx] -- outbound mempool - -> Bool -checkMempools _ [] = True -- all outbound `tx` were found in the inbound - -- mempool -checkMempools [] (_:_) = False -- outbound mempool contains `tx`s which were - -- not transferred to the inbound mempool -checkMempools (i : is') os@(o : os') - | i == o - = checkMempools is' os' - - | otherwise - -- `_i` is not present in the outbound mempool, we can skip it. - = checkMempools is' os - - --- | Split a list into sub list of at most `n` elements. --- -divvy :: Int -> [a] -> [[a]] -divvy _ [] = [] -divvy n as = take n as : divvy n (drop n as) diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/Types.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/Types.hs index 41a5d08e0db..52d694c8aca 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/Types.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/Types.hs @@ -49,6 +49,7 @@ import Codec.CBOR.Encoding qualified as CBOR import Codec.CBOR.Read qualified as CBOR import Data.ByteString.Lazy (ByteString) +import Data.Functor.Contravariant import Data.Typeable (Typeable) import GHC.Generics (Generic) @@ -123,8 +124,8 @@ getMempoolReader :: forall txid m. getMempoolReader = Mempool.getReader getTxId getTxAdvSize -data InvalidTx = InvalidTx - deriving Show +data InvalidTx = InvalidTx | DuplicateTx + deriving (Eq, Show) getMempoolWriter :: forall txid m. ( MonadSTM m @@ -135,18 +136,21 @@ getMempoolWriter :: forall txid m. , Typeable txid , Show txid ) - => Mempool m txid (Tx txid) + => TVar m [txid] + -> Mempool m txid (Tx txid) -> TxSubmissionMempoolWriter txid (Tx txid) Integer m InvalidTx -getMempoolWriter = Mempool.getWriter InvalidTx - getTxId - (\_ txs -> return - [ if getTxValid tx - then Right tx - else Left (getTxId tx, InvalidTx) - | tx <- txs - ] - ) - (\_ -> return ()) +getMempoolWriter duplicateVar = + Mempool.getWriter DuplicateTx + getTxId + (\_ txs -> return + [ if getTxValid tx + then Right tx + else Left (getTxId tx, InvalidTx) + | tx <- txs + ] + ) + (\t -> atomically $ modifyTVar' duplicateVar + (map fst (filter ((== DuplicateTx) . snd) t) <>)) txSubmissionCodec2 :: MonadST m @@ -225,7 +229,7 @@ verboseTracer :: forall a m. verboseTracer = threadAndTimeTracer $ showTracing $ Tracer say debugTracer :: forall a s. Show a => Tracer (IOSim s) a -debugTracer = threadAndTimeTracer $ showTracing $ Tracer (traceM . show) +debugTracer = threadAndTimeTracer $ show >$< Tracer traceM threadAndTimeTracer :: forall a m. ( MonadAsync m diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Version.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/Version.hs deleted file mode 100644 index 292028943f8..00000000000 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Version.hs +++ /dev/null @@ -1,96 +0,0 @@ -{-# LANGUAGE ScopedTypeVariables #-} - --- | Test `NodeToNodeVersion` and `NodeToClientVersion` codecs. --- -module Test.Ouroboros.Network.Version (tests) where - -import Ouroboros.Network.CodecCBORTerm -import Ouroboros.Network.NodeToClient (NodeToClientVersion (..), - nodeToClientVersionCodec) -import Ouroboros.Network.NodeToNode (NodeToNodeVersion (..), - nodeToNodeVersionCodec) - -import Test.Tasty (TestTree, testGroup) -import Test.Tasty.HUnit - - -tests :: TestTree -tests = - testGroup "Ouroboros.Network.Protocol.Handshake.Version" - [ testGroup "NodeToClientVersion" - [ testCase "NodeToClientVersion round-trip codec property" - (roundTripPropAll nodeToClientVersionCodec) - , testCase "NodeToClientVersion should not deserialise as NodeToNode" - (crossFailurePropAll - nodeToClientVersionCodec - nodeToNodeVersionCodec - ([minBound .. maxBound] :: [NodeToClientVersion])) - ] - , testGroup "NodeToNodeVersion" - [ testCase "NodeToNodeVersion round-trip codec property" - (roundTripPropAll nodeToNodeVersionCodec) - , testCase "NodeToNodeVersion should not deserialise as NodeToClient" - (crossFailurePropAll - nodeToNodeVersionCodec - nodeToClientVersionCodec - ([minBound .. maxBound] :: [NodeToNodeVersion])) - ] - ] - - -roundTripProp :: ( Eq a - , Show a - , Eq failure - , Show failure - ) - => CodecCBORTerm failure a - -> a -> Assertion -roundTripProp codec a = - Right a @=? decodeTerm codec (encodeTerm codec a) - - --- Using `Monoid` instance of `IO ()` -roundTripPropAll - :: forall failure a. - ( Eq a - , Enum a - , Bounded a - , Show a - , Eq failure - , Show failure - ) - => CodecCBORTerm failure a -> Assertion -roundTripPropAll codec = - foldMap (roundTripProp codec) ([minBound..maxBound] :: [a]) - - -crossFailureProp - :: forall failure a b. - ( Show a - , Show b - , Eq failure - , Show failure - ) - => CodecCBORTerm failure a - -> CodecCBORTerm failure b - -> a - -> Assertion -crossFailureProp codecA codecB a = - case decodeTerm codecB (encodeTerm codecA a) of - Right b -> assertFailure (show a ++ "should not deserialise as " ++ show b) - Left _ -> pure () - - -crossFailurePropAll - :: forall failure a b. - ( Show a - , Show b - , Eq failure - , Show failure - ) - => CodecCBORTerm failure a - -> CodecCBORTerm failure b - -> [a] - -> Assertion -crossFailurePropAll codecA codecB = foldMap (crossFailureProp codecA codecB) -