tx-submission: generalised CollectPipelined

Allow monadic action when trying to pipeline more messages, while
collecting responses.

Author: coot

ouroboros-network-protocols/src/Ouroboros/Network/Protocol/TxSubmission2/Server.hs

@@ -83,9 +83,9 @@ data ServerStIdle (n :: N) txid tx m a where
   -- | Collect a pipelined result.
   --
   CollectPipelined
-    :: Maybe                 (ServerStIdle (S n) txid tx m a)
-    -> (Collect txid tx -> m (ServerStIdle    n  txid tx m a))
-    ->                        ServerStIdle (S n) txid tx m a
+    :: Maybe                 (m (ServerStIdle (S n) txid tx m a))
+    -> (Collect txid tx -> m (   ServerStIdle    n  txid tx m a))
+    ->                           ServerStIdle (S n) txid tx m a
 
 
 -- | Transform a 'TxSubmissionServerPipelined' into a 'PeerPipelined'.
@@ -145,6 +145,5 @@ txSubmissionServerPeerPipelined (TxSubmissionServerPipelined server) =
 
     go (CollectPipelined mNone collect) =
       SenderCollect
-        (fmap go mNone)
-        (SenderEffect . fmap go . collect)
-
+        (SenderEffect . fmap go <$> mNone)
+        (SenderEffect . fmap go  .  collect)

ouroboros-network-protocols/testlib/Ouroboros/Network/Protocol/TxSubmission2/Direct.hs

@@ -54,7 +54,8 @@ directPipelined (TxSubmissionServerPipelined mserver)
       SendMsgReplyTxs txs client' <- recvMsgRequestTxs txids
       directSender (enqueue (CollectTxs txids txs) q) server' client'
 
-    directSender q (CollectPipelined (Just server') _) client =
+    directSender q (CollectPipelined (Just server) _) client = do
+      server' <- server
       directSender q server' client
 
     directSender (ConsQ c q) (CollectPipelined _ collect) client = do

ouroboros-network-protocols/testlib/Ouroboros/Network/Protocol/TxSubmission2/Examples.hs

@@ -272,7 +272,7 @@ txSubmissionServer tracer txId maxUnacked maxTxIdsToRequest maxTxToRequest =
         --
       | canRequestMoreTxs st
       = CollectPipelined
-          (Just (serverReqTxs accum (Succ n) st))
+          (Just (pure $ serverReqTxs accum (Succ n) st))
           (handleReply accum n st)
 
         -- In this case there is nothing else to do so we block until we

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

@@ -18,6 +18,7 @@ import Prelude hiding (seq)
 
 import NoThunks.Class
 
+import Control.Concurrent.Class.MonadMVar (MonadMVar)
 import Control.Concurrent.Class.MonadSTM
 import Control.Exception (SomeException (..), assert)
 import Control.Monad.Class.MonadAsync
@@ -265,6 +266,7 @@ txSubmissionSimulation
      , MonadDelay m
      , MonadFork  m
      , MonadMask  m
+     , MonadMVar  m
      , MonadSay   m
      , MonadST    m
      , MonadSTM   m
@@ -279,13 +281,14 @@ txSubmissionSimulation
 
      , txid ~ Int
      )
-  => NumTxIdsToAck
+  => Tracer m (String, TraceSendRecv (TxSubmission2 txid (Tx txid)))
+  -> NumTxIdsToAck
   -> [Tx txid]
   -> ControlMessageSTM m
   -> Maybe DiffTime
   -> Maybe DiffTime
   -> m ([Tx txid], [Tx txid])
-txSubmissionSimulation maxUnacked outboundTxs
+txSubmissionSimulation tracer maxUnacked outboundTxs
                        controlMessageSTM
                        inboundDelay outboundDelay = do
 
@@ -294,7 +297,7 @@ txSubmissionSimulation maxUnacked outboundTxs
     (outboundChannel, inboundChannel) <- createConnectedChannels
     outboundAsync <-
       async $ runPeerWithLimits
-                (("OUTBOUND",) `contramap` verboseTracer)
+                (("OUTBOUND",) `contramap` tracer)
                 txSubmissionCodec2
                 (byteLimitsTxSubmission2 (fromIntegral . BSL.length))
                 timeLimitsTxSubmission2
@@ -361,6 +364,7 @@ prop_txSubmission (Positive maxUnacked) (NonEmpty outboundTxs) delay =
                     * realToFrac (length outboundTxs `div` 4))
                 atomically (writeTVar controlMessageVar Terminate)
             txSubmissionSimulation
+              verboseTracer
               (NumTxIdsToAck maxUnacked) outboundTxs
               (readTVar controlMessageVar)
               mbDelayTime mbDelayTime

ouroboros-network/src/Ouroboros/Network/TxSubmission/Inbound.hs

@@ -252,7 +252,7 @@ txSubmissionInbound tracer (NumTxIdsToAck maxUnacked) mpReader mpWriter _version
             --
             traceWith tracer (TraceTxInboundCanRequestMoreTxs (natToInt n))
             pure $ CollectPipelined
-              (Just (continueWithState (serverReqTxs (Succ n')) st))
+              (Just (pure $ continueWithState (serverReqTxs (Succ n')) st))
               (collectAndContinueWithState (handleReply n') st)
 
           else do