tx-submission V2 reduce tx pipelining

ouroboros-network/bench/Bench/TxSubmissionV2Server.hs

@@ -28,7 +28,7 @@ import GHC.Generics (Generic)
 import Ouroboros.Network.Protocol.TxSubmission2.Server
 import Ouroboros.Network.Protocol.TxSubmission2.Type
            (NumTxIdsToReq (getNumTxIdsToReq), SizeInBytes (..))
-import Ouroboros.Network.TxSubmission.Inbound.V2 (TxDecisionPolicy (..),
+import Ouroboros.Network.TxSubmission.Inbound.V2 (TxDecisionPolicy,
            TxSubmissionInitDelay (NoTxSubmissionInitDelay),
            defaultTxDecisionPolicy, txSubmissionInboundV2)
 import Ouroboros.Network.TxSubmission.Inbound.V2.Registry (newPeerTxRegistry,
@@ -82,10 +82,11 @@ mkDirectServerFixture batches =
 -- peers find the txid in the retained set and ack-skip without ever
 -- requesting the body.
 --
--- The fixture disables pipelined txid requests so the benchmark measures
--- STM contention on the shared state without picker cycles introduced by
--- the pipelined-txid loosening.  Body requests still use the protocol's
--- pipelined wire form.
+-- What this measures is STM contention on the shared state: the txid
+-- picker only ever ends up issuing blocking requests here (a pipelined
+-- txid request needs a non-zero ack and a non-zero request count), so
+-- the txid path contributes no wire-level churn of its own.  Body
+-- requests still use the protocol's pipelined wire form.
 mkMultiPeerFixture
   :: Int -- ^ peer count
   -> Int -- ^ batches per peer
@@ -95,9 +96,7 @@ mkMultiPeerFixture peers batches =
     { dsPeerCount        = peers
     , dsTxIdReplyBatches = batches
     , dsTxSize           = SizeInBytes 1024
-    , dsPolicy           = defaultTxDecisionPolicy {
-                             disablePipelinedTxIdRequests = True
-                           }
+    , dsPolicy           = defaultTxDecisionPolicy
     }
 
 

ouroboros-network/changelog.d/20260824_165520_karl.fb.knutsson_tx_depipeline_main.md

@@ -0,0 +1,8 @@
+### Breaking
+
+- Remove the disablePipelinedTxIdRequests field from TxDecisionPolicy
+
+### Non-Breaking
+
+- Make V2 tx-submission more V1 like by pipelining only when you can ack and request more TXs
+

ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs

@@ -73,14 +73,8 @@ data TxDecisionPolicy = TxDecisionPolicy {
       -- ^ Maximum time a peer's attempt may sit between claim and
       -- entering submission before the per-entry inflight-multiplicity
       -- cap is bumped, allowing another peer to attempt in parallel.
-      maxPeerClaimDelay              :: !DiffTime,
+      maxPeerClaimDelay              :: !DiffTime
       -- ^ Maximum delay penalty for poor performing peers.
-
-      disablePipelinedTxIdRequests   :: !Bool
-      -- ^ When 'True', the txid picker never issues pipelined
-      -- @MsgRequestTxIds@ messages; only blocking requests fire (and
-      -- only when the unack window has been fully drained). Used
-      -- by some benchmarks.
     }
   deriving (Eq, Show)
 
@@ -101,8 +95,7 @@ defaultTxDecisionPolicy =
     scoreAcceptDecrement   = 3,
     interTxSpace           = 0.250,
     inflightTimeout        = 0.600,
-    maxPeerClaimDelay      = 0.250,
-    disablePipelinedTxIdRequests = False
+    maxPeerClaimDelay      = 0.250
   }
 
 -- | Sanity check for a 'TxDecisionPolicy': 'True' when every field lies

ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs

@@ -483,22 +483,11 @@ pickRequestTxIdsAction :: TxIdRequestMode
 pickRequestTxIdsAction txIdRequestMode ctx@PeerActionContext { pacPolicy, pacPeerState }
   | txIdsToAcknowledge <= 0 && txIdsToRequest <= 0 = Nothing
 
-  -- Benchmark hook: when 'disablePipelinedTxIdRequests' is set, never
-  -- fire a request that would be sent as pipelined.  Pipelined fires
-  -- whenever the post-ack queue is non-empty (or we're already in
-  -- pipelined mode), so this guard suppresses both.  The peer then
-  -- parks until the queue can drain via acks, yielding pure blocking
-  -- request behaviour at the wire.
-  | disablePipelinedTxIdRequests pacPolicy
-  , txIdRequestMode == AllowPipelinedTxIdRequests
-    || not (StrictSeq.null unacknowledgedTxIds) = Nothing
-
-  -- A pure-ack pipelined message would burn a pipeline slot for an
-  -- empty reply ('req=0' forces the response empty by construction) and
-  -- shrink our window without growing it.  Defer the ack until we can
-  -- also request more txids.
+  -- Strict pipelining: a pipelined txid request must both ack and
+  -- request; with nothing to ack the peer blocks instead of polling
+  -- with 'ack=0, req=N' messages. Matches V1's pipelining.
   | txIdRequestMode == AllowPipelinedTxIdRequests
-  , txIdsToRequest <= 0 = Nothing
+  , txIdsToAcknowledge <= 0 || txIdsToRequest <= 0 = Nothing
 
   -- Spec: a pipelined (non-blocking) request requires the post-ack
   -- queue to be non-empty.  When the peer's unacked queue is empty to
@@ -551,14 +540,17 @@ pickRequestTxIdsAction txIdRequestMode ctx@PeerActionContext { pacPolicy, pacPee
       (peerUnacknowledgedTxIds pacPeerState)
     unackedAndRequested = numOfUnacked + numOfRequested
 
-    -- How many new txids we can request: capped by the unack-window
+    -- How many new txids we can request: zero while txids sit unacked
+    -- with nothing ackable yet, otherwise capped by the unack-window
     -- room left after this round's ack ('maxUnacknowledgedTxIds -
     -- unackedAndRequested + numOfAcked') and by the per-message
     -- request limit ('maxNumTxIdsToRequest - numOfRequested').
-    txIdsToRequest =
-      fromIntegral $ max 0 $ min
-        (fromIntegral (maxUnacknowledgedTxIds pacPolicy) - unackedAndRequested + numOfAcked)
-        (fromIntegral (maxNumTxIdsToRequest pacPolicy) - numOfRequested)
+    txIdsToRequest
+      | numOfAcked == 0 && numOfUnacked > 0 = 0
+      | otherwise =
+          fromIntegral $ max 0 $ min
+            (fromIntegral (maxUnacknowledgedTxIds pacPolicy) - unackedAndRequested + numOfAcked)
+            (fromIntegral (maxNumTxIdsToRequest pacPolicy) - numOfRequested)
 
 -- | Compute the time delay until the peer should next wake to check for work.
 nextWakeDelay :: PeerActionContext peeraddr txid tx

ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs

@@ -94,7 +94,7 @@ tests =
     , testCase     "nextPeerAction clears in-flight tracking when an orphaned body is pruned" unit_nextPeerAction_clearsInFlightForPrunedBody
     , testProperty "nextPeerAction prunes expired retained txs" prop_nextPeerAction_prunesExpiredRetained
     , testProperty "nextPeerAction keeps retained txs before expiry" prop_nextPeerAction_keepsRetained
-    , testProperty "nextPeerActionPipelined allows request-only (ack=0) when the queue is non-empty" prop_nextPeerActionPipelined_allowsRequestOnly
+    , testProperty "nextPeerActionPipelined rejects request-only (ack=0, req=N) messages" prop_nextPeerActionPipelined_rejectsRequestOnly
     , testCaseSteps "nextPeerActionPipelined rejects pure-ack (ack=N, req=0) messages" unit_nextPeerActionPipelined_rejectsPureAck
     , testProperty "nextPeerActionPipelined emits a pipelined txid request when ack and request fire together" prop_nextPeerActionPipelined_requestsTxIds
     , testProperty "nextPeerActionPipelined opens a second outstanding body batch" prop_nextPeerActionPipelined_secondBodyBatch
@@ -542,8 +542,7 @@ instance Arbitrary ArbTxDecisionPolicy where
                 <*> choose (0, 5)
                 <*> pure interTxSpaceVal
                 <*> pure inflightTimeoutVal
-                <*> (realToFrac <$> choose (0.1, 1 :: Double))
-                <*> pure False))
+                <*> (realToFrac <$> choose (0.1, 1 :: Double))))
         ]
 
     shrink (ArbTxDecisionPolicy a)
@@ -1435,10 +1434,6 @@ prop_nextPeerAction_keepsRetained (ArbTxDecisionPolicy policy)
 -- nextPeerActionPipelined
 --
 
--- | In pipelined mode 'pickRequestTxIdsAction' must return @Nothing@
--- when either the ack count or the request count is zero (the wire
--- format would otherwise produce an ack-only or request-only
--- pipelined message, which is not allowed).
 -- | When the peer has at least one ackable txid AND room to request
 -- more, 'nextPeerActionPipelined' emits a 'PeerRequestTxIds' with both
 -- counts non-zero.
@@ -1515,15 +1510,16 @@ unit_nextPeerActionPipelined_keepsOneUnackedWithOutstandingBodyReply step = do
   where
     peerAddr = 1 :: PeerAddr
 
--- | When the peer has an unacked txid that is *not* ackable (so the
--- ack prefix is empty) but the unack window still has room, the
--- pipelined picker emits a request-only message ('ack=0, req=N').
--- Filling the window without acking is spec-compliant and matches V1's
--- behaviour.
-prop_nextPeerActionPipelined_allowsRequestOnly
+-- | Request-only pipelined messages ('ack=0, req=N') poll the peer for
+-- txids without letting its unack window grow, so the pipelined picker
+-- must not emit them: a pipelined txid request has to both ack and
+-- request.  When the peer has an unacked txid that is *not* ackable (so
+-- the ack prefix is empty), the picker parks instead, even though the
+-- unack window still has room.
+prop_nextPeerActionPipelined_rejectsRequestOnly
   :: ArbTxDecisionPolicy
   -> Property
-prop_nextPeerActionPipelined_allowsRequestOnly (ArbTxDecisionPolicy policy0) =
+prop_nextPeerActionPipelined_rejectsRequestOnly (ArbTxDecisionPolicy policy0) =
   let
     -- Cap to small windows so the property is easy to reason about.
     policy = policy0 { maxUnacknowledgedTxIds = 4
@@ -1563,13 +1559,8 @@ prop_nextPeerActionPipelined_allowsRequestOnly (ArbTxDecisionPolicy policy0) =
   in
     counterexample ("got: " ++ show action) $
       case action of
-        PeerRequestTxIds ack req ->
-          conjoin [ counterexample "ack should be zero (nothing ackable)"
-                      (ack == 0)
-                  , counterexample "req should be non-zero (window has room)"
-                      (req /= 0)
-                  ]
-        _ -> property False
+        PeerDoNothing {} -> property True
+        _                -> property False
   where
     peerAddr = 1 :: PeerAddr