diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,5 +1,10 @@ # Revision history for network-control +## 0.1.3 + +* Simplify `maybeOpenRxWindow` and improve docs + [#7](https://github.com/kazu-yamamoto/network-control/pull/7) + ## 0.1.2 * introducing a minimum size for window update diff --git a/Network/Control/Flow.hs b/Network/Control/Flow.hs --- a/Network/Control/Flow.hs +++ b/Network/Control/Flow.hs @@ -81,25 +81,55 @@ -- | Flow for receiving. -- +-- The goal of 'RxFlow' is to ensure that our network peer does not send us data +-- faster than we can consume it. We therefore impose a maximum number of +-- unconsumed bytes that we are willing to receive from the peer, which we refer +-- to as the buffer size: +-- -- @ --- rxfBufSize --- |------------------------| --- --------------------------------------> --- ^ ^ ^ --- rxfConsumed rxfReceived rxfLimit +-- rxfBufSize +-- |---------------------------| +-- --------------------------------------------> +-- ^ ^ +-- rxfConsumed rxvReceived +-- @ -- --- |-----------| The size which the peer can send --- rxWindowSize +-- The peer does not know of course how many bytes we have consumed of the data +-- that they sent us, so they keep track of their own limit of how much data +-- they are allowed to send. We keep track of this limit also: +-- -- @ +-- rxfBufSize +-- |---------------------------| +-- --------------------------------------------> +-- ^ ^ ^ +-- rxfConsumed rxvReceived | +-- rxfLimit +-- @ +-- +-- Each time we receive data from the peer, we check that they do not exceed the +-- limit ('checkRxLimit'). When we consume data, we periodically send the peer +-- an update (known as a _window update_) of what their new limit is +-- ('maybeOpenRxWindow'). To decrease overhead, we only this if the window +-- update is at least half the window size. data RxFlow = RxFlow { rxfBufSize :: Int - -- ^ Receive buffer size. + -- ^ Maxinum number of unconsumed bytes the peer can send us + -- + -- See discussion above for details. , rxfConsumed :: Int - -- ^ The total size which the application is consumed. + -- ^ How much of the data that the peer has sent us have we consumed? + -- + -- This is an absolute number: the total about of bytes consumed over the + -- lifetime of the connection or stream (i.e., not relative to the window). , rxfReceived :: Int - -- ^ The total already-received size. + -- ^ How much data have we received from the peer? + -- + -- Like 'rxfConsumed', this is an absolute number. , rxfLimit :: Int - -- ^ The total size which can be recived. + -- ^ Current limit on how many bytes the peer is allowed to send us. + -- + -- Like 'rxfConsumed, this is an absolute number. } deriving (Eq, Show) @@ -108,6 +138,9 @@ newRxFlow win = RxFlow win 0 0 win -- | 'rxfLimit' - 'rxfReceived'. +-- +-- This is the number of bytes the peer is still allowed to send before they +-- must wait for a window update; see 'RxFlow' for details. rxWindowSize :: RxFlow -> WindowSize rxWindowSize RxFlow{..} = rxfLimit - rxfReceived @@ -118,43 +151,9 @@ | -- | QUIC style FCTMaxData --- | When an application consumed received data, this function should --- be called to update 'rxfConsumed'. If the available buffer size --- is less than the half of the total buffer size AND window size update --- is greater than 1/8 of the the total buffer size, --- the representation of the window size update is returned. --- --- @ --- Example: --- --- rxfBufSize --- |------------------------| --- --------------------------------------> --- ^ ^ ^ --- rxfConsumed rxfReceived rxfLimit --- |01234567890| --- --- In the case where the window update should be informed to the peer, --- 'rxfConsumed' and 'rxfLimit' move to the right. The difference --- of old and new 'rxfLimit' is window update. --- --- rxfBufSize --- |------------------------| --- --------------------------------------> --- ^ ^ ^ --- rxfConsumed rxfReceived rxfLimit --- |0123456789012| : window glows --- --- Otherwise, only 'rxfConsumed' moves to the right. --- --- rxfBufSize --- |------------------------| --- --------------------------------------> --- ^ ^ ^ --- rxfConsumed rxfReceived rxfLimit --- |01234567890| : window stays +-- | Record that we have consumed some received data -- --- @ +-- May return a window update; see 'RxFlow' for details. maybeOpenRxWindow :: Int -- ^ The consumed size. @@ -163,10 +162,10 @@ -> (RxFlow, Maybe Int) -- ^ 'Just' if the size should be informed to the peer. maybeOpenRxWindow consumed fct flow@RxFlow{..} - | available < threshold && winUpdate > minSize = + | winUpdate >= threshold = let flow' = flow - { rxfConsumed = consumed' + { rxfConsumed = rxfConsumed' , rxfLimit = rxfLimit' } update = case fct of @@ -174,14 +173,16 @@ FCTMaxData -> rxfLimit' in (flow', Just update) | otherwise = - let flow' = flow{rxfConsumed = consumed'} + let flow' = flow{rxfConsumed = rxfConsumed'} in (flow', Nothing) where - available = rxfLimit - rxfReceived + rxfConsumed' = rxfConsumed + consumed + + -- Minimum window update size threshold = rxfBufSize `unsafeShiftR` 1 - minSize = rxfBufSize `unsafeShiftR` 3 - consumed' = rxfConsumed + consumed - rxfLimit' = consumed' + rxfBufSize + + -- The window update, /if/ we choose to send it + rxfLimit' = rxfConsumed' + rxfBufSize winUpdate = rxfLimit' - rxfLimit -- | Checking if received data is acceptable against the diff --git a/network-control.cabal b/network-control.cabal --- a/network-control.cabal +++ b/network-control.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: network-control -version: 0.1.2 +version: 0.1.3 license: BSD-3-Clause license-file: LICENSE maintainer: kazu@iij.ad.jp diff --git a/test/Network/Control/FlowSpec.hs b/test/Network/Control/FlowSpec.hs --- a/test/Network/Control/FlowSpec.hs +++ b/test/Network/Control/FlowSpec.hs @@ -37,11 +37,14 @@ maxWindowSize :: Int maxWindowSize = 200 -- (more realistic: 2_000_000) -minFrameSize :: Int -minFrameSize = -20 - instance Arbitrary RxFlow where - arbitrary = newRxFlow <$> chooseInt (1, maxWindowSize) + -- Prefer to generate a simple window size + arbitrary = + newRxFlow + <$> oneof + [ elements [1, 10, 50, 100] + , chooseInt (1, maxWindowSize) + ] instance Arbitrary Op where arbitrary = elements [minBound ..] @@ -54,23 +57,23 @@ where runManySteps :: Int -> Int -> RxFlow -> Gen [(Int, Step OpWithResult, RxFlow)] runManySteps 0 _ _ = pure [] - runManySteps len ix oldFlow | len > 0 = do + runManySteps len ix oldFlow = do (newStep, newFlow) <- runStep oldFlow <$> genStep oldFlow ((ix, newStep, newFlow) :) <$> runManySteps (len - 1) (ix + 1) newFlow - -- Not sure frame size > window size or 0 or engative consumed or received bytes are - -- legal, but RxFlow works fine with them. :) genStep :: RxFlow -> Gen (Step Op) genStep oldFlow = oneof [mkConsume, mkReceive] where + -- Negative frames are non-sensical; frames larger than the window + -- size are theoretically possible (but will trivially be rejected + -- as exceeding the window). mkReceive = - Step Receive <$> chooseInt (minFrameSize, rxfBufSize oldFlow * 2) + Step Receive <$> chooseInt (0, rxfBufSize oldFlow * 2) + -- We can only consume as much as we have received + -- (but it is in principle not a problem to consume 0 bytes) mkConsume = - let recv = rxfReceived oldFlow - in if recv > 0 - then Step Consume <$> chooseInt (minFrameSize, rxfReceived oldFlow) - else mkReceive + Step Consume <$> chooseInt (0, rxfReceived oldFlow - rxfConsumed oldFlow) runStep :: RxFlow -> Step Op -> (Step OpWithResult, RxFlow) runStep oldFlow = \case @@ -81,14 +84,17 @@ let (newFlow, isAcceptable) = checkRxLimit arg oldFlow in (Step (ReceiveWithResult isAcceptable) arg, newFlow) - shrink trace@(Trace initialFlow steps) = - trunc trace <> (Trace initialFlow <$> init (inits steps)) + shrink (Trace initialFlow steps) = + concat + [ -- Take a prefix (starting with the same initialFlow) + Trace initialFlow <$> init (inits steps) + , -- Take a suffix (starting with a later initialFlow) + map shiftInitialFlow $ tail (tails steps) + ] where - trunc :: Trace -> [Trace] - trunc (Trace _ stp) = case reverse stp of - [] -> [] - [_] -> [] - ((ix, lastStep, lastFlow) : (_, _, initFlow) : _) -> [Trace initFlow [(ix, lastStep, lastFlow)]] + shiftInitialFlow :: [(Int, Step OpWithResult, RxFlow)] -> Trace + shiftInitialFlow [] = Trace initialFlow [] + shiftInitialFlow ((_, _, initialFlow') : rest) = Trace initialFlow' rest -- invariants @@ -103,20 +109,42 @@ check :: Expectation check = case step of Step (ConsumeWithResult limitDelta) arg -> do + -- There is no point duplicating precisely the same logic here as in + -- 'maybeOpenRxWindow': that would result in circular reasoning. + -- Instead, we leave 'maybeOpenRxWindow' some implementation + -- freedom, and only verify that the window update makes sense: + -- + -- (a) It can't be too large: the new window after the update should + -- never exceed the specified buffer size. + -- (b) It can't be too late: if we consume /all/ received data, and + -- do not allow the peer to send any further data, then the + -- system deadlocks. + -- (c) It shouldn't be too small: very small window updates are + -- wasteful. + -- + -- Within these parameters 'maybeOpenRxWindow' can decide when to + -- send window updates and how large they should be. We also don't + -- set the bound on (c) too strict. newFlow `shouldBe` RxFlow - { rxfBufSize = rxfBufSize newFlow + { rxfBufSize = rxfBufSize oldFlow , rxfConsumed = rxfConsumed oldFlow + arg , rxfReceived = rxfReceived oldFlow - , rxfLimit = - if rxfLimit oldFlow - rxfReceived oldFlow < rxfBufSize oldFlow `div` 2 - then rxfConsumed oldFlow + arg + rxfBufSize oldFlow - else rxfLimit oldFlow + , rxfLimit = case limitDelta of + Nothing -> rxfLimit oldFlow + Just upd -> rxfLimit oldFlow + upd } - limitDelta - `shouldBe` case rxfLimit newFlow - rxfLimit oldFlow of - 0 -> Nothing - n -> Just n + -- Condition (a) + newFlow `shouldSatisfy` \flow -> + rxfLimit flow - rxfConsumed flow <= rxfBufSize flow + -- Condition (b) + newFlow `shouldSatisfy` \flow -> + rxfLimit flow > rxfConsumed flow + -- Condition (c) + limitDelta `shouldSatisfy` \mUpd -> + case mUpd of + Nothing -> True + Just upd -> upd >= rxfBufSize newFlow `div` 8 Step (ReceiveWithResult isAcceptable) arg -> do newFlow `shouldBe` if isAcceptable