network-control 0.1.2 → 0.1.3
raw patch · 4 files changed
+117/−83 lines, 4 files
Files
- Changelog.md +5/−0
- Network/Control/Flow.hs +55/−54
- network-control.cabal +1/−1
- test/Network/Control/FlowSpec.hs +56/−28
Changelog.md view
@@ -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
Network/Control/Flow.hs view
@@ -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
network-control.cabal view
@@ -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
test/Network/Control/FlowSpec.hs view
@@ -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