feat(simulation): support disabling the TCP congestion control modeling (#186)

Author: wenkokke

data/simulation/config-idealised.yaml

@@ -6,4 +6,4 @@
 
 relay-strategy: "request-from-first"
 tcp-congestion-control: false
-multiplex-mini-protocols: false
+multiplex-mini-protocols: true

simulation/ouroboros-leios-sim.cabal

@@ -46,6 +46,7 @@ library
     Chan.Driver
     Chan.Mux
     Chan.TCP
+    Chan.Simple
     Diffusion
     ExamplesLayout
     ExamplesRelay

simulation/src/Chan.hs

@@ -21,6 +21,7 @@ module Chan (
 import Chan.Core
 import Chan.Driver
 import Chan.Mux
+import Chan.Simple (SimpleConnProps (..), newConnectionSimple)
 import Chan.TCP
 import Control.Concurrent.Class.MonadMVar (MonadMVar (..))
 import Control.Monad.Class.MonadAsync (MonadAsync)
@@ -40,11 +41,11 @@ mkConnectionConfig tcp mux tcpLatency maybeTcpBandwidth = ConnectionConfig{..}
  where
   transportConfig
     | tcp = TransportTcp (mkTcpConnProps tcpLatency (fromMaybe defaultTcpBandwidth maybeTcpBandwidth))
-    | otherwise = TransportBasic
+    | otherwise = TransportSimple (SimpleConnProps tcpLatency maybeTcpBandwidth)
   defaultTcpBandwidth = (kilobytes 1000)
 
 data TransportConfig
-  = TransportBasic
+  = TransportSimple !SimpleConnProps
   | TransportTcp !TcpConnProps
 
 newConnectionBundle ::
@@ -54,22 +55,15 @@ newConnectionBundle ::
   ConnectionConfig ->
   m (bundle (Chan m), bundle (Chan m))
 newConnectionBundle tracer = \case
-  ConnectionConfig TransportBasic _mux@False ->
+  ConnectionConfig (TransportSimple _simpleConnProps) _mux@False ->
     error "Unsupported configuration (no TCP, no mux)"
-  ConnectionConfig TransportBasic _mux@True ->
-    error "Unsupported configuration (no TCP)"
+  ConnectionConfig (TransportSimple simpleConnProps) _mux@True -> do
+    let tracer' = contramap ((fmap . fmap) fromBearerMsg) tracer
+    (mA, mB) <- newConnectionSimple tracer' simpleConnProps
+    (,) <$> newMuxChan mA <*> newMuxChan mB
   ConnectionConfig (TransportTcp _tcpConnProps) _mux@False ->
     error "Unsupported configuration (no mux)"
-  ConnectionConfig (TransportTcp tcpConnProps) _mux@True ->
-    newConnectionBundleTCPMux tracer tcpConnProps
-
-newConnectionBundleTCPMux ::
-  forall bundle m.
-  (ConnectionBundle bundle, MonadTime m, MonadMonotonicTimeNSec m, MonadDelay m, MonadAsync m, MessageSize (BundleMsg bundle), MonadMVar m, MonadFork m) =>
-  Tracer m (LabelTcpDir (TcpEvent (BundleMsg bundle))) ->
-  TcpConnProps ->
-  m (bundle (Chan m), bundle (Chan m))
-newConnectionBundleTCPMux tracer tcpprops = do
-  let tracer' = contramap ((fmap . fmap) fromBearerMsg) tracer
-  (mA, mB) <- newConnectionTCP tracer' tcpprops
-  (,) <$> newMuxChan mA <*> newMuxChan mB
+  ConnectionConfig (TransportTcp tcpConnProps) _mux@True -> do
+    let tracer' = contramap ((fmap . fmap) fromBearerMsg) tracer
+    (mA, mB) <- newConnectionTCP tracer' tcpConnProps
+    (,) <$> newMuxChan mA <*> newMuxChan mB

simulation/src/Chan/Mux.hs

@@ -28,12 +28,7 @@ import STMCompat
 class ConnectionBundle bundle where
   type BundleMsg bundle
   toFromBundleMsgBundle :: bundle (ToFromBundleMsg (BundleMsg bundle))
-
-  traverseConnectionBundle ::
-    Monad m =>
-    (forall a. f a -> m (g a)) ->
-    bundle f ->
-    m (bundle g)
+  traverseConnectionBundle :: Monad m => (forall a. f a -> m (g a)) -> bundle f -> m (bundle g)
 
 -- | Injection, projection, between a common mux message type, and an
 -- individual message type. The following must hold:

simulation/src/Chan/Simple.hs

@@ -0,0 +1,86 @@
+{-# LANGUAGE RecordWildCards #-}
+
+module Chan.Simple (
+  SimpleConnProps (..),
+  newConnectionSimple,
+) where
+
+import Chan.Core (Chan (..))
+import Chan.TCP (LabelTcpDir (..), MessageSize (..), RecvBuf, SendBuf, TcpEvent (..), TcpMsgForecast (..), mkChan, newRecvBuf, newSendBuf)
+import Control.Monad.Class.MonadAsync (MonadAsync (..))
+import Control.Tracer as Tracer (Contravariant (..), Tracer, traceWith)
+import ModelTCP (Bytes)
+import STMCompat
+import TimeCompat
+
+data SimpleConnProps = SimpleConnProps
+  { latency :: DiffTime
+  , bandwidthBytesPerSecond :: Maybe Bytes
+  }
+
+newConnectionSimple ::
+  forall m a.
+  (MonadTime m, MonadMonotonicTimeNSec m, MonadDelay m, MonadAsync m, MessageSize a) =>
+  Tracer m (LabelTcpDir (TcpEvent a)) ->
+  SimpleConnProps ->
+  m (Chan m a, Chan m a)
+newConnectionSimple tracer simpleConnProps = do
+  clientSendBuf <- newSendBuf
+  serverSendBuf <- newSendBuf
+  clientRecvBuf <- newRecvBuf
+  serverRecvBuf <- newRecvBuf
+
+  _ <-
+    async
+      ( transport
+          (contramap DirClientToServer tracer)
+          simpleConnProps
+          clientSendBuf
+          serverRecvBuf
+      )
+  _ <-
+    async
+      ( transport
+          (contramap DirServerToClient tracer)
+          simpleConnProps
+          serverSendBuf
+          clientRecvBuf
+      )
+
+  let clientChan, serverChan :: Chan m a
+      clientChan = mkChan clientSendBuf clientRecvBuf
+      serverChan = mkChan serverSendBuf serverRecvBuf
+
+  return (clientChan, serverChan)
+
+transport ::
+  (MonadSTM m, MonadMonotonicTimeNSec m, MonadDelay m, MessageSize a) =>
+  Tracer m (TcpEvent a) ->
+  SimpleConnProps ->
+  SendBuf m a ->
+  RecvBuf m a ->
+  m ()
+transport tracer SimpleConnProps{..} sendbuf recvbuf = go
+ where
+  go = do
+    msg <- atomically $ readTMVar sendbuf -- read now but keep buffer full
+    now <- getMonotonicTime
+
+    let msgSize = messageSizeBytes msg
+        msgSerialisationTime = maybe 0 (fromIntegral . (msgSize `div`)) bandwidthBytesPerSecond
+        msgSendLeadingEdge = now
+        msgSendTrailingEdge = msgSerialisationTime `addTime` now
+        msgRecvLeadingEdge = latency `addTime` now
+        msgRecvTrailingEdge = latency `addTime` msgSendTrailingEdge
+        msgAcknowledgement = latency `addTime` msgRecvTrailingEdge
+    let msgForecast = TcpMsgForecast{..}
+
+    -- schedule the arrival, and wait until it has finished sending
+    atomically $ writeTQueue recvbuf (msgRecvTrailingEdge, msg)
+    traceWith tracer (TcpSendMsg msg msgForecast [])
+    threadDelay (msgSendTrailingEdge `diffTime` now)
+    -- We keep the sendbuf full until the message has finished sending
+    -- so that there's less buffering, and better simulates the TCP buffer
+    -- rather than an extra app-level buffer.
+    _ <- atomically $ takeTMVar sendbuf
+    go

simulation/src/Chan/TCP.hs

@@ -3,12 +3,8 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Chan.TCP (
-  newConnectionTCP,
-  MessageSize (..),
-  Bytes,
-  LabelTcpDir (..),
-  TcpEvent (..),
-  TcpConnProps (..),
+  module Chan.TCP,
+  module ModelTCP,
 ) where
 
 import Chan.Core (Chan (..))