diff --git a/cardano-diffusion/changelog.d/20260330_134910_crocodile-dentist_mux_single_peer_performance.md b/cardano-diffusion/changelog.d/20260330_134910_crocodile-dentist_mux_single_peer_performance.md new file mode 100644 index 00000000000..8dc740b9aa4 --- /dev/null +++ b/cardano-diffusion/changelog.d/20260330_134910_crocodile-dentist_mux_single_peer_performance.md @@ -0,0 +1,23 @@ + + + +### Non-Breaking + +- Integrate weighted fair queue + burst mux + + diff --git a/cardano-diffusion/demo/chain-sync.hs b/cardano-diffusion/demo/chain-sync.hs index 32efff35323..2d53c0344c6 100644 --- a/cardano-diffusion/demo/chain-sync.hs +++ b/cardano-diffusion/demo/chain-sync.hs @@ -209,7 +209,8 @@ rmIfExists path = do maximumMiniProtocolLimits :: MiniProtocolLimits maximumMiniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = maxBound + maximumIngressQueue = maxBound, + burst = Nothing } @@ -226,7 +227,8 @@ demoProtocol2 chainSync = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = chainSync + miniProtocolRun = chainSync, + miniProtocolWeight = 1 } ] @@ -336,13 +338,15 @@ demoProtocol3 chainSync blockFetch = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = chainSync + miniProtocolRun = chainSync, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = MiniProtocolNum 3, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = blockFetch + miniProtocolRun = blockFetch, + miniProtocolWeight = 1 } ] diff --git a/cardano-diffusion/lib/Cardano/Network/NodeToClient.hs b/cardano-diffusion/lib/Cardano/Network/NodeToClient.hs index a180c3ea8c0..56e51d5a0d9 100644 --- a/cardano-diffusion/lib/Cardano/Network/NodeToClient.hs +++ b/cardano-diffusion/lib/Cardano/Network/NodeToClient.hs @@ -52,8 +52,8 @@ module Cardano.Network.NodeToClient , Handshake ) where -import Control.Exception (SomeException) import Control.DeepSeq (NFData) +import Control.Exception (SomeException) import Control.Monad (forever) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadTimer.SI @@ -149,35 +149,40 @@ nodeToClientProtocols protocols _version _versionData = miniProtocolNum = MiniProtocolNum 5, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = localChainSyncProtocol + miniProtocolRun = localChainSyncProtocol, + miniProtocolWeight = 1 } localTxSubmissionMiniProtocol localTxSubmissionProtocol = MiniProtocol { miniProtocolNum = MiniProtocolNum 6, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = localTxSubmissionProtocol + miniProtocolRun = localTxSubmissionProtocol, + miniProtocolWeight = 1 } localStateQueryMiniProtocol localStateQueryProtocol = MiniProtocol { miniProtocolNum = MiniProtocolNum 7, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = localStateQueryProtocol + miniProtocolRun = localStateQueryProtocol, + miniProtocolWeight = 1 } localTxMonitorMiniProtocol localTxMonitorProtocol = MiniProtocol { miniProtocolNum = MiniProtocolNum 9, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = localTxMonitorProtocol + miniProtocolRun = localTxMonitorProtocol, + miniProtocolWeight = 1 } maximumMiniProtocolLimits :: MiniProtocolLimits maximumMiniProtocolLimits = MiniProtocolLimits { #if !defined(wasm32_HOST_ARCH) - maximumIngressQueue = 0xffffffff + maximumIngressQueue = 0xffffffff, #else - maximumIngressQueue = 0x7fffffff + maximumIngressQueue = 0x7fffffff, #endif + burst = Nothing } diff --git a/cardano-diffusion/lib/Cardano/Network/NodeToNode.hs b/cardano-diffusion/lib/Cardano/Network/NodeToNode.hs index 3427f01fe86..40fd18eaee6 100644 --- a/cardano-diffusion/lib/Cardano/Network/NodeToNode.hs +++ b/cardano-diffusion/lib/Cardano/Network/NodeToNode.hs @@ -263,19 +263,22 @@ nodeToNodeProtocols _featureFlags miniProtocolParameters protocols miniProtocolNum = chainSyncMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = chainSyncProtocolLimits miniProtocolParameters, - miniProtocolRun = chainSyncProtocol + miniProtocolRun = chainSyncProtocol, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = blockFetchMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = blockFetchProtocolLimits miniProtocolParameters, - miniProtocolRun = blockFetchProtocol + miniProtocolRun = blockFetchProtocol, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = txSubmissionMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = txSubmissionProtocolLimits miniProtocolParameters, - miniProtocolRun = txSubmissionProtocol + miniProtocolRun = txSubmissionProtocol, + miniProtocolWeight = 1 } ] <> case perasSupport of @@ -287,13 +290,15 @@ nodeToNodeProtocols _featureFlags miniProtocolParameters protocols miniProtocolNum = perasCertDiffusionMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = perasCertDiffusionProtocolLimits miniProtocolParameters, - miniProtocolRun = perasCertDiffusionProtocol + miniProtocolRun = perasCertDiffusionProtocol, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = perasVoteDiffusionMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = perasVoteDiffusionProtocolLimits miniProtocolParameters, - miniProtocolRun = perasVoteDiffusionProtocol + miniProtocolRun = perasVoteDiffusionProtocol, + miniProtocolWeight = 1 } ]) @@ -309,7 +314,8 @@ nodeToNodeProtocols _featureFlags miniProtocolParameters protocols miniProtocolNum = keepAliveMiniProtocolNum, miniProtocolStart = StartOnDemandAny, miniProtocolLimits = keepAliveProtocolLimits miniProtocolParameters, - miniProtocolRun = keepAliveProtocol + miniProtocolRun = keepAliveProtocol, + miniProtocolWeight = 1 } : case peerSharing of PeerSharingEnabled -> @@ -317,7 +323,8 @@ nodeToNodeProtocols _featureFlags miniProtocolParameters protocols miniProtocolNum = peerSharingMiniProtocolNum, miniProtocolStart = StartOnDemand, miniProtocolLimits = peerSharingProtocolLimits miniProtocolParameters, - miniProtocolRun = peerSharingProtocol + miniProtocolRun = peerSharingProtocol, + miniProtocolWeight = 1 } ] PeerSharingDisabled -> @@ -342,7 +349,8 @@ chainSyncProtocolLimits MiniProtocolParameters { chainSyncPipeliningHighMark } = -- TODO: 1400 comes from maxBlockHeaderSize in genesis, but should come -- from consensus rather than being hard coded. maximumIngressQueue = addSafetyMargin $ - fromIntegral chainSyncPipeliningHighMark * 1400 + fromIntegral chainSyncPipeliningHighMark * 1400, + burst = Nothing } blockFetchProtocolLimits MiniProtocolParameters { blockFetchPipeliningMax } = MiniProtocolLimits { @@ -364,7 +372,8 @@ blockFetchProtocolLimits MiniProtocolParameters { blockFetchPipeliningMax } = Mi -- relaxed limit here. -- maximumIngressQueue = addSafetyMargin $ - max (10 * 2_097_154 :: Int) (fromIntegral blockFetchPipeliningMax * 90_112) + max (10 * 2_097_154 :: Int) (fromIntegral blockFetchPipeliningMax * 90_112), + burst = Just $ Mx.ProtocolBurst 90_112 10_000 } txSubmissionProtocolLimits MiniProtocolParameters @@ -432,13 +441,15 @@ txSubmissionProtocolLimits MiniProtocolParameters -- 10% as a safety margin. -- maximumIngressQueue = addSafetyMargin $ - fromIntegral maxUnacknowledgedTxIds * (44 + fromIntegral @SizeInBytes @Int max_TX_SIZE) + fromIntegral maxUnacknowledgedTxIds * (44 + fromIntegral @SizeInBytes @Int max_TX_SIZE), + burst = Nothing } keepAliveProtocolLimits _ = MiniProtocolLimits { -- One small outstanding message. - maximumIngressQueue = addSafetyMargin 1280 + maximumIngressQueue = addSafetyMargin 1280, + burst = Nothing } peerSharingProtocolLimits _ = @@ -449,7 +460,8 @@ peerSharingProtocolLimits _ = -- window size of 4 and a TCP segment is 1440, which gives us 4 * 1440 = -- 5760 bytes to fit into a single RTT. So setting the maximum ingress -- queue to be a single RTT should be enough to cover for CBOR overhead. - maximumIngressQueue = 4 * 1440 + maximumIngressQueue = 4 * 1440, + burst = Nothing } perasCertDiffusionProtocolLimits MiniProtocolParameters { perasCertDiffusionMaxObjectsUnacknowledged } = @@ -460,7 +472,8 @@ perasCertDiffusionProtocolLimits MiniProtocolParameters { perasCertDiffusionMaxO -- even much smaller. -- See https://github.com/tweag/cardano-peras/issues/97 maximumIngressQueue = addSafetyMargin $ - fromIntegral perasCertDiffusionMaxObjectsUnacknowledged * 20_000 + fromIntegral perasCertDiffusionMaxObjectsUnacknowledged * 20_000, + burst = Nothing } perasVoteDiffusionProtocolLimits MiniProtocolParameters { perasVoteDiffusionMaxObjectsUnacknowledged } = @@ -469,7 +482,8 @@ perasVoteDiffusionProtocolLimits MiniProtocolParameters { perasVoteDiffusionMaxO -- We assume an upper bound of 1 kB per vote. -- See https://github.com/tweag/cardano-peras/issues/97 maximumIngressQueue = addSafetyMargin $ - fromIntegral perasVoteDiffusionMaxObjectsUnacknowledged * 1_000 + fromIntegral perasVoteDiffusionMaxObjectsUnacknowledged * 1_000, + burst = Nothing } chainSyncMiniProtocolNum :: MiniProtocolNum diff --git a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs index 9236cf4cf68..48969ca5a95 100644 --- a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs +++ b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/MiniProtocols.hs @@ -340,14 +340,16 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node -> MiniProtocolWithExpandedCtx Mx.InitiatorMode NtNAddr PeerTrustable ByteString m () Void f MiniProtocol { miniProtocolNum , miniProtocolLimits - , miniProtocolRun } = + , miniProtocolRun + , miniProtocolWeight } = MiniProtocol { miniProtocolNum , miniProtocolStart = StartEagerly , miniProtocolLimits , miniProtocolRun = case miniProtocolRun of InitiatorAndResponderProtocol initiator _respnder -> - InitiatorProtocolOnly initiator + InitiatorProtocolOnly initiator, + miniProtocolWeight } initiatorAndResponderApp @@ -362,7 +364,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , miniProtocolRun = InitiatorAndResponderProtocol chainSyncInitiator - chainSyncResponder + chainSyncResponder, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = blockFetchMiniProtocolNum @@ -371,7 +374,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , miniProtocolRun = InitiatorAndResponderProtocol blockFetchInitiator - blockFetchResponder + blockFetchResponder, + miniProtocolWeight = 1 } , MiniProtocol { @@ -384,7 +388,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node (txSubmissionResponder (nkMempool nodeKernel) (nkTxChannelsVar nodeKernel) (nkTxMempoolSem nodeKernel) - (nkSharedTxStateVar nodeKernel)) + (nkSharedTxStateVar nodeKernel)), + miniProtocolWeight = 1 } ] , withWarm = WithWarm @@ -395,7 +400,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , miniProtocolRun = InitiatorAndResponderProtocol pingPongInitiator - pingPongResponder + pingPongResponder, + miniProtocolWeight = 1 } ] , withEstablished = WithEstablished $ @@ -406,7 +412,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , miniProtocolRun = InitiatorAndResponderProtocol keepAliveInitiator - keepAliveResponder + keepAliveResponder, + miniProtocolWeight = 1 } : case peerSharing of PSTypes.PeerSharingEnabled -> @@ -417,7 +424,8 @@ applications debugTracer txSubmissionInboundTracer txSubmissionInboundDebug node , miniProtocolRun = InitiatorAndResponderProtocol peerSharingInitiator - (peerSharingResponder (nkPeerSharingAPI nodeKernel)) + (peerSharingResponder (nkPeerSharingAPI nodeKernel)), + miniProtocolWeight = 1 } ] PSTypes.PeerSharingDisabled -> diff --git a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs index c3ff4b7b20a..de8f3abe313 100644 --- a/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs +++ b/cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet/Simulation.hs @@ -38,11 +38,10 @@ module Test.Cardano.Network.Diffusion.Testnet.Simulation , module PeerSelection ) where -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 (MonadPlus, forM, when) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -1033,8 +1032,7 @@ data Churn = CardanoChurn | OuroborosChurn -- | Run an arbitrary topology in a generic monad `m`. -- diffusionSimulationM - :: forall m. ( Alternative (STM m) - , MonadAsync m + :: forall m. ( MonadAsync m , MonadDelay m , MonadFix m , MonadEvaluate m @@ -1045,6 +1043,7 @@ diffusionSimulationM , MonadLabelledSTM m , MonadTraceSTM m , MonadMask m + , MonadPlus (STM m) , MonadTime m , MonadTimer m , MonadThrow (STM m) @@ -1210,7 +1209,7 @@ diffusionSimulationM acceptVersion = acceptableVersion defaultMiniProtocolsLimit :: MiniProtocolLimits defaultMiniProtocolsLimit = - MiniProtocolLimits { maximumIngressQueue = 64000 } + MiniProtocolLimits { maximumIngressQueue = 64000, burst = Nothing } blockGeneratorArgs :: Node.BlockGeneratorArgs Block StdGen blockGeneratorArgs = diff --git a/network-mux/bench/socket_read_write/Main.hs b/network-mux/bench/socket_read_write/Main.hs index 02a9262ceda..4916e02d07f 100644 --- a/network-mux/bench/socket_read_write/Main.hs +++ b/network-mux/bench/socket_read_write/Main.hs @@ -144,7 +144,7 @@ mkMiniProtocolState num = do mpv <- newTVarIO StatusRunning let mpi = MiniProtocolInfo (MiniProtocolNum num) InitiatorDirectionOnly - (MiniProtocolLimits maxBound) Nothing + (MiniProtocolLimits maxBound Nothing) Nothing 1 return $ MiniProtocolState mpi mpq mpv -- | Run a server that accept connections on `ad`. @@ -208,7 +208,8 @@ startServerEgresss pollInterval sndSizeV ad = forever $ do withReadBufferIO (\buffer -> do bearer <- getBearer (makeSocketBearer' pollInterval) sduTimeout sd buffer sndSize <- atomically $ takeTMVar sndSizeV - eq <- atomically $ newTBQueue 100 + eq' <- atomically $ newTBQueue 100 + let eq = [(1, eq')] w42 <- newTVarIO BL.empty w41 <- newTVarIO BL.empty @@ -222,13 +223,13 @@ startServerEgresss pollInterval sndSizeV ad = forever $ do replicateM_ numberOfCalls $ do let payload42s = replicate 10 $ BL.replicate sndSize 42 let payload41s = replicate 10 $ BL.replicate 10 41 - mapM_ (sendToMux w42 eq (MiniProtocolNum 42) ResponderDir) payload42s - mapM_ (sendToMux w41 eq (MiniProtocolNum 41) ResponderDir) payload41s + mapM_ (sendToMux w42 eq' (MiniProtocolNum 42) ResponderDir) payload42s + mapM_ (sendToMux w41 eq' (MiniProtocolNum 41) ResponderDir) payload41s when (runtSdus > 0) $ do let payload42s = replicate runtSdus $ BL.replicate sndSize 42 let payload41s = replicate runtSdus $ BL.replicate 10 41 - mapM_ (sendToMux w42 eq (MiniProtocolNum 42) ResponderDir) payload42s - mapM_ (sendToMux w41 eq (MiniProtocolNum 41) ResponderDir) payload41s + mapM_ (sendToMux w42 eq' (MiniProtocolNum 42) ResponderDir) payload42s + mapM_ (sendToMux w41 eq' (MiniProtocolNum 41) ResponderDir) payload41s -- Wait for the egress queue to empty atomically $ do @@ -253,7 +254,7 @@ startServerEgresss pollInterval sndSizeV ad = forever $ do let wasEmpty = BL.null buf writeTVar w (BL.append buf msg) when wasEmpty $ - writeTBQueue eq (TLSRDemand mc md $ Wanton w) + writeTBQueue eq (TLSRDemand mc md (Wanton w undefined undefined) undefined) else retry setupServer :: Socket -> IO Socket.SockAddr diff --git a/network-mux/changelog.d/20260330_133633_crocodile-dentist_mux_single_peer_performance.md b/network-mux/changelog.d/20260330_133633_crocodile-dentist_mux_single_peer_performance.md new file mode 100644 index 00000000000..02d71ee5551 --- /dev/null +++ b/network-mux/changelog.d/20260330_133633_crocodile-dentist_mux_single_peer_performance.md @@ -0,0 +1,32 @@ + + +### Breaking + +- Added support for burst and weighted fair queuing: + - Added `wLastSent` and `wBucket` to Wanton + - Introduced `ProtocolBurst` type, holding token bucket size and refill rate + - Added `ProtocolBurst` to `TLSRDemand` + - Added `burst` to `MiniProtocolLimits`. Value of Nothing denotes no burst + ability for the protocol, a Just holds a `ProtocolBurst` value. + - Added `miniProtocolWeight` to `MiniProtocolInfo`, denoting the queue weight + the protocol shares with other protocols of the same weight. + - Enhances 'starvation' test to properly handle weighted fair queueing + +--> + + diff --git a/network-mux/demo/mux-demo.hs b/network-mux/demo/mux-demo.hs index ab63ce3bf60..b4795c17d8d 100644 --- a/network-mux/demo/mux-demo.hs +++ b/network-mux/demo/mux-demo.hs @@ -81,7 +81,8 @@ debugTracer = show >$< Tracer putStrLn_ defaultProtocolLimits :: MiniProtocolLimits defaultProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = 64_000 + maximumIngressQueue = 64_000, + burst = Nothing } -- @@ -146,7 +147,8 @@ serverWorker bearer = do miniProtocolNum = MiniProtocolNum 2, miniProtocolDir = ResponderDirectionOnly, miniProtocolLimits = defaultProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] @@ -206,7 +208,8 @@ clientWorker bearer n msg = do miniProtocolNum = MiniProtocolNum 2, miniProtocolDir = InitiatorDirectionOnly, miniProtocolLimits = defaultProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] diff --git a/network-mux/demo/mux-leios-demo.hs b/network-mux/demo/mux-leios-demo.hs index 8dd7ca05677..86dd8e40f30 100644 --- a/network-mux/demo/mux-leios-demo.hs +++ b/network-mux/demo/mux-leios-demo.hs @@ -115,7 +115,8 @@ reqrespTracer tag = Tracer $ \case defaultProtocolLimits :: MiniProtocolLimits defaultProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = 10_000_000 + maximumIngressQueue = 10_000_000, + burst = Nothing } @@ -125,13 +126,15 @@ protocols miniProtocolDir = miniProtocolNum = MiniProtocolNum 2, miniProtocolDir, miniProtocolLimits = defaultProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } , MiniProtocolInfo { miniProtocolNum = MiniProtocolNum 3, miniProtocolDir, miniProtocolLimits = defaultProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] diff --git a/network-mux/src/Network/Mux.hs b/network-mux/src/Network/Mux.hs index 8df0b266f50..7e1e7790b49 100644 --- a/network-mux/src/Network/Mux.hs +++ b/network-mux/src/Network/Mux.hs @@ -1,5 +1,6 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE DataKinds #-} +{-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTSyntax #-} @@ -26,6 +27,7 @@ module Network.Mux , MiniProtocolNum (..) , MiniProtocolDirection (..) , MiniProtocolLimits (..) + , ProtocolBurst (..) -- * Running the Mux , run , stop @@ -66,9 +68,10 @@ import Data.ByteString.Lazy qualified as BL import Data.Int (Int64) import Data.Map (Map) import Data.Map.Strict qualified as Map -import Data.Maybe (isNothing) -import Data.Monoid.Synchronisation (FirstToFinish (..)) +import Data.Maybe (fromMaybe) +import Data.Monoid.Synchronisation (FirstToFinish (..), LastToFinish (..)) import Data.Strict.Tuple (pattern (:!:)) +import Data.Word (Word8) import Control.Applicative import Control.Concurrent.Class.MonadSTM.Strict @@ -78,6 +81,7 @@ import Control.Monad import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadThrow +import Control.Monad.Class.MonadTime.SI (Time (..)) import Control.Monad.Class.MonadTimer.SI hiding (timeout) import Control.Tracer @@ -226,10 +230,10 @@ run :: forall m (mode :: Mode). , MonadEvaluate m , MonadFork m , MonadLabelledSTM m - , Alternative (STM m) , MonadThrow (STM m) , MonadTimer m , MonadMask m + , MonadPlus (STM m) ) => Mux mode m -> Bearer m @@ -245,16 +249,27 @@ run Mux { muxMiniProtocols, bearer@Bearer{name} = do traceWith tracer_ TraceStarting - egressQueue <- atomically $ newTBQueue 100 + + let step mMap MiniProtocolState { miniProtocolInfo } = do + let weight = miniProtocolWeight miniProtocolInfo + Map.alterF (\case + Nothing -> do + q <- newTBQueue 100 + labelTBQueue q (name ++ "-mux-egress-" ++ show weight) + pure $ Just q + pass -> pure pass) + weight + =<< mMap + egressQueues <- atomically $ Map.foldl step (pure Map.empty) muxMiniProtocols -- label shared variables - labelTBQueueIO egressQueue (name ++ "-mux-egress") + labelTVarIO muxStatus (name ++ "-mux-status") labelTQueueIO muxControlCmdQueue (name ++ "-mux-ctrl") JobPool.withJobPool (\jobpool -> do - JobPool.forkJob jobpool (muxerJob egressQueue) + JobPool.forkJob jobpool (muxerJob (Map.assocs egressQueues)) JobPool.forkJob jobpool demuxerJob traceWith tracer_ (TraceState Mature) @@ -264,7 +279,7 @@ run Mux { muxMiniProtocols, monitor tracers timeout jobpool - egressQueue + egressQueues muxControlCmdQueue muxStatus ) @@ -277,8 +292,8 @@ run Mux { muxMiniProtocols, throwIO e where - muxerJob egressQueue = - JobPool.Job (muxer egressQueue bearerTracer_ bearer) + muxerJob egressQueues = + JobPool.Job (muxer egressQueues bearerTracer_ bearer) (return . MuxerException) MuxJob (name ++ "-muxer") @@ -312,7 +327,8 @@ miniProtocolJob TracersI { miniProtocolInfo = MiniProtocolInfo { miniProtocolNum, - miniProtocolDir + miniProtocolDir, + miniProtocolLimits }, miniProtocolIngressQueue, miniProtocolStatusVar @@ -328,9 +344,11 @@ miniProtocolJob TracersI { where jobAction = do w <- newTVarIO BL.empty - let chan = muxChannel channelTracer_ egressQueue (Wanton w) + lastSent <- newTVarIO (Time 0) + bucket <- newTVarIO 0 + let chan = muxChannel channelTracer_ egressQueue (Wanton w lastSent bucket) miniProtocolNum miniProtocolDirEnum - miniProtocolIngressQueue + miniProtocolIngressQueue (burst miniProtocolLimits) (result, remainder) <- miniProtocolAction chan traceWith tracer_ (TraceTerminating miniProtocolNum miniProtocolDirEnum) atomically $ do @@ -405,13 +423,13 @@ monitor :: forall mode m. ( MonadAsync m , MonadEvaluate m , MonadMask m - , Alternative (STM m) , MonadThrow (STM m) + , MonadPlus (STM m) ) => Tracers m -> TimeoutFn m -> JobPool.JobPool Group m JobResult - -> EgressQueue m + -> Map Word8 (EgressQueue m) -> StrictTQueue m (ControlCmd mode m) -> StrictTVar m Status -> m () @@ -419,7 +437,7 @@ monitor tracers@TracersI { tracer_ = tracer, bearerTracer_ = bearerTracer } - timeout jobpool egressQueue cmdQueue muxStatus = + timeout jobpool egressQueues cmdQueue muxStatus = go (MonitorCtx Map.empty Map.empty) where go :: MonitorCtx m mode -> m () @@ -492,7 +510,8 @@ monitor tracers@TracersI { miniProtocolInfo = MiniProtocolInfo { miniProtocolNum, miniProtocolDir, - miniProtocolCapability + miniProtocolCapability, + miniProtocolWeight } } ptclAction) -> do @@ -503,14 +522,14 @@ monitor tracers@TracersI { JobPool.forkJob jobpool $ miniProtocolJob tracers - egressQueue + (egressQueues Map.! miniProtocolWeight) ptclState ptclAction Just cap -> JobPool.forkJobOn cap jobpool $ miniProtocolJob tracers - egressQueue + (egressQueues Map.! miniProtocolWeight) ptclState ptclAction go monitorCtx @@ -556,10 +575,9 @@ monitor tracers@TracersI { atomically $ writeTVar muxStatus Stopping JobPool.cancelGroup jobpool MiniProtocolJob -- wait for 2 seconds before the egress queue is drained - _ <- timeout 2 $ - atomically $ - tryPeekTBQueue egressQueue - >>= check . isNothing + _ <- timeout 2 . atomically $ + let qs = map isEmptyTBQueue (Map.elems egressQueues) + in runLastToFinish $ foldl1 (<>) (LastToFinish <$> qs) atomically $ writeTVar muxStatus Stopped traceWith tracer TraceStopped -- by exiting the 'monitor' loop we let the job pool kill demuxer and @@ -597,7 +615,8 @@ monitor tracers@TracersI { miniProtocolInfo = MiniProtocolInfo { miniProtocolNum, miniProtocolDir, - miniProtocolCapability + miniProtocolCapability, + miniProtocolWeight }, miniProtocolStatusVar } @@ -610,14 +629,14 @@ monitor tracers@TracersI { JobPool.forkJob jobpool $ miniProtocolJob tracers - egressQueue + (egressQueues Map.! miniProtocolWeight) ptclState ptclAction Just cap -> JobPool.forkJobOn cap jobpool $ miniProtocolJob tracers - egressQueue + (egressQueues Map.! miniProtocolWeight) ptclState ptclAction @@ -676,8 +695,9 @@ muxChannel -> MiniProtocolNum -> MiniProtocolDir -> IngressQueue m + -> Maybe ProtocolBurst -> ByteChannel m -muxChannel tracer egressQueue want@(Wanton w) mc md q = +muxChannel tracer egressQueue want@(Wanton w _ _) mc md q mBurst = Channel { send, recv } where -- A soft limit on the egress buffer (Wanton) size. @@ -690,6 +710,8 @@ muxChannel tracer egressQueue want@(Wanton w) mc md q = egressSoftBufferLimit :: Int64 egressSoftBufferLimit = 0x3ffff + burst = fromMaybe (ProtocolBurst 0 0) mBurst + send :: BL.ByteString -> m () send encoding = do -- We send CBOR encoded messages by encoding them into by ByteString @@ -704,7 +726,7 @@ muxChannel tracer egressQueue want@(Wanton w) mc md q = let wasEmpty = BL.null buf writeTVar w (BL.append buf encoding) when wasEmpty $ - writeTBQueue egressQueue (TLSRDemand mc md want) + writeTBQueue egressQueue (TLSRDemand mc md want burst) else retry traceWith tracer $ TraceChannelSendEnd mc diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index ef8fcd5b769..d837d11f5bb 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -1,8 +1,14 @@ {-# LANGUAGE BangPatterns #-} +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-} module Network.Mux.Egress @@ -14,8 +20,13 @@ module Network.Mux.Egress , Wanton (..) ) where -import Control.Monad +import Control.Applicative +import Control.Exception import Data.ByteString.Lazy qualified as BL +import Data.List (tails) +import Data.Monoid (All (..), Ap (..)) +import Data.Monoid.Synchronisation +import Data.Word (Word32, Word8) import Control.Concurrent.Class.MonadSTM.Strict import Control.Monad.Class.MonadAsync @@ -120,13 +131,108 @@ type EgressQueue m = StrictTBQueue m (TranslocationServiceRequest m) -- arbitrary (yet bounded) size. This multiplexing layer is -- responsible for the segmentation of concrete representation into -- appropriate SDU's for onward transmission. -data TranslocationServiceRequest m = - TLSRDemand !MiniProtocolNum !MiniProtocolDir !(Wanton m) +data TranslocationServiceRequest m = TLSRDemand { + miniProtocolNum :: !MiniProtocolNum, + miniProtocolDir :: !MiniProtocolDir, + wanton :: !(Wanton m), + protocolBurst :: !ProtocolBurst + } -- | A Wanton represent the concrete data to be translocated, note that the -- TVar becoming empty indicates -- that the last fragment of the data has -- been enqueued on the -- underlying bearer. -newtype Wanton m = Wanton { want :: StrictTVar m BL.ByteString } +data Wanton m = Wanton { + wanton :: !(StrictTVar m BL.ByteString), + -- ^ data buffer + lastSent :: !(StrictTVar m Time), + -- ^ the last time the protocol has sent a message + burstBucket :: !(StrictTVar m TokenSize) + -- ^ the number of tokens available to burst + } + + +-- | A sequence of SDU's collected by the muxer to be sent to the kernel +-- in one syscall. +-- +data SDUBatch = SDUBatch { + getCount :: !Int, + -- ^ how many SDU's in the batch + getSdusLength :: !Int, + -- ^ The aggregate length of all sdu's (incl. header) + getSdus :: ![SDU] + -- ^ the payload itself + } + +instance Semigroup SDUBatch where + (SDUBatch a b c) <> (SDUBatch a' b' c') = SDUBatch (a + a') (b + b') (c <> c') + +mkSingletonBatch :: SDU -> SDUBatch +mkSingletonBatch sdu = SDUBatch 1 (sduLength sdu) [sdu] + + +sduLength :: SDU -> Int +sduLength sdu = fromIntegral msHeaderLength + fromIntegral (msLength sdu) + + +-- | By forming an `SDU` we also return state of the `Wanton`. +-- +data SDUWithWantonState = EmptyWanton SDU | NonEmptyWanton SDU + +-- | Next `SDUSize` when building a batch of `SDU`s +-- +data NextSDUSize + = BurstSize SDUSize + -- ^ Use ` `computeSDUSize` to compute the allowed `SDUSize` + | BearerSize + -- ^ Use `Bearer`'s `sduSize` + +computeSDUSize + :: SDUSize -- ^ Bearer SDUSize + -> TokenSize -- ^ token utilised when building a batch of `SDU`s. + -> SDUSize -- ^ the effective `SDUSize` +computeSDUSize sduSize = + min sduSize + . fromIntegral @TokenSize @SDUSize -- Word32 -> Word16 + . min (fromIntegral @SDUSize @TokenSize maxBound) -- Word16 -> Word32 (but at most Word16) + +nextSDUSizeToSDUSize :: SDUSize -> NextSDUSize -> SDUSize +nextSDUSizeToSDUSize _sduSize (BurstSize sduSize) = sduSize +nextSDUSizeToSDUSize sduSize BearerSize = sduSize +{-# INLINE nextSDUSizeToSDUSize #-} + +type TokenSize = Word32 + +-- | Tokens consumed by an SDU. +-- +consumedTokens :: NextSDUSize -> SDU -> TokenSize +-- in burst mode, we charge tokens based on the SDU payload length +consumedTokens BurstSize{} sdu = fromIntegral (msLength sdu) +-- in non-burst mode, SDU token size is 0 +consumedTokens BearerSize _sdu = 0 +{-# INLINE consumedTokens #-} + + +-- | Can we burst a single mini-protocol. +-- +data CanBurst = BurstAllowed + | BurstNotAllowed + + +-- | Can we batch more SDUs from different mini-protocols. +-- +data CanBatch = BatchAllowed + | BatchNotAllowed + + +-- | Maximal number of `SDU`s in a `SDUBatch`. +-- +maxSDUsPerBatch :: Int +maxSDUsPerBatch = 100 + +-- | Minimal SDUSize for an SDU to be burst. +-- +burstMinSdu :: SDUSize +burstMinSdu = truncate @Double $ fromIntegral msHeaderLength / 0.02 -- | Process the messages from the mini protocols - there is a single @@ -141,78 +247,230 @@ muxer , MonadMask m , MonadThrow (STM m) , MonadTimer m + , Alternative (STM m) ) - => EgressQueue m + => [(Word8, EgressQueue m)] -> Tracer m BearerTrace -> Bearer m -> m void -muxer egressQueue tracer Bearer { writeMany, sduSize, batchSize, egressInterval } = - withTimeoutSerial $ \timeout -> - forever $ do +muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterval } = + withTimeoutSerial $ \timeout -> muxerLoop timeout (cycle egressQueues0) + where + numQueues :: Int + numQueues = length egressQueues0 + toDouble :: DiffTime -> Double + toDouble = realToFrac + + -- main muxer loop + muxerLoop :: (forall a. DiffTime -> m a -> m (Maybe a)) + -> [(Word8, EgressQueue m)] + -- ^ a cycle of egress queues + -> m void + muxerLoop timeout egressQueues = do start <- getMonotonicTime - TLSRDemand mpc md d <- atomically $ readTBQueue egressQueue - sdu <- processSingleWanton egressQueue sduSize mpc md d - sdus <- buildBatch [sdu] (sduLength sdu) - void $ writeMany tracer timeout sdus - end <- getMonotonicTime - empty <- atomically $ isEmptyTBQueue egressQueue - when empty $ do - let delta = diffTime end start - threadDelay (egressInterval - delta) + (sdu, egressQueues', canBatch) <- atomically do + let -- All distinct `EgressQueue`s and their tail (so we keep reading + -- them in a round robin way). + available :: [(EgressQueue m, [(Word8, EgressQueue m)])] + available = take numQueues (snd <$> egressQueues) + `zip` + tails egressQueues - where - maxSDUsPerBatch :: Int - maxSDUsPerBatch = 100 + -- read first available `EgressQueue` and return its tail + job <- runFirstToFinish + . foldMap + ( FirstToFinish + . \(egressQueue, egressQueues') -> + (,egressQueues') <$> readTBQueue egressQueue + ) + $ available + case job of + (demand@(TLSRDemand mpc md d ProtocolBurst{maxBytes}) + , egressQueues' + ) -> do + let ((weight, egressQueue), rest) = assert (weight > 0) + case egressQueues' of + [] -> error "impossible" + x:xs -> (x, xs) + egressQueues'' | weight > 1 = (pred weight, egressQueue) + : rest + | otherwise = rest + eSdu <- processSingleWanton mpc md d sduSize + case eSdu of + NonEmptyWanton sdu + | maxBytes > 0 + -> -- we do not check if the protocol has any tokens to + -- burst, that is deferred to buildBatch below. + (sdu, egressQueues', BatchAllowed) + <$ unGetTBQueue egressQueue demand + + | otherwise + -> (sdu, egressQueues'', BatchNotAllowed) + <$ writeTBQueue egressQueue demand - sduLength :: SDU -> Int - sduLength sdu = fromIntegral msHeaderLength + fromIntegral (msLength sdu) + EmptyWanton sdu -> + pure (sdu, egressQueues'', BatchNotAllowed) + + (egressQueues'', SDUBatch { getSdus = sdus }) <- + buildBatch (mkSingletonBatch sdu) egressQueues' canBatch start + _ <- writeMany tracer timeout sdus + end <- getMonotonicTime + threadDelay $ egressInterval - end `diffTime` start + muxerLoop timeout egressQueues'' -- Build a batch of SDUs to submit in one go to the bearer. - -- The egress queue is still processed one SDU at the time - -- to ensure that we don't cause starvation. + -- Streams which are permitted to burst will have that many + -- sdu's serviced back-to-back before the scheduler moves to process the + -- next request on the queue. Any remaining sdu's which did not + -- fit in the burst allowance are placed on the back of the queue + -- to ensure that we don't cause starvation. In particular, a burst + -- of 1 will have the muxer process one sdu at a time from the queue, + -- and any remaining work is put on the back of the queue. -- The batch size is either limited by the bearer -- (e.g the SO_SNDBUF for Socket) or number of SDUs. -- - buildBatch s sl = reverse <$> go s sl + buildBatch + :: SDUBatch + -> [(Word8, EgressQueue m)] + -> CanBatch -- ^ can we batch more SDUs + -> Time + -> m ([(Word8, EgressQueue m)], SDUBatch) + buildBatch batch0 egressQueues1 canBatch0 start = do + (qs, batch) <- go batch0 egressQueues1 canBatch0 + pure (qs, batch { getSdus = reverse (getSdus batch) }) where - go sdus _ | length sdus >= maxSDUsPerBatch = return sdus - go sdus sdusLength | sdusLength >= batchSize = return sdus - go sdus !sdusLength = do - demand_m <- atomically $ tryReadTBQueue egressQueue - case demand_m of - Just (TLSRDemand mpc md d) -> do - sdu <- processSingleWanton egressQueue sduSize mpc md d - go (sdu:sdus) (sdusLength + sduLength sdu) - Nothing -> return sdus + go :: SDUBatch + -> [(Word8, EgressQueue m)] + -> CanBatch + -> m ([(Word8, EgressQueue m)], SDUBatch) + go !_batch [] !_canBatch = error "impossible" + go !batch egressQueues !_canBatch + | getCount batch >= maxSDUsPerBatch || getSdusLength batch >= batchSize + = return (egressQueues, batch) + go !batch egressQueues@((weight, queue):rest) !canBatch = do + -- since the list of queues cycles, we only need to check the prefix + -- to see if there is any more work to do. + All allEmpty0 <- + atomically $ getAp $ foldMap (Ap . fmap All . isEmptyTBQueue . snd) + (take numQueues egressQueues) + if allEmpty0 + then return (egressQueues, batch) + else + atomically (tryReadTBQueue queue) >>= \case + Nothing -> go batch rest BatchNotAllowed + Just demand@TLSRDemand { + wanton = Wanton { lastSent, burstBucket }, + protocolBurst = ProtocolBurst { maxBytes, refillRate } + } -> do + (batch', canBurst) <- atomically do + delta <- (start `diffTime`) <$> stateTVar lastSent (, start) + + nextSduSize <- stateTVar burstBucket \tokens -> + let tokens' :: TokenSize + tokens' = truncate $ + min (fromIntegral maxBytes) + (fromIntegral tokens + fromIntegral refillRate * toDouble delta) + -- we leverage burst and deduct credits only where there is contention + -- between protocols + nextSduSize :: NextSDUSize + nextSduSize = + case canBatch of + BatchAllowed -> BurstSize $ computeSDUSize sduSize tokens' + BatchNotAllowed -> BearerSize + in (nextSduSize, tokens') + if nextSDUSizeToSDUSize maxBound nextSduSize <= burstMinSdu + then do + -- edge case where the protocol is bursty, but there aren't enough tokens + -- available. The muxer forever loop does not check this + -- when it calls to build a batch, so we handle it here. + writeTBQueue queue demand + return (batch, BurstAllowed) + else + burstLoop demand batch nextSduSize + + case canBurst of + BurstAllowed + | weight > 1 -> + let egressQueues' = (pred weight, queue) : rest in + go batch' egressQueues' BatchNotAllowed + | otherwise -> + go batch' rest BatchNotAllowed + BurstNotAllowed -> + go batch' rest BatchNotAllowed + where + -- burst SDUs from a single mini-protocol until we consume all tokens + -- (`TokenSize`). + burstLoop :: TranslocationServiceRequest m + -> SDUBatch + -> NextSDUSize + -> STM m (SDUBatch, CanBurst) + burstLoop demand@TLSRDemand { miniProtocolNum, + miniProtocolDir, + wanton = want@Wanton {burstBucket} + } + !batch' !nextSDUSize = do + -- The first SDU is always free. For `BearerSize` (no bursting), + -- we don't count the wanton bytes against the burst allowance + -- to permit a full sdu in the first iteration + x <- processSingleWanton miniProtocolNum miniProtocolDir + want + (nextSDUSizeToSDUSize sduSize nextSDUSize) + case x of + EmptyWanton sdu -> do + -- the `Wanton` is empty + modifyTVar burstBucket \tokens -> + let consumed, tokens' :: TokenSize + consumed = consumedTokens nextSDUSize sdu + tokens' = tokens - consumed + in assert (tokens >= consumed) + tokens' + continue <- (\case + True -> BurstNotAllowed + False -> BurstAllowed) + <$> isEmptyTBQueue queue + return (mkSingletonBatch sdu <> batch', continue) + + NonEmptyWanton sdu -> do + -- the `Wanton` is non-empty + nextSdu <- stateTVar burstBucket \tokens -> + let consumed, tokens' :: TokenSize + consumed = consumedTokens nextSDUSize sdu + tokens' = tokens - consumed + in assert (tokens >= consumed) + (computeSDUSize sduSize tokens', tokens') + let batch'' = mkSingletonBatch sdu <> batch' + if nextSdu <= burstMinSdu -- 8 bytes header / 2% burst efficiency + then do + -- burst allowance has been exhausted, next + -- SDU would be too small + writeTBQueue queue demand + return (batch'', BurstAllowed) + else burstLoop demand batch'' (BurstSize nextSdu) + -- | Pull a `maxSDU`s worth of data out out the `Wanton` - if there is -- data remaining requeue the `TranslocationServiceRequest` (this -- ensures that any other items on the queue will get some service -- first. processSingleWanton :: MonadSTM m - => EgressQueue m - -> SDUSize - -> MiniProtocolNum + => MiniProtocolNum -> MiniProtocolDir -> Wanton m - -> m SDU -processSingleWanton egressQueue (SDUSize sduSize) - mpc md wanton = do - blob <- atomically $ do + -> SDUSize + -> STM m SDUWithWantonState +processSingleWanton mpc md Wanton{wanton} sduSize = do + (blob, wrap) <- do -- extract next SDU - d <- readTVar (want wanton) + d <- readTVar wanton let (frag, rest) = BL.splitAt (fromIntegral sduSize) d -- if more to process then enqueue remaining work if BL.null rest - then writeTVar (want wanton) BL.empty + then (frag, EmptyWanton) <$ writeTVar wanton BL.empty else do -- Note that to preserve bytestream ordering within a given -- miniprotocol the readTVar and writeTVar operations -- must be inside the same STM transaction. - writeTVar (want wanton) rest - writeTBQueue egressQueue (TLSRDemand mpc md wanton) - -- return data to send - pure frag + (frag, NonEmptyWanton) <$ writeTVar wanton rest let sdu = SDU { msHeader = SDUHeader { mhTimestamp = RemoteClockModel 0, @@ -222,5 +480,5 @@ processSingleWanton egressQueue (SDUSize sduSize) }, msBlob = blob } - return sdu + pure $ wrap sdu --paceTransmission tNow diff --git a/network-mux/src/Network/Mux/Types.hs b/network-mux/src/Network/Mux/Types.hs index 00fd1e47dcc..4af8c9cfd94 100644 --- a/network-mux/src/Network/Mux/Types.hs +++ b/network-mux/src/Network/Mux/Types.hs @@ -25,6 +25,7 @@ module Network.Mux.Types , IngressQueue , MiniProtocolIx , MiniProtocolDir (..) + , ProtocolBurst (..) , protocolDirEnum , MiniProtocolState (..) , MiniProtocolStatus (..) @@ -93,15 +94,25 @@ newtype MiniProtocolNum = MiniProtocolNum Word16 deriving (Eq, Ord, Enum, Ix, Show) -- | Per Miniprotocol limits -newtype MiniProtocolLimits = +data MiniProtocolLimits = MiniProtocolLimits { -- | Limit on the maximum number of bytes that can be queued in the -- miniprotocol's ingress queue. -- - maximumIngressQueue :: Int + maximumIngressQueue :: !Int, + burst :: !(Maybe ProtocolBurst) } deriving Show + +data ProtocolBurst = ProtocolBurst { + maxBytes :: !Word32, + -- ^ token bucket max size + refillRate :: !Word32 + -- ^ token bucket refill rate, [1/s] + } + deriving (Eq, Show) + -- $interface -- -- To run a node you will also need a bearer and a way to run a server, see @@ -146,8 +157,12 @@ data MiniProtocolInfo (mode :: Mode) = -- ^ Mini-protocol direction. miniProtocolLimits :: !MiniProtocolLimits, -- ^ ingress queue limits for the protocol - miniProtocolCapability :: !(Maybe Int) + miniProtocolCapability :: !(Maybe Int), -- ^ capability on which the mini-protocol should run + miniProtocolWeight :: !Word8 + -- ^ Protocols with the same weight will share + -- an egress queue of that value which biases + -- the muxer relative to other protocols } deriving Show @@ -287,7 +302,7 @@ newtype SDUSize = SDUSize { getSDUSize :: Word16 } deriving Generic deriving Show via Quiet SDUSize deriving (Eq, Ord, Enum) - deriving (Num, Real, Integral) + deriving (Bounded, Num, Real, Integral) -- | A channel which wraps each message as an 'SDU' using giving -- 'MiniProtocolNum' and 'MiniProtocolDir'. diff --git a/network-mux/test/Test/Mux.hs b/network-mux/test/Test/Mux.hs index 8b7627b4112..118871fa413 100644 --- a/network-mux/test/Test/Mux.hs +++ b/network-mux/test/Test/Mux.hs @@ -3,6 +3,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PackageImports #-} {-# LANGUAGE RankNTypes #-} @@ -29,7 +30,7 @@ import Data.Bits import Data.ByteString.Lazy qualified as BL import Data.ByteString.Lazy.Char8 qualified as BL8 (pack) import Data.Functor.Contravariant ((>$<)) -import Data.List (dropWhileEnd, nub) +import Data.List (dropWhileEnd, group, nub) import Data.List qualified as List import Data.Map qualified as M import Data.Maybe (isNothing) @@ -116,7 +117,8 @@ tests = defaultMiniProtocolLimits :: MiniProtocolLimits defaultMiniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = defaultMiniProtocolLimit + maximumIngressQueue = defaultMiniProtocolLimit, + burst = Nothing } defaultMiniProtocolLimit :: Int @@ -125,7 +127,8 @@ defaultMiniProtocolLimit = 3000000 smallMiniProtocolLimits :: MiniProtocolLimits smallMiniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = smallMiniProtocolLimit + maximumIngressQueue = smallMiniProtocolLimit, + burst = Nothing } smallMiniProtocolLimit :: Int @@ -312,6 +315,35 @@ instance Arbitrary DummyCapability where ] +newtype MiniProtocolWeights = MiniProtocolWeights (MiniProtocolWeight, MiniProtocolWeight) + deriving (Eq, Show) + +instance Arbitrary MiniProtocolWeights where + arbitrary = do + wt1 <- arbitrary + wt2 <- arbitrary + let mkWt = MiniProtocolWeight + if wt1 == wt2 + then pure $ MiniProtocolWeights (mkWt 1, mkWt 1) + else pure $ MiniProtocolWeights (wt1, wt2) + + shrink (MiniProtocolWeights (wt1, wt2)) = + let mkWt = MiniProtocolWeight + in MiniProtocolWeights (mkWt 1, mkWt 1) + : [ MiniProtocolWeights (wt1', wt2') + | wt1' <- shrink wt1 + , wt2' <- shrink wt2 + ] + + +newtype MiniProtocolWeight = MiniProtocolWeight Word8 + deriving (Eq, Show) + +instance Arbitrary MiniProtocolWeight where + arbitrary = MiniProtocolWeight <$> choose (1, 4) + shrink (MiniProtocolWeight wt) = MiniProtocolWeight <$> filter (> 0) (shrink wt) + + -- | A pair of two bytestrings which lengths are unevenly distributed -- data Uneven = Uneven DummyPayload DummyPayload @@ -372,14 +404,16 @@ prop_mux_snd_recv (DummyRun messages) = ioProperty $ do miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.InitiatorDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } serverApp = MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } clientMux <- Mx.new clientTracer [clientApp] @@ -445,13 +479,15 @@ prop_mux_snd_recv_bi (DummyRun messages) (DummyCapability clientCap) (DummyCapab miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.InitiatorDirection, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } , MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirection, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = clientCap + miniProtocolCapability = clientCap, + miniProtocolWeight = 1 } ] @@ -460,13 +496,15 @@ prop_mux_snd_recv_bi (DummyRun messages) (DummyCapability clientCap) (DummyCapab miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirection, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = serverCap + miniProtocolCapability = serverCap, + miniProtocolWeight = 1 } , MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.InitiatorDirection, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] @@ -560,14 +598,16 @@ prop_mux_snd_recv_compat messages = ioProperty $ do miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolLimits = defaultMiniProtocolLimits, miniProtocolDir = Mx.InitiatorDirectionOnly, - miniProtocolCapability = Nothing } + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] serverBundle = [ MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolLimits = defaultMiniProtocolLimits, miniProtocolDir = Mx.ResponderDirectionOnly, - miniProtocolCapability = Nothing } + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] clientAsync <- async $ do @@ -766,7 +806,8 @@ runMuxApplication (DummyCapability rspCap) initApps initBearer respApps respBear miniProtocolNum = Mx.MiniProtocolNum pn, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = rspCap + miniProtocolCapability = rspCap, + miniProtocolWeight = 1 } ) respApps' @@ -785,7 +826,8 @@ runMuxApplication (DummyCapability rspCap) initApps initBearer respApps respBear miniProtocolNum = Mx.MiniProtocolNum pn, miniProtocolDir = Mx.InitiatorDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ) initApps' @@ -996,8 +1038,10 @@ prop_mux_2_minis_Socket_buf cap a b = ioProperty $ -- The Mux bearer should alternate between sending data for the two responders. -- prop_mux_starvation :: Uneven + -> MiniProtocolWeights -> Property -prop_mux_starvation (Uneven response0 response1) = +prop_mux_starvation (Uneven response0 response1) + (MiniProtocolWeights (MiniProtocolWeight wt1, MiniProtocolWeight wt2)) = let sduLen = Mx.SDUSize 1280 in (BL.length (unDummyPayload response0) > 2 * fromIntegral (Mx.getSDUSize sduLen)) && (BL.length (unDummyPayload response1) > 2 * fromIntegral (Mx.getSDUSize sduLen)) ==> @@ -1043,13 +1087,15 @@ prop_mux_starvation (Uneven response0 response1) = miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.InitiatorDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } clientApp3 = MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 3, miniProtocolDir = Mx.InitiatorDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } serverApp2, serverApp3 :: MiniProtocolInfo Mx.ResponderMode @@ -1057,13 +1103,15 @@ prop_mux_starvation (Uneven response0 response1) = miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = wt1 } serverApp3 = MiniProtocolInfo { miniProtocolNum = Mx.MiniProtocolNum 3, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = wt2 } serverMux <- Mx.new serverTracer [serverApp2, serverApp3] @@ -1103,28 +1151,30 @@ prop_mux_starvation (Uneven response0 response1) = -- Then look at the message trace to check for starvation. trace <- atomically $ readTVar traceHeaderVar - let es = map Mx.mhNum (take 100 (reverse trace)) - ls = dropWhile (\e -> e == head es) es - fair = verifyStarvation ls + let es = map Mx.mhNum (take 100 (reverse trace)) + -- We can't make 100% sure that both servers start responding at the same + -- time but once they are both up and running messages should alternate + -- between ReqResp2 and ReqResp3, so we drop the prefix of the protocol + -- which goes first, and trim the suffix when the first protocol finishes + ls = dropWhile (\e -> e == head es) es + ls' = dropWhileEnd (\e -> e == last ls) ls + fair = counterexample "muxer didn't interleave" (not . null $ ls') + .&&. label ("shrinkage " ++ labelPr_ ((length ls' * 100) `div` length es) ++ "%") + (verifyStarvation ls' (\case (Mx.MiniProtocolNum 2) -> wt1; _otherwise -> wt2)) return $ res_short .&&. res_long .&&. fair where - -- We can't make 100% sure that both servers start responding at the same - -- time but once they are both up and running messages should alternate - -- between ReqResp2 and ReqResp3 - verifyStarvation :: Eq a => [a] -> Property - verifyStarvation [] = property True - verifyStarvation ms = - let ms' = dropWhileEnd (\e -> e == last ms) - (head ms : dropWhile (\e -> e == head ms) ms) - ++ [last ms] - in - label ("length " ++ labelPr_ ((length ms' * 100) `div` length ms) ++ "%") - $ label ("length " ++ label_ (length ms')) $ alternates ms' + verifyStarvation :: Eq a => [a] -> (a -> Word8) -> Property + verifyStarvation [] _atowt = property True + verifyStarvation ms atowt = label ("length " ++ label_ (length ms)) . + label ("groups " ++ label_ (length (group ms))) . + alternates $ group ms where alternates [] = True alternates (_:[]) = True - alternates (a : b : as) = a /= b && alternates (b : as) + alternates (a : b : []) = length a <= fromIntegral (atowt (head a)) + && length b <= fromIntegral (atowt (head b)) + alternates (a : b : as) = (length a == fromIntegral (atowt (head a))) && alternates (b : as) label_ :: Int -> String label_ n = mconcat @@ -1151,8 +1201,7 @@ encodeInvalidMuxSDU sdu = -- | Verify ingress processing of valid and invalid SDUs. -- prop_demux_sdu :: forall m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -1161,6 +1210,7 @@ prop_demux_sdu :: forall m. , MonadSay m , MonadThrow (STM m) , MonadTimer m + , MonadPlus (STM m) ) => ArbitrarySDU -> m Property @@ -1179,7 +1229,8 @@ prop_demux_sdu a = do miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = smallMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } (client_w, said, waitServerRes, mux) <- plainServer server_mps (serverRsp stopVar) @@ -1208,7 +1259,8 @@ prop_demux_sdu a = do miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } (client_w, said, waitServerRes, mux) <- plainServer server_mps (serverRsp stopVar) @@ -1236,7 +1288,8 @@ prop_demux_sdu a = do miniProtocolNum = Mx.MiniProtocolNum 2, miniProtocolDir = Mx.ResponderDirectionOnly, miniProtocolLimits = defaultMiniProtocolLimits, - miniProtocolCapability = Nothing + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } (client_w, said, waitServerRes, mux) <- plainServer server_mps (serverRsp stopVar) @@ -1472,7 +1525,7 @@ dummyRestartingAppToChannel (app, r) = \_ -> do appToInfo :: Mx.MiniProtocolDirection mode -> DummyApp -> MiniProtocolInfo mode -appToInfo d da = MiniProtocolInfo (daNum da) d defaultMiniProtocolLimits Nothing +appToInfo d da = MiniProtocolInfo (daNum da) d defaultMiniProtocolLimits Nothing 1 triggerApp :: forall m. ( MonadAsync m @@ -1491,8 +1544,7 @@ triggerApp bearer app = do return () prop_mux_start_mX :: forall m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -1501,6 +1553,7 @@ prop_mux_start_mX :: forall m. , MonadSay m , MonadThrow (STM m) , MonadTimer m + , MonadPlus (STM m) ) => DummyApps -> DiffTime @@ -1555,8 +1608,7 @@ prop_mux_start_mX apps runTime = do Right _ -> return (counterexample "not-failed" False, r) prop_mux_restart_m :: forall m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -1565,6 +1617,7 @@ prop_mux_restart_m :: forall m. , MonadSay m , MonadThrow (STM m) , MonadTimer m + , MonadPlus (STM m) ) => DummyRestartingApps -> m Property @@ -1726,8 +1779,7 @@ prop_mux_restart_m (DummyRestartingInitiatorResponderApps rapps) = do -- | Verifying starting and stopping of miniprotocols. Both normal exits and by exception. prop_mux_start_m :: forall m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -1736,6 +1788,7 @@ prop_mux_start_m :: forall m. , MonadSay m , MonadThrow (STM m) , MonadTimer m + , MonadPlus (STM m) ) => Mx.Bearer m -- ^ Mux bearer @@ -1973,8 +2026,7 @@ withNetworkCtx NetworkCtx { ncSocket, ncClose, ncMuxBearer } k = close_experiment :: forall sock acc req resp m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -1983,6 +2035,7 @@ close_experiment , MonadTimer m , MonadThrow (STM m) , MonadST m + , MonadPlus (STM m) , Serialise req , Serialise resp , Eq resp @@ -2016,8 +2069,9 @@ close_experiment [ MiniProtocolInfo { miniProtocolNum, miniProtocolDir = Mx.InitiatorDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ]) Mx.stop $ \mux -> @@ -2036,8 +2090,9 @@ close_experiment [ MiniProtocolInfo { miniProtocolNum, miniProtocolDir = Mx.ResponderDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ]) Mx.stop $ \mux -> @@ -2371,13 +2426,13 @@ instance Arbitrary NonEmptyByteString where ] prop_mux_trailing_bytes - :: ( Alternative (STM m) - , MonadAsync m + :: ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , MonadTimer m , MonadThrow (STM m) ) @@ -2395,8 +2450,9 @@ prop_mux_trailing_bytes reminder (NonEmptyByteString received) = do [ MiniProtocolInfo { miniProtocolNum, miniProtocolDir = Mx.ResponderDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] withAsync (Mx.run mux bearer) $ \_ -> do @@ -2471,13 +2527,13 @@ prop_mux_trailing_bytes_io reminder received = prop_mux_pure_exception - :: ( Alternative (STM m) - , MonadAsync m + :: ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , MonadTimer m , MonadThrow (STM m) ) @@ -2493,8 +2549,9 @@ prop_mux_pure_exception = do [ MiniProtocolInfo { miniProtocolNum, miniProtocolDir = Mx.ResponderDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] withAsync (Mx.run mux bearer) $ \_ -> do diff --git a/ouroboros-network/changelog.d/20260330_134651_crocodile-dentist_mux_single_peer_performance.md b/ouroboros-network/changelog.d/20260330_134651_crocodile-dentist_mux_single_peer_performance.md new file mode 100644 index 00000000000..8dc740b9aa4 --- /dev/null +++ b/ouroboros-network/changelog.d/20260330_134651_crocodile-dentist_mux_single_peer_performance.md @@ -0,0 +1,23 @@ + + + +### Non-Breaking + +- Integrate weighted fair queue + burst mux + + diff --git a/ouroboros-network/demo/connection-manager.hs b/ouroboros-network/demo/connection-manager.hs index 1ad711fb72c..aed0600e092 100644 --- a/ouroboros-network/demo/connection-manager.hs +++ b/ouroboros-network/demo/connection-manager.hs @@ -25,6 +25,7 @@ module Main (main) where import Control.Concurrent.Class.MonadSTM qualified as LazySTM import Control.Concurrent.Class.MonadSTM.Strict import Control.Exception (IOException) +import Control.Monad import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -190,6 +191,7 @@ withBidirectionalConnectionManager , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m + , MonadPlus (STM m) ) => Snocket m socket peerAddr -> Mux.MakeBearer m socket @@ -321,11 +323,12 @@ withBidirectionalConnectionManager snocket makeBearer socket in MiniProtocol { miniProtocolNum, miniProtocolStart = StartOnDemand, - miniProtocolLimits = Mux.MiniProtocolLimits maxBound, + miniProtocolLimits = Mux.MiniProtocolLimits maxBound Nothing, miniProtocolRun = reqRespInitiatorAndResponder miniProtocolNum - hotRequestsVar + hotRequestsVar, + miniProtocolWeight = 1 } ], withWarm = WithWarm @@ -333,11 +336,12 @@ withBidirectionalConnectionManager snocket makeBearer socket in MiniProtocol { miniProtocolNum, miniProtocolStart = StartOnDemand, - miniProtocolLimits = Mux.MiniProtocolLimits maxBound, + miniProtocolLimits = Mux.MiniProtocolLimits maxBound Nothing, miniProtocolRun = reqRespInitiatorAndResponder miniProtocolNum - warmRequestsVar + warmRequestsVar, + miniProtocolWeight = 1 } ], withEstablished = WithEstablished @@ -345,11 +349,12 @@ withBidirectionalConnectionManager snocket makeBearer socket in MiniProtocol { miniProtocolNum, miniProtocolStart = StartOnDemandAny, - miniProtocolLimits = Mux.MiniProtocolLimits maxBound, + miniProtocolLimits = Mux.MiniProtocolLimits maxBound Nothing, miniProtocolRun = reqRespInitiatorAndResponder (Mux.MiniProtocolNum 3) - establishedRequestsVar + establishedRequestsVar, + miniProtocolWeight = 1 } ] } diff --git a/ouroboros-network/demo/ping-pong.hs b/ouroboros-network/demo/ping-pong.hs index a3626b26485..0a1ff3e508b 100644 --- a/ouroboros-network/demo/ping-pong.hs +++ b/ouroboros-network/demo/ping-pong.hs @@ -90,7 +90,8 @@ rmIfExists path = do maximumMiniProtocolLimits :: MiniProtocolLimits maximumMiniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = maxBound + maximumIngressQueue = maxBound, + burst = Nothing } tracer :: Show a => Tracer IO a @@ -112,7 +113,8 @@ demoProtocol0 pingPong = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = pingPong + miniProtocolRun = pingPong, + miniProtocolWeight = 1 } ] @@ -204,13 +206,15 @@ demoProtocol1 pingPong pingPong' = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = pingPong + miniProtocolRun = pingPong, + miniProtocolWeight = 1 } , MiniProtocol { miniProtocolNum = MiniProtocolNum 3, miniProtocolStart = StartOnDemandAny, miniProtocolLimits = maximumMiniProtocolLimits, - miniProtocolRun = pingPong' + miniProtocolRun = pingPong', + miniProtocolWeight = 1 } ] @@ -300,5 +304,3 @@ serverPingPong2 = , codecPingPong , pingPongServerPeer pingPongServerStandard ) - - diff --git a/ouroboros-network/framework/io-tests/Test/Ouroboros/Network/Socket.hs b/ouroboros-network/framework/io-tests/Test/Ouroboros/Network/Socket.hs index 0e7037b4dcc..99be869a07f 100644 --- a/ouroboros-network/framework/io-tests/Test/Ouroboros/Network/Socket.hs +++ b/ouroboros-network/framework/io-tests/Test/Ouroboros/Network/Socket.hs @@ -118,9 +118,11 @@ testProtocols2 reqResp = miniProtocolNum = MiniProtocolNum 4, miniProtocolStart = StartOnDemand, miniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = defaultMiniProtocolLimit + maximumIngressQueue = defaultMiniProtocolLimit, + burst = Nothing }, - miniProtocolRun = reqResp + miniProtocolRun = reqResp, + miniProtocolWeight = 1 } ] diff --git a/ouroboros-network/framework/lib/Ouroboros/Network/ConnectionHandler.hs b/ouroboros-network/framework/lib/Ouroboros/Network/ConnectionHandler.hs index b4b93544114..cafbf4b5eb6 100644 --- a/ouroboros-network/framework/lib/Ouroboros/Network/ConnectionHandler.hs +++ b/ouroboros-network/framework/lib/Ouroboros/Network/ConnectionHandler.hs @@ -42,10 +42,10 @@ module Ouroboros.Network.ConnectionHandler , ConnectionHandlerTrace (..) ) where -import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadSTM.Strict import Control.DeepSeq (NFData) import Control.Exception (SomeAsyncException) +import Control.Monad (MonadPlus) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadThrow hiding (handle) @@ -242,8 +242,7 @@ type ConnectionManagerWithExpandedCtx muxMode socket peerAddr extraFlags version -- makeConnectionHandler :: forall initiatorCtx responderCtx peerAddr muxMode socket versionNumber versionData m a b. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -251,6 +250,7 @@ makeConnectionHandler , MonadThrow (STM m) , MonadTimer m , MonadMask m + , MonadPlus (STM m) , NFData versionData , NFData versionNumber , Ord versionNumber diff --git a/ouroboros-network/framework/lib/Ouroboros/Network/Mux.hs b/ouroboros-network/framework/lib/Ouroboros/Network/Mux.hs index 0904451bc23..420203c6fb8 100644 --- a/ouroboros-network/framework/lib/Ouroboros/Network/Mux.hs +++ b/ouroboros-network/framework/lib/Ouroboros/Network/Mux.hs @@ -67,6 +67,7 @@ import Data.Foldable (fold) import Data.Hashable import Data.Kind (Type) import Data.Void (Void) +import Data.Word (Word8) import Network.TypedProtocol.Codec import Network.TypedProtocol.Core @@ -247,8 +248,9 @@ data MiniProtocol (mode :: Mux.Mode) initiatorCtx responderCtx bytes m a b = -- started using `StartEagerly`. miniProtocolLimits :: !MiniProtocolLimits, -- ^ mini-protocol limits - miniProtocolRun :: !(RunMiniProtocol mode initiatorCtx responderCtx bytes m a b) + miniProtocolRun :: !(RunMiniProtocol mode initiatorCtx responderCtx bytes m a b), -- ^ mini-protocol callback(s) + miniProtocolWeight :: !Word8 } mkMiniProtocolInfo :: ForkPolicyCb @@ -257,13 +259,15 @@ mkMiniProtocolInfo :: ForkPolicyCb mkMiniProtocolInfo forkPolicy MiniProtocol { miniProtocolNum, miniProtocolLimits, - miniProtocolRun + miniProtocolRun, + miniProtocolWeight } = [ Mux.MiniProtocolInfo { Mux.miniProtocolNum, Mux.miniProtocolDir = dir, Mux.miniProtocolLimits, + Mux.miniProtocolWeight, Mux.miniProtocolCapability = forkPolicy miniProtocolNum (Mux.protocolDirEnum dir) @@ -528,4 +532,3 @@ mkMiniProtocolInfos :: ForkPolicyCb -> OuroborosBundle mode initiatorCtx responderCtx bytes m a b -> [MiniProtocolInfo mode] mkMiniProtocolInfos forkPolicy = foldMap (foldMap (mkMiniProtocolInfo forkPolicy)) - diff --git a/ouroboros-network/framework/lib/Ouroboros/Network/Server/Simple.hs b/ouroboros-network/framework/lib/Ouroboros/Network/Server/Simple.hs index 44e426e4233..befbadcebcf 100644 --- a/ouroboros-network/framework/lib/Ouroboros/Network/Server/Simple.hs +++ b/ouroboros-network/framework/lib/Ouroboros/Network/Server/Simple.hs @@ -14,9 +14,9 @@ module Ouroboros.Network.Server.Simple , ServerTracer (..) ) where -import Control.Applicative (Alternative) import Control.Concurrent.JobPool qualified as JobPool import Control.DeepSeq (NFData) +import Control.Monad (MonadPlus) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadSTM import Control.Monad.Class.MonadThrow @@ -43,13 +43,13 @@ data ServerTracer addr deriving Show with :: forall fd addr vNumber vData m a b. - ( Alternative (STM m), - MonadAsync m, + ( MonadAsync m, MonadDelay m, MonadEvaluate m, MonadFork m, MonadLabelledSTM m, MonadMask m, + MonadPlus (STM m), MonadTimer m, MonadThrow (STM m), NFData vData, diff --git a/ouroboros-network/framework/lib/Ouroboros/Network/Socket.hs b/ouroboros-network/framework/lib/Ouroboros/Network/Socket.hs index 989a180596e..a8f1d892926 100644 --- a/ouroboros-network/framework/lib/Ouroboros/Network/Socket.hs +++ b/ouroboros-network/framework/lib/Ouroboros/Network/Socket.hs @@ -63,7 +63,7 @@ import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadSTM.Strict import Control.DeepSeq (NFData) #if !defined(wasm32_HOST_ARCH) -import Control.Monad (unless, when) +import Control.Monad (MonadPlus, unless, when) #endif import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork @@ -264,13 +264,13 @@ data ConnectToArgs m fd addr vNumber vData = ConnectToArgs { -- Exceptions thrown by 'MuxApplication' are rethrown by 'connectToNode'. connectToNode :: forall muxMode vNumber vData fd addr m a b. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , Mx.MonadReadBuffer m , MonadSTM m , MonadTimer m @@ -303,13 +303,13 @@ connectToNode sn mkBearer args configureSock versions localAddr remoteAddr = -- to execute on a given connection. connectToNodeWithMux :: forall muxMode vNumber vData fd addr m a b x. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , Mx.MonadReadBuffer m , MonadSTM m , MonadTimer m @@ -366,13 +366,13 @@ connectToNodeWithMux sn mkBearer args configureSock versions localAddr remoteAdd -- Exceptions thrown by @'MuxApplication'@ are rethrown by @'connectTo'@. connectToNode' :: forall muxMode vNumber vData fd addr m a b. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadFork m , MonadEvaluate m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , Mx.MonadReadBuffer m , MonadSTM m , MonadTimer m @@ -401,13 +401,13 @@ connectToNode' sn mkBearer args versions as = connectToNodeWithMux' :: forall muxMode vNumber vData fd addr m a b x. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadFork m , MonadEvaluate m , MonadLabelledSTM m , MonadMask m + , MonadPlus (STM m) , Mx.MonadReadBuffer m , MonadSTM m , MonadTimer m diff --git a/ouroboros-network/framework/sim-tests/Test/Ouroboros/Network/Server/Sim.hs b/ouroboros-network/framework/sim-tests/Test/Ouroboros/Network/Server/Sim.hs index 47085e89ff4..11d7994776e 100644 --- a/ouroboros-network/framework/sim-tests/Test/Ouroboros/Network/Server/Sim.hs +++ b/ouroboros-network/framework/sim-tests/Test/Ouroboros/Network/Server/Sim.hs @@ -29,7 +29,7 @@ import Control.Concurrent.Class.MonadSTM qualified as LazySTM import Control.Concurrent.Class.MonadSTM.Strict import Control.DeepSeq (NFData) import Control.Exception (SomeAsyncException (..), SomeException (..)) -import Control.Monad (replicateM) +import Control.Monad (MonadPlus, replicateM) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -628,6 +628,7 @@ multinodeExperiment , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m + , MonadPlus (STM m) , acc ~ [req], resp ~ [req] , Ord peerAddr , PrettyShow peerAddr @@ -2300,7 +2301,7 @@ prop_server_accept_error (Fixed rnd) (AbsIOError ioerr) = -multiNodeSimTracer :: ( Alternative (STM m), Monad m, MonadFix m +multiNodeSimTracer :: ( Monad m, MonadFix m, MonadPlus (STM m) , MonadDelay m, MonadTimer m, MonadLabelledSTM m , MonadTraceSTM m, MonadMask m, MonadTime m , MonadThrow (STM m), MonadSay m, MonadAsync m diff --git a/ouroboros-network/framework/sim-tests/Test/Simulation/Network/Snocket.hs b/ouroboros-network/framework/sim-tests/Test/Simulation/Network/Snocket.hs index 47e87c74b4c..e0607b31673 100644 --- a/ouroboros-network/framework/sim-tests/Test/Simulation/Network/Snocket.hs +++ b/ouroboros-network/framework/sim-tests/Test/Simulation/Network/Snocket.hs @@ -18,8 +18,8 @@ module Test.Simulation.Network.Snocket , toBearerInfo ) where -import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadSTM.Strict +import Control.Monad import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -39,7 +39,6 @@ import Codec.Serialise qualified as Serialise import Data.ByteString.Lazy (ByteString) import Data.ByteString.Lazy qualified as BL import Data.Foldable (traverse_) -import Data.Functor (void) import Data.Map qualified as Map import Data.Maybe (isNothing) import Data.Set (Set) @@ -186,8 +185,7 @@ untilSuccess go = clientServerSimulation :: forall m addr payload. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFork m @@ -197,7 +195,7 @@ clientServerSimulation , MonadST m , MonadThrow (STM m) , MonadTimer m - + , MonadPlus (STM m) , Serialise payload , Eq payload , Show payload @@ -277,8 +275,9 @@ clientServerSimulation payloads = [ MiniProtocolInfo { miniProtocolNum = reqRespProtocolNum, miniProtocolDir = Mx.ResponderDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ]) Mx.stop @@ -331,8 +330,9 @@ clientServerSimulation payloads = [ MiniProtocolInfo { miniProtocolNum = reqRespProtocolNum, miniProtocolDir = Mx.InitiatorDirectionOnly, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, - miniProtocolCapability = Nothing + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, + miniProtocolCapability = Nothing, + miniProtocolWeight = 1 } ] diff --git a/ouroboros-network/framework/tests-lib/Test/Ouroboros/Network/ConnectionManager/Experiments.hs b/ouroboros-network/framework/tests-lib/Test/Ouroboros/Network/ConnectionManager/Experiments.hs index 9ff757b416b..4a325a20d4d 100644 --- a/ouroboros-network/framework/tests-lib/Test/Ouroboros/Network/ConnectionManager/Experiments.hs +++ b/ouroboros-network/framework/tests-lib/Test/Ouroboros/Network/ConnectionManager/Experiments.hs @@ -39,7 +39,7 @@ import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadSTM.Strict import Control.DeepSeq (NFData) import Control.Exception (AssertionFailed) -import Control.Monad (replicateM, (>=>)) +import Control.Monad (MonadPlus, replicateM, (>=>)) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -252,6 +252,7 @@ withInitiatorOnlyConnectionManager , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m + , MonadPlus (STM m) , Show name ) => name @@ -352,9 +353,10 @@ withInitiatorOnlyConnectionManager name timeouts trTracer tracer stdGen snocket [MiniProtocol { miniProtocolNum, miniProtocolStart = StartOnDemand, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, miniProtocolRun = reqRespInitiator miniProtocolNum - nextRequest + nextRequest, + miniProtocolWeight = 1 }] reqRespInitiator :: Mx.MiniProtocolNum @@ -437,6 +439,7 @@ withBidirectionalConnectionManager , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m, Show req + , MonadPlus (STM m) , NFData req , Show name ) @@ -580,11 +583,12 @@ withBidirectionalConnectionManager name timeouts [MiniProtocol { miniProtocolNum, miniProtocolStart = Mx.StartOnDemand, - miniProtocolLimits = Mx.MiniProtocolLimits maxBound, + miniProtocolLimits = Mx.MiniProtocolLimits maxBound Nothing, miniProtocolRun = reqRespInitiatorAndResponder miniProtocolNum accumulatorInit - nextRequest + nextRequest, + miniProtocolWeight = 1 }] reqRespInitiatorAndResponder @@ -741,7 +745,7 @@ unidirectionalExperiment , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m - + , MonadPlus (STM m) , acc ~ [req], resp ~ [req] , Ord peerAddr , PrettyShow peerAddr @@ -822,7 +826,7 @@ bidirectionalExperiment , MonadLabelledSTM m , MonadTraceSTM m , MonadSay m - + , MonadPlus (STM m) , acc ~ [req], resp ~ [req] , Ord peerAddr , PrettyShow peerAddr diff --git a/ouroboros-network/lib/Ouroboros/Network/Diffusion.hs b/ouroboros-network/lib/Ouroboros/Network/Diffusion.hs index 7e87dcfbecf..1b40822be81 100644 --- a/ouroboros-network/lib/Ouroboros/Network/Diffusion.hs +++ b/ouroboros-network/lib/Ouroboros/Network/Diffusion.hs @@ -20,11 +20,11 @@ module Ouroboros.Network.Diffusion ) where -import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadMVar (MonadMVar) import Control.Concurrent.Class.MonadSTM.Strict import Control.DeepSeq (NFData) import Control.Exception (IOException) +import Control.Monad (MonadPlus) import Control.Monad.Class.MonadAsync (Async, MonadAsync) import Control.Monad.Class.MonadAsync qualified as Async import Control.Monad.Class.MonadFork @@ -97,8 +97,7 @@ runM extraState extraDebugState extraPeers extraAPI extraFlags extraChurnArgs. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFix m @@ -106,6 +105,7 @@ runM , MonadLabelledSTM m , MonadTraceSTM m , MonadMask m + , MonadPlus (STM m) , MonadThrow (STM m) , MonadTime m , MonadTimer m diff --git a/ouroboros-network/tests/io/Test/Ouroboros/Network/Pipe.hs b/ouroboros-network/tests/io/Test/Ouroboros/Network/Pipe.hs index 99d33234757..b09a8c18987 100644 --- a/ouroboros-network/tests/io/Test/Ouroboros/Network/Pipe.hs +++ b/ouroboros-network/tests/io/Test/Ouroboros/Network/Pipe.hs @@ -95,9 +95,11 @@ demoProtocols chainSync = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = defaultMiniProtocolLimit + maximumIngressQueue = defaultMiniProtocolLimit, + burst = Nothing }, - miniProtocolRun = chainSync + miniProtocolRun = chainSync, + miniProtocolWeight = 1 } ] diff --git a/ouroboros-network/tests/io/Test/Ouroboros/Network/Socket.hs b/ouroboros-network/tests/io/Test/Ouroboros/Network/Socket.hs index 788c1cf2ea1..5d6df16bf8f 100644 --- a/ouroboros-network/tests/io/Test/Ouroboros/Network/Socket.hs +++ b/ouroboros-network/tests/io/Test/Ouroboros/Network/Socket.hs @@ -175,9 +175,11 @@ testProtocols1 chainSync = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemandAny, miniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = defaultMiniProtocolLimit + maximumIngressQueue = defaultMiniProtocolLimit, + burst = Nothing }, - miniProtocolRun = chainSync + miniProtocolRun = chainSync, + miniProtocolWeight = 1 } ] diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Diffusion/Node.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/Diffusion/Node.hs index 1d21078c834..b0b35dfb016 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Diffusion/Node.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/Diffusion/Node.hs @@ -36,10 +36,9 @@ module Test.Ouroboros.Network.Diffusion.Node , Node.ntnAddrToRelayAccessPoint ) where -import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadMVar (MonadMVar) import Control.Concurrent.Class.MonadSTM.Strict -import Control.Monad ((>=>)) +import Control.Monad (MonadPlus, (>=>)) import Control.Monad.Class.MonadAsync (MonadAsync (wait, withAsync)) import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -178,8 +177,7 @@ data Arguments extraChurnArgs extraFlags m = Arguments run :: forall extraState extraDebugState extraAPI extraPeers extraFlags extraChurnArgs exception resolver m. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadEvaluate m , MonadFix m @@ -193,7 +191,7 @@ run :: forall extraState extraDebugState extraAPI , MonadTimer m , MonadThrow (STM m) , MonadMVar m - + , MonadPlus (STM m) , Eq extraFlags , Monoid extraPeers , SupportsPeerSelectionState extraPeers NtNAddr diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Mux.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/Mux.hs index f39d3f550fc..873508c73dc 100644 --- a/ouroboros-network/tests/lib/Test/Ouroboros/Network/Mux.hs +++ b/ouroboros-network/tests/lib/Test/Ouroboros/Network/Mux.hs @@ -10,11 +10,10 @@ module Test.Ouroboros.Network.Mux (tests) where import Codec.Serialise (Serialise (..)) -import Data.Functor (void) import Data.Monoid.Synchronisation (FirstToFinish (..)) -import Control.Applicative (Alternative) import Control.Concurrent.Class.MonadSTM.Strict +import Control.Monad import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadFork import Control.Monad.Class.MonadSay @@ -75,16 +74,17 @@ testProtocols chainSync = miniProtocolNum = MiniProtocolNum 2, miniProtocolStart = StartOnDemand, miniProtocolLimits = MiniProtocolLimits { - maximumIngressQueue = 0xffff + maximumIngressQueue = 0xffff, + burst = Nothing }, - miniProtocolRun = chainSync + miniProtocolRun = chainSync, + miniProtocolWeight = 1 } ] demo :: forall m block. - ( Alternative (STM m) - , MonadAsync m + ( MonadAsync m , MonadDelay m , MonadCatch m , MonadEvaluate m @@ -97,6 +97,7 @@ demo :: forall m block. , MonadThrow (STM m) , MonadTime m , MonadTimer m + , MonadPlus (STM m) , Chain.HasHeader block , Serialise (Chain.HeaderHash block) , Serialise block