From a77ca48addd462d25f0420411be0b9a06f9d8885 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 11:11:48 +0200 Subject: [PATCH 1/9] mux: refactoring - SDUWithWantonState * added type signatures * change argument order of `processSingleWanton` * added `SDUWithWantonState` * added bangs to all `go` definitions --- network-mux/src/Network/Mux/Egress.hs | 79 ++++++++++++++++++--------- 1 file changed, 53 insertions(+), 26 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index f51b9972e1f..879c995f7a2 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -172,6 +172,11 @@ 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 + + -- | Process the messages from the mini protocols - there is a single -- shared FIFO that contains the items of work. This is processed so -- that each active demand gets a `maxSDU`s work of data processed @@ -206,14 +211,21 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva x:xs -> (x, xs) egressQueues'' | weight > 1 = (pred weight, queue) : rest | otherwise = rest - eSdu <- processSingleWanton sduSize mpc md d + eSdu <- processSingleWanton mpc md d sduSize case eSdu of - Right sdu | pbMaxBytes > 0 -> - -- we do not check if the protocol has any tokens to burst, - -- that is deferred to buildBatch below. - (sdu, egressQueues', True) <$ unGetTBQueue queue demand - | otherwise -> (sdu, egressQueues'', False) <$ writeTBQueue queue demand - Left sdu -> pure (sdu, egressQueues'', False) + NonEmptyWanton sdu + | pbMaxBytes > 0 + -> -- we do not check if the protocol has any tokens to + -- burst, that is deferred to buildBatch below. + (sdu, egressQueues', True) + <$ unGetTBQueue queue demand + + | otherwise + -> (sdu, egressQueues'', False) + <$ writeTBQueue queue demand + + EmptyWanton sdu -> + pure (sdu, egressQueues'', False) (egressQueues'', batch'') <- lift $ buildBatch (mkSingletonBatch sdu) egressQueues' burst start @@ -246,7 +258,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -- (e.g the SO_SNDBUF for Socket) or number of SDUs. -- buildBatch - :: SDUBatch -> [(Word8, EgressQueue m)] -> Bool -> Time -> m ([(Word8, EgressQueue m)], SDUBatch) + :: SDUBatch + -> [(Word8, EgressQueue m)] + -> Bool + -> Time + -> m ([(Word8, EgressQueue m)], SDUBatch) buildBatch batch0 egressQueues1 mBurst0 start = do (qs, batch) <- go batch0 egressQueues1 mBurst0 pure (qs, batch { getSdus = reverse (getSdus batch) }) @@ -257,12 +273,15 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva res <- f x if res then allM f xs else pure False - go :: SDUBatch -> [(Word8, EgressQueue m)] -> Bool -> m ([(Word8, EgressQueue m)], SDUBatch) + go :: SDUBatch + -> [(Word8, EgressQueue m)] + -> Bool + -> m ([(Word8, EgressQueue m)], SDUBatch) go !_batch [] !_burst = error "impossible" - go batch egressQueues _burst + go !batch egressQueues !_burst | getCount batch >= maxSDUsPerBatch || getSdusLength batch >= batchSize - = return (egressQueues, batch) - go batch egressQueues@((weight, queue):rest) mBurst = do + = return (egressQueues, batch) + go !batch egressQueues@((weight, queue):rest) !mBurst = do -- since the list of queues cycles, we only need to check the prefix -- to see if there is any more work to do. allEmpty0 <- atomically $ allM isEmptyTBQueue (snd <$> take numQueues egressQueues) @@ -278,20 +297,29 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva thisEmpty <- isEmptyTBQueue queue let boundedTokens = min sduSize . fromIntegral . min (fromIntegral $ maxBound @SDUSize) - step (!batch', !eSize) mx = do + step :: (SDUBatch, Either SDUSize SDUSize) + -> (SDUSize -> STM m SDUWithWantonState) + -- ^ process one Wanton, `SDUSize` instruments how + -- many bytes take from a `Wanton`. + -> ExceptT (SDUBatch, Bool) + (STM m) + (SDUBatch, Either SDUSize SDUSize) + step (!batch', !eSize) processWanton = do -- the first one is always free -- For Left's, we don't count the wanton bytes against the burst allowance -- to permit a full sdu in the first iteration let (size, consumedTokens) = either (, const 0) (, id) eSize - x <- lift $ mx size + x <- lift $ processWanton size case x of - Left sdu -> do + EmptyWanton sdu -> do + -- the `Wanton` is empty lift $ modifyTVar wBucket \tokens -> let tokens' = tokens - consumedTokens (fromIntegral (msLength sdu)) in assert (tokens >= consumedTokens (fromIntegral $ msLength sdu)) tokens' throwE (mkSingletonBatch sdu <> batch', not thisEmpty) - Right sdu -> do + NonEmptyWanton sdu -> do + -- the `Wanton` is non-empty nextSdu <- lift $ stateTVar wBucket \tokens -> let tokens' = tokens - consumedTokens (fromIntegral (msLength sdu)) in assert (tokens >= consumedTokens (fromIntegral $ msLength sdu)) @@ -303,6 +331,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva lift $ writeTBQueue queue demand throwE (batch'', True) else pure (batch'', Right nextSdu) + either pure (error "impossible") =<< runExceptT do sduSize0 <- lift $ stateTVar wBucket \tokens -> @@ -321,7 +350,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva lift $ writeTBQueue queue demand throwE (batch, True) foldM step (batch, sduSize0) - (repeat (\sduSize' -> processSingleWanton sduSize' mpc md d)) + (repeat (processSingleWanton mpc md d)) if weight > 1 && goAgain then let egressQueues' = (pred weight, queue) : rest @@ -333,27 +362,25 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -- data remaining requeue the `TranslocationServiceRequest` (this -- ensures that any other items on the queue will get some service -- first. -processSingleWanton :: (MonadSTM m) - => SDUSize - -> MiniProtocolNum +processSingleWanton :: MonadSTM m + => MiniProtocolNum -> MiniProtocolDir -> Wanton m - -- Right: more sdu's remain; Left: finished - -> STM m (Either SDU SDU) -processSingleWanton sduSize - mpc md wanton = do + -> SDUSize + -> STM m SDUWithWantonState +processSingleWanton mpc md wanton sduSize = do (blob, wrap) <- do -- extract next SDU d <- readTVar (want wanton) let (frag, rest) = BL.splitAt (fromIntegral sduSize) d -- if more to process then enqueue remaining work if BL.null rest - then (frag, Left) <$ writeTVar (want wanton) BL.empty + then (frag, EmptyWanton) <$ writeTVar (want 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. - (frag, Right) <$ writeTVar (want wanton) rest + (frag, NonEmptyWanton) <$ writeTVar (want wanton) rest let sdu = SDU { msHeader = SDUHeader { mhTimestamp = RemoteClockModel 0, From 67766401595ef5f721f5e34e0f537ef9a666c132 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 12:04:55 +0200 Subject: [PATCH 2/9] mux: refactoring - NextSDUSize, TokenSize Added dedicated API: * `NextSDUSize`, `nextSDUSizeToSDUSize` * `TokenSize` (type alias), `consumedTokens` --- network-mux/src/Network/Mux/Egress.hs | 84 ++++++++++++++++++++------- 1 file changed, 62 insertions(+), 22 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index 879c995f7a2..d89330ca4df 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -26,7 +26,6 @@ import Control.Monad.Trans.Class import Control.Monad.Trans.Except import Control.Monad.Trans.State import Data.ByteString.Lazy qualified as BL -import Data.Either (fromRight) import Data.List (tails) import Data.Monoid.Synchronisation import Data.Word (Word32, Word8) @@ -144,7 +143,7 @@ data Wanton m = Wanton { want :: !(StrictTVar m BL.ByteString), wLastSent :: !(StrictTVar m Time), -- ^ the last time the protocol has sent a message - wBucket :: !(StrictTVar m Word32) + wBucket :: !(StrictTVar m TokenSize) -- ^ the number of tokens available to burst } @@ -176,6 +175,38 @@ sduLength sdu = fromIntegral msHeaderLength + fromIntegral (msLength sdu) -- 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 #-} -- | Process the messages from the mini protocols - there is a single -- shared FIFO that contains the items of work. This is processed so @@ -295,61 +326,70 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva (batch', goAgain) <- atomically do delta <- (start `diffTime`) <$> stateTVar wLastSent (, start) thisEmpty <- isEmptyTBQueue queue - let boundedTokens = min sduSize . fromIntegral . min (fromIntegral $ maxBound @SDUSize) - step :: (SDUBatch, Either SDUSize SDUSize) + let -- take current `SDUBatch` and `NextSDUSize`, and compute + -- the new `SDUBatch` and `NextSDUSize` or return the + -- `SDUBatch` and a boolean value which indicates if any + -- step was taken. + step :: (SDUBatch, NextSDUSize) -> (SDUSize -> STM m SDUWithWantonState) -- ^ process one Wanton, `SDUSize` instruments how -- many bytes take from a `Wanton`. -> ExceptT (SDUBatch, Bool) (STM m) - (SDUBatch, Either SDUSize SDUSize) - step (!batch', !eSize) processWanton = do + (SDUBatch, NextSDUSize) + step (!batch', !nextSDUSize) processWanton = do -- the first one is always free -- For Left's, we don't count the wanton bytes against the burst allowance -- to permit a full sdu in the first iteration - let (size, consumedTokens) = either (, const 0) (, id) eSize - x <- lift $ processWanton size + x <- lift $ processWanton (nextSDUSizeToSDUSize sduSize nextSDUSize) case x of EmptyWanton sdu -> do -- the `Wanton` is empty lift $ modifyTVar wBucket \tokens -> - let tokens' = tokens - consumedTokens (fromIntegral (msLength sdu)) - in assert (tokens >= consumedTokens (fromIntegral $ msLength sdu)) + let consumed, tokens' :: TokenSize + consumed = consumedTokens nextSDUSize sdu + tokens' = tokens - consumed + in assert (tokens >= consumed) tokens' throwE (mkSingletonBatch sdu <> batch', not thisEmpty) NonEmptyWanton sdu -> do -- the `Wanton` is non-empty nextSdu <- lift $ stateTVar wBucket \tokens -> - let tokens' = tokens - consumedTokens (fromIntegral (msLength sdu)) - in assert (tokens >= consumedTokens (fromIntegral $ msLength sdu)) - (boundedTokens tokens', 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 - -- there is more payload, but burst allowance has been exhausted + -- burst allowance has been exhausted, next + -- SDU would be too small lift $ writeTBQueue queue demand throwE (batch'', True) - else pure (batch'', Right nextSdu) + else pure (batch'', BurstSize nextSdu) either pure (error "impossible") =<< runExceptT do - sduSize0 <- lift $ stateTVar wBucket \tokens -> - let tokens' = truncate $ + nextSduSize <- lift $ stateTVar wBucket \tokens -> + let tokens' :: TokenSize + tokens' = truncate $ min (fromIntegral pbMaxBytes) (fromIntegral tokens + fromIntegral pbRefillRate * toDouble delta) -- we leverage burst and deduct credits only where there is contention -- between protocols - sduSize0 | mBurst = Right $ boundedTokens tokens' - | otherwise = Left sduSize - in (sduSize0, tokens') - when (fromRight maxBound sduSize0 <= burstMinSdu) do + nextSduSize :: NextSDUSize + nextSduSize | mBurst = BurstSize $ computeSDUSize sduSize tokens' + | otherwise = BearerSize + in (nextSduSize, tokens') + when (nextSDUSizeToSDUSize maxBound nextSduSize <= burstMinSdu) 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. lift $ writeTBQueue queue demand throwE (batch, True) - foldM step (batch, sduSize0) + foldM step (batch, nextSduSize) (repeat (processSingleWanton mpc md d)) if weight > 1 && goAgain From ef56a32136e0a28691760da22ad42b7bd03e8e5e Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 14:30:57 +0200 Subject: [PATCH 3/9] mux: refactoring - explicit burst loop An explicit loop is easier to follow. We also avoid calling `error "impossible"`. --- network-mux/src/Network/Mux/Egress.hs | 141 +++++++++++++------------- 1 file changed, 70 insertions(+), 71 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index d89330ca4df..e0616519f8f 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -291,11 +291,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva buildBatch :: SDUBatch -> [(Word8, EgressQueue m)] - -> Bool + -> Bool -- ^ can we burst -> Time -> m ([(Word8, EgressQueue m)], SDUBatch) - buildBatch batch0 egressQueues1 mBurst0 start = do - (qs, batch) <- go batch0 egressQueues1 mBurst0 + buildBatch batch0 egressQueues1 canBurst0 start = do + (qs, batch) <- go batch0 egressQueues1 canBurst0 pure (qs, batch { getSdus = reverse (getSdus batch) }) where allM f = \case @@ -312,7 +312,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva go !batch egressQueues !_burst | getCount batch >= maxSDUsPerBatch || getSdusLength batch >= batchSize = return (egressQueues, batch) - go !batch egressQueues@((weight, queue):rest) !mBurst = do + go !batch egressQueues@((weight, queue):rest) !canBurst = do -- since the list of queues cycles, we only need to check the prefix -- to see if there is any more work to do. allEmpty0 <- atomically $ allM isEmptyTBQueue (snd <$> take numQueues egressQueues) @@ -322,80 +322,79 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva mResult <- atomically $ tryReadTBQueue queue case mResult of Nothing -> go batch rest False - Just demand@(TLSRDemand mpc md d@Wanton { wLastSent, wBucket } (ProtocolBurst pbMaxBytes pbRefillRate)) -> do + Just demand@(TLSRDemand _ _ + Wanton { wLastSent, wBucket } + ProtocolBurst { pbMaxBytes, pbRefillRate }) -> do (batch', goAgain) <- atomically do delta <- (start `diffTime`) <$> stateTVar wLastSent (, start) - thisEmpty <- isEmptyTBQueue queue - - let -- take current `SDUBatch` and `NextSDUSize`, and compute - -- the new `SDUBatch` and `NextSDUSize` or return the - -- `SDUBatch` and a boolean value which indicates if any - -- step was taken. - step :: (SDUBatch, NextSDUSize) - -> (SDUSize -> STM m SDUWithWantonState) - -- ^ process one Wanton, `SDUSize` instruments how - -- many bytes take from a `Wanton`. - -> ExceptT (SDUBatch, Bool) - (STM m) - (SDUBatch, NextSDUSize) - step (!batch', !nextSDUSize) processWanton = do - -- the first one is always free - -- For Left's, we don't count the wanton bytes against the burst allowance - -- to permit a full sdu in the first iteration - x <- lift $ processWanton (nextSDUSizeToSDUSize sduSize nextSDUSize) - case x of - EmptyWanton sdu -> do - -- the `Wanton` is empty - lift $ modifyTVar wBucket \tokens -> - let consumed, tokens' :: TokenSize - consumed = consumedTokens nextSDUSize sdu - tokens' = tokens - consumed - in assert (tokens >= consumed) - tokens' - throwE (mkSingletonBatch sdu <> batch', not thisEmpty) - NonEmptyWanton sdu -> do - -- the `Wanton` is non-empty - nextSdu <- lift $ stateTVar wBucket \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 - lift $ writeTBQueue queue demand - throwE (batch'', True) - else pure (batch'', BurstSize nextSdu) - - either pure (error "impossible") - =<< runExceptT do - nextSduSize <- lift $ stateTVar wBucket \tokens -> - let tokens' :: TokenSize - tokens' = truncate $ - min (fromIntegral pbMaxBytes) - (fromIntegral tokens + fromIntegral pbRefillRate * toDouble delta) - -- we leverage burst and deduct credits only where there is contention - -- between protocols - nextSduSize :: NextSDUSize - nextSduSize | mBurst = BurstSize $ computeSDUSize sduSize tokens' - | otherwise = BearerSize - in (nextSduSize, tokens') - when (nextSDUSizeToSDUSize maxBound nextSduSize <= burstMinSdu) 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. - lift $ writeTBQueue queue demand - throwE (batch, True) - foldM step (batch, nextSduSize) - (repeat (processSingleWanton mpc md d)) + + nextSduSize <- stateTVar wBucket \tokens -> + let tokens' :: TokenSize + tokens' = truncate $ + min (fromIntegral pbMaxBytes) + (fromIntegral tokens + fromIntegral pbRefillRate * toDouble delta) + -- we leverage burst and deduct credits only where there is contention + -- between protocols + nextSduSize :: NextSDUSize + nextSduSize | canBurst = BurstSize $ computeSDUSize sduSize tokens' + | otherwise = 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, True) + else + burstLoop demand batch nextSduSize if weight > 1 && goAgain then let egressQueues' = (pred weight, queue) : rest in go batch' egressQueues' False else go batch' rest False + where + -- burst SDUs from a single mini-protocol until we consume all tokens + -- (`TokenSize`). + burstLoop :: TranslocationServiceRequest m + -> SDUBatch + -> NextSDUSize + -> STM m (SDUBatch, Bool) + burstLoop demand@(TLSRDemand miniProtocolNum miniProtocolDir want@Wanton {wBucket} _) !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 wBucket \tokens -> + let consumed, tokens' :: TokenSize + consumed = consumedTokens nextSDUSize sdu + tokens' = tokens - consumed + in assert (tokens >= consumed) + tokens' + thisEmpty <- isEmptyTBQueue queue + return (mkSingletonBatch sdu <> batch', not thisEmpty) + + NonEmptyWanton sdu -> do + -- the `Wanton` is non-empty + nextSdu <- stateTVar wBucket \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'', True) + else burstLoop demand batch'' (BurstSize nextSdu) -- | Pull a `maxSDU`s worth of data out out the `Wanton` - if there is From 0386b0827ba910f67eab1720cc4dd8dabe94457c Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 15:14:40 +0200 Subject: [PATCH 4/9] mux: refactoring - replaced custom allM Replaced a custom `allM` with `foldMap All . traverse`. In the future we can fuse `foldMap` with `traverse`. --- network-mux/src/Network/Mux/Egress.hs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index e0616519f8f..6db0d2f8946 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -23,10 +23,10 @@ import Control.Applicative import Control.Exception import Control.Monad import Control.Monad.Trans.Class -import Control.Monad.Trans.Except import Control.Monad.Trans.State 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) @@ -298,12 +298,6 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva (qs, batch) <- go batch0 egressQueues1 canBurst0 pure (qs, batch { getSdus = reverse (getSdus batch) }) where - allM f = \case - [] -> pure True - (x:xs) -> do - res <- f x - if res then allM f xs else pure False - go :: SDUBatch -> [(Word8, EgressQueue m)] -> Bool @@ -315,7 +309,12 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva go !batch egressQueues@((weight, queue):rest) !canBurst = do -- since the list of queues cycles, we only need to check the prefix -- to see if there is any more work to do. - allEmpty0 <- atomically $ allM isEmptyTBQueue (snd <$> take numQueues egressQueues) + -- + -- TODO: we could use `atomically $ foldMap (fmap All . isEmptyTBQueue + -- . snd)` if `transformers` had `Monoid a => Monoid (m a)` instance. + All allEmpty0 <- + atomically $ getAp $ foldMap (Ap . fmap All . isEmptyTBQueue . snd) + (take numQueues egressQueues) if allEmpty0 then return (egressQueues, batch) else do From d1e1d0e42255548108f3c0c4422fa624e37aeda3 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 15:38:31 +0200 Subject: [PATCH 5/9] mux: refactoring - moved constants Move constants to top level bindings: * `maxSDUsPerBatch` * `burstMinSdu` --- network-mux/src/Network/Mux/Egress.hs | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index 6db0d2f8946..2bca946298d 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -208,6 +208,18 @@ consumedTokens BurstSize{} sdu = fromIntegral (msLength sdu) consumedTokens BearerSize _sdu = 0 {-# INLINE consumedTokens #-} + +-- | 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 -- shared FIFO that contains the items of work. This is processed so -- that each active demand gets a `maxSDU`s work of data processed @@ -269,14 +281,9 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva numQueues :: Int numQueues = length egressQueues0 - maxSDUsPerBatch :: Int - maxSDUsPerBatch = 100 - toDouble :: DiffTime -> Double toDouble = realToFrac - burstMinSdu = truncate @Double @SDUSize $ fromIntegral msHeaderLength / 0.02 - -- Build a batch of SDUs to submit in one go to the bearer. -- Streams which are permitted to burst will have that many -- sdu's serviced back-to-back before the scheduler moves to process the From 3335505a64ee860669f1e0cd92a4da738c1bb77a Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Fri, 3 Apr 2026 16:59:43 +0200 Subject: [PATCH 6/9] mux: refactoring - CanBatch, CanBurst Replaced `Bool` with: * `CanBatch` - can we batch more SDUs from different mini-protocols. * `CanBurst` - can we burst more SDUs from a single mini-protocol. --- network-mux/src/Network/Mux/Egress.hs | 75 +++++++++++++++++---------- 1 file changed, 47 insertions(+), 28 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index 2bca946298d..00da1422872 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -209,6 +209,18 @@ 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 @@ -245,7 +257,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva (zip (tails egressQueues) (take numQueues (snd <$> egressQueues))) start <- lift getMonotonicTime - (sdu, egressQueues', burst) <- lift $ atomically do + (sdu, egressQueues', canBatch) <- lift $ atomically do job <- runFirstToFinish jobs case job of (egressQueues', demand@(TLSRDemand mpc md d (ProtocolBurst pbMaxBytes _pbRefillRate))) -> do @@ -260,18 +272,18 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva | pbMaxBytes > 0 -> -- we do not check if the protocol has any tokens to -- burst, that is deferred to buildBatch below. - (sdu, egressQueues', True) + (sdu, egressQueues', BatchAllowed) <$ unGetTBQueue queue demand | otherwise - -> (sdu, egressQueues'', False) + -> (sdu, egressQueues'', BatchNotAllowed) <$ writeTBQueue queue demand EmptyWanton sdu -> - pure (sdu, egressQueues'', False) + pure (sdu, egressQueues'', BatchNotAllowed) (egressQueues'', batch'') <- - lift $ buildBatch (mkSingletonBatch sdu) egressQueues' burst start + lift $ buildBatch (mkSingletonBatch sdu) egressQueues' canBatch start put egressQueues'' void . lift $ writeMany tracer timeout (getSdus batch'') delta <- (`diffTime` start) <$> lift getMonotonicTime @@ -298,27 +310,24 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva buildBatch :: SDUBatch -> [(Word8, EgressQueue m)] - -> Bool -- ^ can we burst + -> CanBatch -- ^ can we batch more SDUs -> Time -> m ([(Word8, EgressQueue m)], SDUBatch) - buildBatch batch0 egressQueues1 canBurst0 start = do - (qs, batch) <- go batch0 egressQueues1 canBurst0 + buildBatch batch0 egressQueues1 canBatch0 start = do + (qs, batch) <- go batch0 egressQueues1 canBatch0 pure (qs, batch { getSdus = reverse (getSdus batch) }) where go :: SDUBatch -> [(Word8, EgressQueue m)] - -> Bool + -> CanBatch -> m ([(Word8, EgressQueue m)], SDUBatch) - go !_batch [] !_burst = error "impossible" - go !batch egressQueues !_burst + go !_batch [] !_canBatch = error "impossible" + go !batch egressQueues !_canBatch | getCount batch >= maxSDUsPerBatch || getSdusLength batch >= batchSize = return (egressQueues, batch) - go !batch egressQueues@((weight, queue):rest) !canBurst = do + 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. - -- - -- TODO: we could use `atomically $ foldMap (fmap All . isEmptyTBQueue - -- . snd)` if `transformers` had `Monoid a => Monoid (m a)` instance. All allEmpty0 <- atomically $ getAp $ foldMap (Ap . fmap All . isEmptyTBQueue . snd) (take numQueues egressQueues) @@ -327,11 +336,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva else do mResult <- atomically $ tryReadTBQueue queue case mResult of - Nothing -> go batch rest False + Nothing -> go batch rest BatchNotAllowed Just demand@(TLSRDemand _ _ Wanton { wLastSent, wBucket } ProtocolBurst { pbMaxBytes, pbRefillRate }) -> do - (batch', goAgain) <- atomically do + (batch', canBurst) <- atomically do delta <- (start `diffTime`) <$> stateTVar wLastSent (, start) nextSduSize <- stateTVar wBucket \tokens -> @@ -342,8 +351,10 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -- we leverage burst and deduct credits only where there is contention -- between protocols nextSduSize :: NextSDUSize - nextSduSize | canBurst = BurstSize $ computeSDUSize sduSize tokens' - | otherwise = BearerSize + nextSduSize = + case canBatch of + BatchAllowed -> BurstSize $ computeSDUSize sduSize tokens' + BatchNotAllowed -> BearerSize in (nextSduSize, tokens') if nextSDUSizeToSDUSize maxBound nextSduSize <= burstMinSdu then do @@ -351,21 +362,26 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -- 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, True) + return (batch, BurstAllowed) else burstLoop demand batch nextSduSize - if weight > 1 && goAgain - then let egressQueues' = (pred weight, queue) : rest - in go batch' egressQueues' False - else go batch' rest False + 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, Bool) + -> STM m (SDUBatch, CanBurst) burstLoop demand@(TLSRDemand miniProtocolNum miniProtocolDir want@Wanton {wBucket} _) !batch' !nextSDUSize = do -- The first SDU is always free. For `BearerSize` (no bursting), -- we don't count the wanton bytes against the burst allowance @@ -382,8 +398,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva tokens' = tokens - consumed in assert (tokens >= consumed) tokens' - thisEmpty <- isEmptyTBQueue queue - return (mkSingletonBatch sdu <> batch', not thisEmpty) + continue <- (\case + True -> BurstNotAllowed + False -> BurstAllowed) + <$> isEmptyTBQueue queue + return (mkSingletonBatch sdu <> batch', continue) NonEmptyWanton sdu -> do -- the `Wanton` is non-empty @@ -399,7 +418,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -- burst allowance has been exhausted, next -- SDU would be too small writeTBQueue queue demand - return (batch'', True) + return (batch'', BurstAllowed) else burstLoop demand batch'' (BurstSize nextSdu) From f3d07ee22fbe484ec92998570873e132254f66e1 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Tue, 7 Apr 2026 14:00:03 +0200 Subject: [PATCH 7/9] mux: refactoring - explicit muxer loop Use an explicit loop in `muxer`, as a result `transformers` dependency is gone. --- network-mux/network-mux.cabal | 1 - network-mux/src/Network/Mux/Egress.hs | 46 ++++++++++++++------------- 2 files changed, 24 insertions(+), 23 deletions(-) diff --git a/network-mux/network-mux.cabal b/network-mux/network-mux.cabal index 5fb51ed3757..76b152d0e0f 100644 --- a/network-mux/network-mux.cabal +++ b/network-mux/network-mux.cabal @@ -67,7 +67,6 @@ library statistics-linreg >=0.3 && <0.4, strict, time >=1.9.1 && <1.16, - transformers, vector >=0.12 && <0.14, if os(windows) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index 00da1422872..e1e083c604e 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -21,9 +21,6 @@ module Network.Mux.Egress import Control.Applicative import Control.Exception -import Control.Monad -import Control.Monad.Trans.Class -import Control.Monad.Trans.State import Data.ByteString.Lazy qualified as BL import Data.List (tails) import Data.Monoid (All (..), Ap (..)) @@ -251,13 +248,25 @@ muxer -> Bearer m -> m void muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterval } = - withTimeoutSerial $ \timeout -> (`evalStateT` cycle egressQueues0) $ forever do - egressQueues <- get - let jobs = foldMap (FirstToFinish . traverse readTBQueue) - (zip (tails egressQueues) (take numQueues (snd <$> egressQueues))) + withTimeoutSerial $ \timeout -> muxerLoop timeout (cycle egressQueues0) + where + numQueues :: Int + numQueues = length egressQueues0 + toDouble :: DiffTime -> Double + toDouble = realToFrac - start <- lift getMonotonicTime - (sdu, egressQueues', canBatch) <- lift $ atomically do + -- 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 + (sdu, egressQueues', canBatch) <- atomically do + let jobs :: FirstToFinish (STM m) + ([(Word8, EgressQueue m)], TranslocationServiceRequest m) + jobs = foldMap (FirstToFinish . traverse readTBQueue) + (zip (tails egressQueues) (take numQueues (snd <$> egressQueues))) job <- runFirstToFinish jobs case job of (egressQueues', demand@(TLSRDemand mpc md d (ProtocolBurst pbMaxBytes _pbRefillRate))) -> do @@ -282,19 +291,12 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva EmptyWanton sdu -> pure (sdu, egressQueues'', BatchNotAllowed) - (egressQueues'', batch'') <- - lift $ buildBatch (mkSingletonBatch sdu) egressQueues' canBatch start - put egressQueues'' - void . lift $ writeMany tracer timeout (getSdus batch'') - delta <- (`diffTime` start) <$> lift getMonotonicTime - lift . threadDelay $ egressInterval - delta - - where - numQueues :: Int - numQueues = length egressQueues0 - - toDouble :: DiffTime -> Double - toDouble = realToFrac + (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. -- Streams which are permitted to burst will have that many From 63fad3f94865a17f7c94a205175194a81b7e9126 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Tue, 7 Apr 2026 14:20:14 +0200 Subject: [PATCH 8/9] mux: refactoring - muxer loop Refactor `muxerLoop` It's easier to read `[(EgressQueue m, [(Word8, Egress m)])]`, than the other way around, since then it's the `EgressQueue` we read and its tail much like `x : xs`. This also avoids using `traverse` over tuple which is a bit surprising. --- network-mux/src/Network/Mux/Egress.hs | 38 ++++++++++++++++++--------- 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index e1e083c604e..65595e5e4f7 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -263,17 +263,31 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva muxerLoop timeout egressQueues = do start <- getMonotonicTime (sdu, egressQueues', canBatch) <- atomically do - let jobs :: FirstToFinish (STM m) - ([(Word8, EgressQueue m)], TranslocationServiceRequest m) - jobs = foldMap (FirstToFinish . traverse readTBQueue) - (zip (tails egressQueues) (take numQueues (snd <$> egressQueues))) - job <- runFirstToFinish jobs + 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 + + -- read first available `EgressQueue` and return its tail + job <- runFirstToFinish + . foldMap + ( FirstToFinish + . \(egressQueue, egressQueues') -> + (,egressQueues') <$> readTBQueue egressQueue + ) + $ available case job of - (egressQueues', demand@(TLSRDemand mpc md d (ProtocolBurst pbMaxBytes _pbRefillRate))) -> do - let ((weight, queue), rest) = assert (weight > 0) case egressQueues' of - [] -> error "impossible" - x:xs -> (x, xs) - egressQueues'' | weight > 1 = (pred weight, queue) : rest + (demand@(TLSRDemand mpc md d (ProtocolBurst pbMaxBytes _pbRefillRate)) + , 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 @@ -282,11 +296,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -> -- we do not check if the protocol has any tokens to -- burst, that is deferred to buildBatch below. (sdu, egressQueues', BatchAllowed) - <$ unGetTBQueue queue demand + <$ unGetTBQueue egressQueue demand | otherwise -> (sdu, egressQueues'', BatchNotAllowed) - <$ writeTBQueue queue demand + <$ writeTBQueue egressQueue demand EmptyWanton sdu -> pure (sdu, egressQueues'', BatchNotAllowed) From 20e116a887b695e87bf1f1b4f3a32e0b96f568b0 Mon Sep 17 00:00:00 2001 From: Marcin Szamotulski Date: Wed, 8 Apr 2026 10:18:26 +0200 Subject: [PATCH 9/9] mux: refactoring - added record fields Use DuplicateRecordFields for: * `Wanton` * `TranslocationServiceRequest` * `ProtocolBurst` These are all internal data types, so we won't force it on `network-mux` user. --- network-mux/src/Network/Mux.hs | 1 + network-mux/src/Network/Mux/Egress.hs | 62 ++++++++++++++++----------- network-mux/src/Network/Mux/Types.hs | 8 ++-- 3 files changed, 41 insertions(+), 30 deletions(-) diff --git a/network-mux/src/Network/Mux.hs b/network-mux/src/Network/Mux.hs index 4c1b3f7d0f9..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 #-} diff --git a/network-mux/src/Network/Mux/Egress.hs b/network-mux/src/Network/Mux/Egress.hs index 65595e5e4f7..d837d11f5bb 100644 --- a/network-mux/src/Network/Mux/Egress.hs +++ b/network-mux/src/Network/Mux/Egress.hs @@ -1,5 +1,6 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} @@ -130,18 +131,23 @@ 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) !ProtocolBurst +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. data Wanton m = Wanton { - want :: !(StrictTVar m BL.ByteString), - wLastSent :: !(StrictTVar m Time), - -- ^ the last time the protocol has sent a message - wBucket :: !(StrictTVar m TokenSize) - -- ^ the number of tokens available to burst + 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 } @@ -279,7 +285,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva ) $ available case job of - (demand@(TLSRDemand mpc md d (ProtocolBurst pbMaxBytes _pbRefillRate)) + (demand@(TLSRDemand mpc md d ProtocolBurst{maxBytes}) , egressQueues' ) -> do let ((weight, egressQueue), rest) = assert (weight > 0) @@ -292,7 +298,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva eSdu <- processSingleWanton mpc md d sduSize case eSdu of NonEmptyWanton sdu - | pbMaxBytes > 0 + | maxBytes > 0 -> -- we do not check if the protocol has any tokens to -- burst, that is deferred to buildBatch below. (sdu, egressQueues', BatchAllowed) @@ -349,21 +355,21 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva (take numQueues egressQueues) if allEmpty0 then return (egressQueues, batch) - else do - mResult <- atomically $ tryReadTBQueue queue - case mResult of + else + atomically (tryReadTBQueue queue) >>= \case Nothing -> go batch rest BatchNotAllowed - Just demand@(TLSRDemand _ _ - Wanton { wLastSent, wBucket } - ProtocolBurst { pbMaxBytes, pbRefillRate }) -> do + Just demand@TLSRDemand { + wanton = Wanton { lastSent, burstBucket }, + protocolBurst = ProtocolBurst { maxBytes, refillRate } + } -> do (batch', canBurst) <- atomically do - delta <- (start `diffTime`) <$> stateTVar wLastSent (, start) + delta <- (start `diffTime`) <$> stateTVar lastSent (, start) - nextSduSize <- stateTVar wBucket \tokens -> + nextSduSize <- stateTVar burstBucket \tokens -> let tokens' :: TokenSize tokens' = truncate $ - min (fromIntegral pbMaxBytes) - (fromIntegral tokens + fromIntegral pbRefillRate * toDouble delta) + min (fromIntegral maxBytes) + (fromIntegral tokens + fromIntegral refillRate * toDouble delta) -- we leverage burst and deduct credits only where there is contention -- between protocols nextSduSize :: NextSDUSize @@ -398,7 +404,11 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva -> SDUBatch -> NextSDUSize -> STM m (SDUBatch, CanBurst) - burstLoop demand@(TLSRDemand miniProtocolNum miniProtocolDir want@Wanton {wBucket} _) !batch' !nextSDUSize = do + 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 @@ -408,7 +418,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva case x of EmptyWanton sdu -> do -- the `Wanton` is empty - modifyTVar wBucket \tokens -> + modifyTVar burstBucket \tokens -> let consumed, tokens' :: TokenSize consumed = consumedTokens nextSDUSize sdu tokens' = tokens - consumed @@ -422,7 +432,7 @@ muxer egressQueues0 tracer Bearer { writeMany, sduSize, batchSize, egressInterva NonEmptyWanton sdu -> do -- the `Wanton` is non-empty - nextSdu <- stateTVar wBucket \tokens -> + nextSdu <- stateTVar burstBucket \tokens -> let consumed, tokens' :: TokenSize consumed = consumedTokens nextSDUSize sdu tokens' = tokens - consumed @@ -448,19 +458,19 @@ processSingleWanton :: MonadSTM m -> Wanton m -> SDUSize -> STM m SDUWithWantonState -processSingleWanton mpc md wanton sduSize = do +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 (frag, EmptyWanton) <$ 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. - (frag, NonEmptyWanton) <$ writeTVar (want wanton) rest + (frag, NonEmptyWanton) <$ writeTVar wanton rest let sdu = SDU { msHeader = SDUHeader { mhTimestamp = RemoteClockModel 0, diff --git a/network-mux/src/Network/Mux/Types.hs b/network-mux/src/Network/Mux/Types.hs index 61b60de28c8..4af8c9cfd94 100644 --- a/network-mux/src/Network/Mux/Types.hs +++ b/network-mux/src/Network/Mux/Types.hs @@ -106,10 +106,10 @@ data MiniProtocolLimits = data ProtocolBurst = ProtocolBurst { - pbMaxBytes :: !Word32, - -- ^ token bucket max size - pbRefillRate :: !Word32 - -- ^ token bucket refill rate, [1/s] + maxBytes :: !Word32, + -- ^ token bucket max size + refillRate :: !Word32 + -- ^ token bucket refill rate, [1/s] } deriving (Eq, Show)