Improve TxSubmission

cardano-diffusion/changelog.d/20260216_123142_crocodile-dentist_fix_tx_submission.md

@@ -0,0 +1,23 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->
+### Non-Breaking
+
+- Integrate TVar to collect duplicate tx's in the mempool writer
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->

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.

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

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

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 })

ouroboros-network/changelog.d/20260216_100847_crocodile-dentist_fix_tx_submission.md

@@ -0,0 +1,27 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->
+
+### 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
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->

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.
+

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