RTT-aware mux

cabal.project

@@ -30,6 +30,7 @@ packages: ./monoidal-synchronisation
           ./cardano-diffusion
           ./ntp-client
           ./acts-generic
+          ./window-stats
 
 tests: True
 benchmarks: True

cardano-diffusion/demo/chain-sync.hs

@@ -35,8 +35,7 @@ import Control.Monad.Class.MonadTime.SI (Time (..))
 import Control.Tracer
 
 import System.Directory
-import System.Random (RandomGen, SplitGen, StdGen)
-import System.Random qualified as Random
+import System.Random
 
 import Options.Applicative qualified as Opts
 
@@ -228,6 +227,7 @@ clientChainSync :: [FilePath]
 clientChainSync sockPaths maxSlotNo = withIOManager $ \iocp ->
     forConcurrently_ (zip [0..] sockPaths) $ \(index, sockPath) -> do
       threadDelay (50000 * index)
+      rttCookieSeed <- newStdGen
       void $ connectToNode
         (localSnocket iocp)
         makeLocalBearer
@@ -236,7 +236,8 @@ clientChainSync sockPaths maxSlotNo = withIOManager $ \iocp ->
           ctaHandshakeTimeLimits = noTimeLimitsHandshake,
           ctaVersionDataCodec    = unversionedProtocolDataCodec,
           ctaConnectTracers      = nullNetworkConnectTracers,
-          ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
+          ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion,
+          ctaRTTCookieSeed       = rttCookieSeed
         }
         mempty
         (simpleSingletonVersions
@@ -263,8 +264,8 @@ serverChainSync :: FilePath
                 -> IO Void
 serverChainSync sockAddr slotLength seed = withIOManager $ \iocp -> do
     prng <- case seed of
-      Nothing -> Random.initStdGen
-      Just a  -> return (Random.mkStdGen a)
+      Nothing -> initStdGen
+      Just a  -> return (mkStdGen a)
     Server.Simple.with
       (localSnocket iocp)
       nullTracer
@@ -479,16 +480,18 @@ clientBlockFetch sockAddrs maxSlotNo = withIOManager $ \iocp -> do
           chainSelection fingerprint'
 
     peerAsyncs <- sequence
-                    [ async . void $
-                        connectToNode
+                    [ async $ do
+                        rttCookieSeed <- newStdGen
+                        void $ connectToNode
                           (localSnocket iocp)
                           makeLocalBearer
                           ConnectToArgs {
                             ctaHandshakeCodec      = unversionedHandshakeCodec,
                             ctaHandshakeTimeLimits = noTimeLimitsHandshake,
                             ctaVersionDataCodec    = unversionedProtocolDataCodec,
                             ctaConnectTracers      = nullNetworkConnectTracers,
-                            ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
+                            ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion,
+                            ctaRTTCookieSeed       = rttCookieSeed
                           }
                           mempty
                           (simpleSingletonVersions
@@ -543,8 +546,8 @@ serverBlockFetch :: FilePath
                  -> IO Void
 serverBlockFetch sockAddr slotLength seed = withIOManager $ \iocp -> do
     prng <- case seed of
-      Nothing -> Random.initStdGen
-      Just a  -> return (Random.mkStdGen a)
+      Nothing -> initStdGen
+      Just a  -> return (mkStdGen a)
     Server.Simple.with
       (localSnocket iocp)
       nullTracer
@@ -792,15 +795,15 @@ genBlockChain !g prevHeader =
     block :< genBlockChain g'' (Just (blockHeader block))
   where
     block     = genBlock g' prevHeader
-    (g', g'') = Random.splitGen g
+    (g', g'') = splitGen g
 
 genBlock :: SplitGen g => g -> Maybe BlockHeader -> Block
 genBlock g prevHeader =
     Block { blockBody, blockHeader }
   where
     blockBody   = genBlockBody g'
     blockHeader = genBlockHeader g'' prevHeader blockBody
-    (g', g'')   = Random.splitGen g
+    (g', g'')   = splitGen g
 
 genBlockHeader :: RandomGen g
                => g -> Maybe BlockHeader -> BlockBody -> BlockHeader
@@ -814,7 +817,7 @@ genBlockHeader g prevHeader body =
       headerBlockNo  = maybe 1 (succ             . headerBlockNo) prevHeader,
       headerBodyHash = hashBody body
     }
-    (slotGap, _) = Random.randomR (1,3) g
+    (slotGap, _) = randomR (1,3) g
 
     addSlotGap :: Int -> SlotNo -> SlotNo
     addSlotGap m (SlotNo n) = SlotNo (n + fromIntegral m)
@@ -823,8 +826,8 @@ genBlockBody :: RandomGen g => g -> BlockBody
 genBlockBody g =
     BlockBody . BSC.take len . BSC.drop offset . BSC.pack $ bodyData
   where
-    (offset, g') = Random.randomR (0, bodyDataCycle-1)    g
-    (len   , _ ) = Random.randomR (1, bodyDataCycle*10-1) g'
+    (offset, g') = randomR (0, bodyDataCycle-1)    g
+    (len   , _ ) = randomR (1, bodyDataCycle*10-1) g'
 
 bodyData :: String
 bodyData = concat

cardano-diffusion/lib/Cardano/Network/NodeToClient.hs

@@ -68,6 +68,7 @@ import Control.Monad.Class.MonadTimer.SI
 import Data.ByteString.Lazy qualified as BL
 import Data.Kind (Type)
 import Data.Void (Void, absurd)
+import System.Random
 
 import Network.Mux qualified as Mx
 import Network.TypedProtocol.Peer.Client
@@ -234,22 +235,24 @@ connectTo
   -> FilePath
   -- ^ path of the unix socket or named pipe
   -> IO (Either SomeException a)
-connectTo snocket tracers versions path =
+connectTo snocket tracers versions path = do
+    rttCookieSeed <- newStdGen
     fmap fn <$>
-    connectToNode
-      snocket
-      makeLocalBearer
-      ConnectToArgs {
-        ctaHandshakeCodec      = nodeToClientHandshakeCodec,
-        ctaHandshakeTimeLimits = noTimeLimitsHandshake,
-        ctaVersionDataCodec    = nodeToClientVersionDataCodec,
-        ctaConnectTracers      = tracers,
-        ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
-      }
-      mempty
-      versions
-      Nothing
-      (localAddressFromPath path)
+      connectToNode
+        snocket
+        makeLocalBearer
+        ConnectToArgs {
+          ctaHandshakeCodec      = nodeToClientHandshakeCodec,
+          ctaHandshakeTimeLimits = noTimeLimitsHandshake,
+          ctaVersionDataCodec    = nodeToClientVersionDataCodec,
+          ctaConnectTracers      = tracers,
+          ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion,
+          ctaRTTCookieSeed       = rttCookieSeed
+        }
+        mempty
+        versions
+        Nothing
+        (localAddressFromPath path)
   where
     fn :: forall x. Either x Void -> x
     fn = either id absurd
@@ -284,7 +287,8 @@ connectToWithMux
   --
   -- NOTE: when the callback returns or errors, the mux thread will be killed.
   -> IO x
-connectToWithMux snocket tracers versions path k =
+connectToWithMux snocket tracers versions path k = do
+  rttCookieSeed <- newStdGen
   connectToNodeWithMux
     snocket
     makeLocalBearer
@@ -293,7 +297,8 @@ connectToWithMux snocket tracers versions path k =
       ctaHandshakeTimeLimits = noTimeLimitsHandshake,
       ctaVersionDataCodec    = nodeToClientVersionDataCodec,
       ctaConnectTracers      = tracers,
-      ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
+      ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion,
+      ctaRTTCookieSeed       = rttCookieSeed
     }
     mempty
     versions

cardano-diffusion/lib/Cardano/Network/NodeToNode.hs

@@ -82,6 +82,7 @@ import Control.Exception (SomeException)
 import Data.ByteString.Lazy qualified as BL
 import Data.Set (Set)
 import Data.Word
+import System.Random
 
 import Network.Mux qualified as Mx
 #if !defined(wasm32_HOST_ARCH)
@@ -152,9 +153,9 @@ data NodeToNodeProtocols appType initiatorCtx responderCtx bytes m a b = NodeToN
 }
 
 type NodeToNodeProtocolsWithExpandedCtx appType ntnAddr bytes m a b =
-    NodeToNodeProtocols appType (ExpandedInitiatorContext ntnAddr PeerTrustable m) (ResponderContext ntnAddr) bytes m a b
+    NodeToNodeProtocols appType (ExpandedInitiatorContext ntnAddr PeerTrustable m) (ResponderContext ntnAddr m) bytes m a b
 type NodeToNodeProtocolsWithMinimalCtx  appType ntnAddr bytes m a b =
-    NodeToNodeProtocols appType (MinimalInitiatorContext ntnAddr)  (ResponderContext ntnAddr) bytes m a b
+    NodeToNodeProtocols appType (MinimalInitiatorContext ntnAddr)  (ResponderContext ntnAddr m) bytes m a b
 
 
 data MiniProtocolParameters = MiniProtocolParameters {
@@ -509,16 +510,18 @@ connectTo
   -> Socket.SockAddr
   -> IO (Either SomeException (Either a b))
 #if !defined(wasm32_HOST_ARCH)
-connectTo sn tr =
+connectTo sn tr versions localAddr remoteAddr = do
+    rttCookieSeed <- newStdGen
     connectToNode sn makeSocketBearer
                   ConnectToArgs {
                     ctaHandshakeCodec      = nodeToNodeHandshakeCodec,
                     ctaHandshakeTimeLimits = timeLimitsHandshake,
                     ctaVersionDataCodec    = nodeToNodeVersionDataCodec,
                     ctaConnectTracers      = tr,
-                    ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion
+                    ctaHandshakeCallbacks  = HandshakeCallbacks acceptableVersion queryVersion,
+                    ctaRTTCookieSeed       = rttCookieSeed
                   }
-                  configureOutboundSocket
+                  configureOutboundSocket versions localAddr remoteAddr
   where
     configureOutboundSocket :: Socket -> IO ()
     configureOutboundSocket sock = do

cardano-diffusion/ping/Cardano/Network/Ping.hs

@@ -1202,6 +1202,7 @@ pingClient' stdout infoTracer headerTracer stderr opts@PingOpts{..} signalVar ad
                   stdGen <- initStdGen
                   mx <- Mx.new
                           Mx.nullTracers
+                          stdGen
                           [MiniProtocolInfo {
                             miniProtocolNum        = case protocol of
                                                        NodeToNode -> NodeToNode.chainSyncMiniProtocolNum
@@ -1241,8 +1242,10 @@ pingClient' stdout infoTracer headerTracer stderr opts@PingOpts{..} signalVar ad
                   --
                   -- run keepalive client to get RTT samples
                   --
+                  g  <- initStdGen
                   mx <- Mx.new
                           Mx.nullTracers
+                          g
                           [MiniProtocolInfo {
                             miniProtocolNum        = NodeToNode.keepAliveMiniProtocolNum,
                             miniProtocolDir        = Mx.InitiatorDirectionOnly,

cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs

@@ -491,7 +491,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
                           )
 
     chainSyncResponder
-      :: MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      :: MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     chainSyncResponder = MiniProtocolCb $
       \ ResponderContext { rcConnectionId = connId }
         channel -> do
@@ -533,7 +533,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
                                      nullTracer clientCtx)
 
     blockFetchResponder
-      :: MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      :: MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     blockFetchResponder =
       MiniProtocolCb $
         \ ResponderContext { rcConnectionId = connId }
@@ -587,7 +587,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
                                    kacApp
 
     keepAliveResponder
-      :: MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      :: MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     keepAliveResponder = MiniProtocolCb $
       \ ResponderContext { rcConnectionId = connId }
         channel -> do
@@ -644,7 +644,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
                (pingPongClientPeer pingPongClient)
 
     pingPongResponder
-      :: MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      :: MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     pingPongResponder  = MiniProtocolCb $
       \ResponderContext { rcConnectionId = connId } channel ->
       runPeerWithLimits
@@ -679,7 +679,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
 
     peerSharingResponder
       :: PeerSharingAPI NtNAddr s m
-      -> MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      -> MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     peerSharingResponder psAPI = MiniProtocolCb $
       \ ResponderContext { rcConnectionId = connId }
         channel -> do
@@ -734,7 +734,7 @@ applications debugTracer txSubmissionInboundTracer nodeKernel
       -> TxSubmissionCountersVar m
       -> SharedTxStateVar m NtNAddr Int
       -> PeerTxRegistry m NtNAddr
-      -> MiniProtocolCb (ResponderContext NtNAddr) ByteString m ()
+      -> MiniProtocolCb (ResponderContext NtNAddr m) ByteString m ()
     txSubmissionResponder mempool txCountersVar sharedTxStateVar inFlightRegistry =
       MiniProtocolCb $
         \ ResponderContext { rcConnectionId = connId@ConnectionId { remoteAddress = them }} channel

network-mux/bench/socket_read_write/Main.hs

@@ -25,9 +25,12 @@ import Network.Mux
 import Network.Mux.Bearer
 import Network.Mux.Egress
 import Network.Mux.Ingress
-import Network.Mux.Timeout (withTimeoutSerial)
+import Network.Mux.RTT
+import Network.Mux.Timeout
 import Network.Mux.Types
 
+import System.Random
+
 activeTracer :: Tracer IO a
 activeTracer = nullTracer
 --activeTracer = show >$< stdoutTracer
@@ -82,7 +85,8 @@ readDemuxerQueueBenchmark sndSizeV sndSize addr = do
       withReadBufferIO (\buffer -> do
         bearer <- getBearer makeSocketBearer sduTimeout sd buffer
         ms42 <- mkMiniProtocolState 42
-        withAsync (demuxer [ms42] activeTracer bearer) $ \aid -> do
+        rtt  <- newRTTState (mkStdGen 0)
+        withAsync (demuxer [ms42] rtt activeTracer bearer) $ \aid -> do
           doRead 0xa5 (totalPayloadLen sndSize) (miniProtocolIngressQueue ms42)
           cancel aid
        )
@@ -115,7 +119,8 @@ readDemuxerBenchmark sndSizeV sndSize addr = do
         bearer <- getBearer makeSocketBearer sduTimeout sd buffer
         ms42 <- mkMiniProtocolState 42
         ms41 <- mkMiniProtocolState 41
-        withAsync (demuxer [ms41, ms42] activeTracer bearer) $ \aid -> do
+        rtt  <- newRTTState (mkStdGen 0)
+        withAsync (demuxer [ms41, ms42] rtt activeTracer bearer) $ \aid -> do
           withAsync (doRead 42 (totalPayloadLen sndSize) (miniProtocolIngressQueue ms42) 0) $ \aid42 -> do
             withAsync (doRead 41 (totalPayloadLen 10) (miniProtocolIngressQueue ms41) 0) $ \aid41 -> do
               _ <- waitBoth aid42 aid41
@@ -190,10 +195,11 @@ startServerMany sndSizeV ad = forever $ do
   wrap blob = SDU {
         -- it will be filled when the 'SDU' is send by the 'bearer'
        msHeader = SDUHeader {
-           mhTimestamp = RemoteClockModel 0,
-           mhNum       = MiniProtocolNum 42,
-           mhDir       = ResponderDir,
-           mhLength    = fromIntegral $ BL.length blob
+           mhSendCookie = noCookie,
+           mhEchoCookie = noCookie,
+           mhNum        = MiniProtocolNum 42,
+           mhDir        = ResponderDir,
+           mhLength     = fromIntegral $ BL.length blob
           },
         msBlob = blob
      }
@@ -217,7 +223,8 @@ startServerEgresss pollInterval sndSizeV ad = forever $ do
           numberOfCalls = numberOfSdus `div` 10 :: Int
           runtSdus = numberOfSdus `mod` 10 :: Int
 
-      withAsync (muxer eq activeTracer bearer) $ \aid -> do
+      rtt <- newRTTState (mkStdGen 0)
+      withAsync (muxer eq rtt activeTracer bearer) $ \aid -> do
 
         replicateM_ numberOfCalls $ do
           let payload42s = replicate 10 $ BL.replicate sndSize 42

network-mux/demo/mux-demo.hs

@@ -24,6 +24,7 @@ import Control.Tracer (Tracer, mkTracer)
 import System.Environment qualified as SysEnv
 import System.Exit
 import System.IO
+import System.Random
 
 #if defined(mingw32_HOST_OS)
 import Data.Bits
@@ -127,7 +128,8 @@ server = do
 
 serverWorker :: Bearer IO -> IO ()
 serverWorker bearer = do
-    mux <- Mx.new Mx.nullTracers ptcls
+    g   <- newStdGen
+    mux <- Mx.new Mx.nullTracers g ptcls
 
     void $ forkIO $ do
       awaitResult <-
@@ -187,7 +189,8 @@ client n msg = do
 
 clientWorker :: Mx.Bearer IO -> Int -> String -> IO ()
 clientWorker bearer n msg = do
-    mux <- Mx.new Mx.nullTracers ptcls
+    g   <- newStdGen
+    mux <- Mx.new Mx.nullTracers g ptcls
 
     void $ forkIO $ do
       awaitResult <-