Skip to content

Commit 3e6298f

Browse files
cootcoot
committed
chain-sync: randomised ProtocolTimeLimits
This patch adds a way to randomise `ProtocolTimeLimits` per message, rather than per connection. This is used in `ChainSync` to pick a random timeout in the `MustReply` state.
1 parent f322dbc commit 3e6298f

File tree

8 files changed

+249
-126
lines changed

8 files changed

+249
-126
lines changed

‎ouroboros-network-framework/src/Ouroboros/Network/Driver/Limits.hs

Lines changed: 194 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,19 @@ module Ouroboros.Network.Driver.Limits
2222
-- * Normal peers
2323
, runPeerWithLimits
2424
, runPipelinedPeerWithLimits
25+
, runPeerWithLimitsRnd
26+
, runPipelinedPeerWithLimitsRnd
2527
, TraceSendRecv (..)
2628
-- * Driver utilities
2729
, driverWithLimits
2830
, runConnectedPeersWithLimits
2931
, runConnectedPipelinedPeersWithLimits
32+
, runConnectedPeersWithLimitsRnd
33+
, runConnectedPipelinedPeersWithLimitsRnd
3034
) where
3135

3236
import Data.Maybe (fromMaybe)
37+
import System.Random
3338

3439
import Control.Monad.Class.MonadAsync
3540
import Control.Monad.Class.MonadFork
@@ -105,6 +110,65 @@ driverWithLimits tracer timeoutFn
105110
Nothing -> throwIO (ExceededTimeLimit tok)
106111

107112

113+
driverWithLimitsRnd :: forall ps (pr :: PeerRole) failure bytes m.
114+
( MonadThrow m
115+
, ShowProxy ps
116+
, forall (st' :: ps) tok. tok ~ StateToken st' => Show tok
117+
, Show failure
118+
)
119+
=> Tracer m (TraceSendRecv ps)
120+
-> TimeoutFn m
121+
-> StdGen
122+
-> Codec ps failure m bytes
123+
-> ProtocolSizeLimits ps bytes
124+
-> (StdGen -> ProtocolTimeLimits ps)
125+
-> Channel m bytes
126+
-> Driver ps pr (Maybe bytes, StdGen) m
127+
driverWithLimitsRnd tracer timeoutFn rnd0
128+
Codec{encode, decode}
129+
ProtocolSizeLimits{sizeLimitForState, dataSize}
130+
genProtocolTimeLimits
131+
channel@Channel{send} =
132+
Driver { sendMessage, recvMessage, initialDState = (Nothing, rnd0) }
133+
where
134+
sendMessage :: forall (st :: ps) (st' :: ps).
135+
StateTokenI st
136+
=> ActiveState st
137+
=> WeHaveAgencyProof pr st
138+
-> Message ps st st'
139+
-> m ()
140+
sendMessage !_ msg = do
141+
send (encode msg)
142+
traceWith tracer (TraceSendMsg (AnyMessage msg))
143+
144+
145+
recvMessage :: forall (st :: ps).
146+
StateTokenI st
147+
=> ActiveState st
148+
=> TheyHaveAgencyProof pr st
149+
-> (Maybe bytes, StdGen)
150+
-> m (SomeMessage st, (Maybe bytes, StdGen))
151+
recvMessage !_ (trailing, !rnd) = do
152+
let tok = stateToken
153+
decoder <- decode tok
154+
let sizeLimit = sizeLimitForState @st stateToken
155+
156+
let (rnd', rnd'') = split rnd
157+
ProtocolTimeLimits{timeLimitForState} = genProtocolTimeLimits rnd''
158+
timeLimit = fromMaybe (-1) $ timeLimitForState @st stateToken
159+
result <- timeoutFn timeLimit $
160+
runDecoderWithLimit sizeLimit dataSize
161+
channel trailing decoder
162+
163+
case result of
164+
Just (Right (x@(SomeMessage msg), trailing')) -> do
165+
traceWith tracer (TraceRecvMsg (AnyMessage msg))
166+
return (x, (trailing', rnd'))
167+
Just (Left (Just failure)) -> throwIO (DecoderFailure tok failure)
168+
Just (Left Nothing) -> throwIO (ExceededSizeLimit tok)
169+
Nothing -> throwIO (ExceededTimeLimit tok)
170+
171+
108172
runDecoderWithLimit
109173
:: forall m bytes failure a. Monad m
110174
=> Word
@@ -152,7 +216,8 @@ runDecoderWithLimit limit size Channel{recv} =
152216
Just bs -> do let sz' = sz + size bs
153217
go sz' Nothing =<< k (Just bs)
154218

155-
219+
-- | Run a peer with limits.
220+
--
156221
runPeerWithLimits
157222
:: forall ps (st :: ps) pr failure bytes m a .
158223
( MonadAsync m
@@ -175,6 +240,37 @@ runPeerWithLimits tracer codec slimits tlimits channel peer =
175240
withTimeoutSerial $ \timeoutFn ->
176241
let driver = driverWithLimits tracer timeoutFn codec slimits tlimits channel
177242
in runPeerWithDriver driver peer
243+
244+
245+
-- | Run a peer with limits. 'ProtocolTimeLimits' have access to
246+
-- a pseudorandom generator.
247+
--
248+
runPeerWithLimitsRnd
249+
:: forall ps (st :: ps) pr failure bytes m a .
250+
( MonadAsync m
251+
, MonadFork m
252+
, MonadMask m
253+
, MonadThrow (STM m)
254+
, MonadTimer m
255+
, ShowProxy ps
256+
, forall (st' :: ps) stok. stok ~ StateToken st' => Show stok
257+
, Show failure
258+
)
259+
=> Tracer m (TraceSendRecv ps)
260+
-> StdGen
261+
-> Codec ps failure m bytes
262+
-> ProtocolSizeLimits ps bytes
263+
-> (StdGen -> ProtocolTimeLimits ps)
264+
-> Channel m bytes
265+
-> Peer ps pr NonPipelined st m a
266+
-> m (a, Maybe bytes)
267+
runPeerWithLimitsRnd tracer rnd codec slimits tlimits channel peer =
268+
withTimeoutSerial $ \timeoutFn ->
269+
let driver = driverWithLimitsRnd tracer timeoutFn rnd codec slimits tlimits channel
270+
in (\(a, (trailing, _)) -> (a, trailing))
271+
<$> runPeerWithDriver driver peer
272+
273+
178274
-- | Run a pipelined peer with the given channel via the given codec.
179275
--
180276
-- This runs the peer to completion (if the protocol allows for termination).
@@ -206,6 +302,35 @@ runPipelinedPeerWithLimits tracer codec slimits tlimits channel peer =
206302
in runPipelinedPeerWithDriver driver peer
207303

208304

305+
-- | Like 'runPipelinedPeerWithLimits' but time limits have access to
306+
-- a pseudorandom generator.
307+
--
308+
runPipelinedPeerWithLimitsRnd
309+
:: forall ps (st :: ps) pr failure bytes m a.
310+
( MonadAsync m
311+
, MonadFork m
312+
, MonadMask m
313+
, MonadTimer m
314+
, MonadThrow (STM m)
315+
, ShowProxy ps
316+
, forall (st' :: ps) stok. stok ~ StateToken st' => Show stok
317+
, Show failure
318+
)
319+
=> Tracer m (TraceSendRecv ps)
320+
-> StdGen
321+
-> Codec ps failure m bytes
322+
-> ProtocolSizeLimits ps bytes
323+
-> (StdGen -> ProtocolTimeLimits ps)
324+
-> Channel m bytes
325+
-> PeerPipelined ps pr st m a
326+
-> m (a, Maybe bytes)
327+
runPipelinedPeerWithLimitsRnd tracer rnd codec slimits tlimits channel peer =
328+
withTimeoutSerial $ \timeoutFn ->
329+
let driver = driverWithLimitsRnd tracer timeoutFn rnd codec slimits tlimits channel
330+
in (\(a, (trailing, _)) -> (a, trailing))
331+
<$> runPipelinedPeerWithDriver driver peer
332+
333+
209334
-- | Run two 'Peer's via a pair of connected 'Channel's and a common 'Codec'.
210335
-- The client side is using 'driverWithLimits'.
211336
--
@@ -248,6 +373,41 @@ runConnectedPeersWithLimits createChannels tracer codec slimits tlimits client s
248373
tracerServer = contramap ((,) Server) tracer
249374

250375

376+
runConnectedPeersWithLimitsRnd
377+
:: forall ps pr st failure bytes m a b.
378+
( MonadAsync m
379+
, MonadFork m
380+
, MonadMask m
381+
, MonadTimer m
382+
, MonadThrow (STM m)
383+
, Exception failure
384+
, ShowProxy ps
385+
, forall (st' :: ps) sing. sing ~ StateToken st' => Show sing
386+
)
387+
=> m (Channel m bytes, Channel m bytes)
388+
-> Tracer m (Role, TraceSendRecv ps)
389+
-> StdGen
390+
-> Codec ps failure m bytes
391+
-> ProtocolSizeLimits ps bytes
392+
-> (StdGen -> ProtocolTimeLimits ps)
393+
-> Peer ps pr NonPipelined st m a
394+
-> Peer ps (FlipAgency pr) NonPipelined st m b
395+
-> m (a, b)
396+
runConnectedPeersWithLimitsRnd createChannels tracer rnd codec slimits tlimits client server =
397+
createChannels >>= \(clientChannel, serverChannel) ->
398+
399+
(do labelThisThread "client"
400+
fst <$> runPeerWithLimitsRnd
401+
tracerClient rnd codec slimits tlimits
402+
clientChannel client)
403+
`concurrently`
404+
(do labelThisThread "server"
405+
fst <$> runPeer tracerServer codec serverChannel server)
406+
where
407+
tracerClient = contramap ((,) Client) tracer
408+
tracerServer = contramap ((,) Server) tracer
409+
410+
251411
-- | Run two 'Peer's via a pair of connected 'Channel's and a common 'Codec'.
252412
-- The client side is using 'driverWithLimits'.
253413
--
@@ -286,3 +446,36 @@ runConnectedPipelinedPeersWithLimits createChannels tracer codec slimits tlimits
286446
where
287447
tracerClient = contramap ((,) Client) tracer
288448
tracerServer = contramap ((,) Server) tracer
449+
450+
451+
runConnectedPipelinedPeersWithLimitsRnd
452+
:: forall ps pr st failure bytes m a b.
453+
( MonadAsync m
454+
, MonadFork m
455+
, MonadMask m
456+
, MonadTimer m
457+
, MonadThrow (STM m)
458+
, Exception failure
459+
, ShowProxy ps
460+
, forall (st' :: ps) sing. sing ~ StateToken st' => Show sing
461+
)
462+
=> m (Channel m bytes, Channel m bytes)
463+
-> Tracer m (Role, TraceSendRecv ps)
464+
-> StdGen
465+
-> Codec ps failure m bytes
466+
-> ProtocolSizeLimits ps bytes
467+
-> (StdGen -> ProtocolTimeLimits ps)
468+
-> PeerPipelined ps pr st m a
469+
-> Peer ps (FlipAgency pr) NonPipelined st m b
470+
-> m (a, b)
471+
runConnectedPipelinedPeersWithLimitsRnd createChannels tracer rnd codec slimits tlimits client server =
472+
createChannels >>= \(clientChannel, serverChannel) ->
473+
474+
(fst <$> runPipelinedPeerWithLimitsRnd
475+
tracerClient rnd codec slimits tlimits
476+
clientChannel client)
477+
`concurrently`
478+
(fst <$> runPeer tracerServer codec serverChannel server)
479+
where
480+
tracerClient = contramap ((,) Client) tracer
481+
tracerServer = contramap ((,) Server) tracer

‎ouroboros-network-protocols/ouroboros-network-protocols.cabal

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ library
105105
nothunks,
106106
ouroboros-network-api ^>=0.9.0,
107107
quiet,
108+
random,
108109
serialise,
109110
si-timers,
110111
singletons,

‎ouroboros-network-protocols/src/Ouroboros/Network/Protocol/ChainSync/Codec.hs

Lines changed: 39 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
{-# LANGUAGE DataKinds #-}
22
{-# LANGUAGE FlexibleContexts #-}
33
{-# LANGUAGE GADTs #-}
4-
{-# LANGUAGE NamedFieldPuns #-}
54
{-# LANGUAGE PolyKinds #-}
65
{-# LANGUAGE RankNTypes #-}
76
{-# LANGUAGE ScopedTypeVariables #-}
@@ -11,19 +10,21 @@ module Ouroboros.Network.Protocol.ChainSync.Codec
1110
, codecChainSyncId
1211
, byteLimitsChainSync
1312
, timeLimitsChainSync
14-
, ChainSyncTimeout (..)
13+
, maxChainSyncTimeout
14+
, minChainSyncTimeout
1515
) where
1616

1717
import Control.Monad.Class.MonadST
1818
import Control.Monad.Class.MonadTime.SI
1919

20-
import Network.TypedProtocol.Codec.CBOR
20+
import Network.TypedProtocol.Codec.CBOR hiding (encode, decode)
2121

2222
import Ouroboros.Network.Protocol.ChainSync.Type
2323
import Ouroboros.Network.Protocol.Limits
2424

2525
import Data.ByteString.Lazy qualified as LBS
2626
import Data.Singletons (withSingI)
27+
import System.Random (StdGen, randomR)
2728

2829
import Codec.CBOR.Decoding (decodeListLen, decodeWord)
2930
import Codec.CBOR.Decoding qualified as CBOR
@@ -47,20 +48,18 @@ byteLimitsChainSync = ProtocolSizeLimits stateToLimit
4748
stateToLimit SingIntersect = smallByteLimit
4849
stateToLimit a@SingDone = notActiveState a
4950

50-
-- | Configurable timeouts
51-
--
52-
-- These are configurable for at least the following reasons.
51+
52+
-- | Chain sync `mustReplayTimeout` lower bound.
5353
--
54-
-- o So that deployment and testing can use different values.
54+
minChainSyncTimeout :: DiffTime
55+
minChainSyncTimeout = 135
56+
57+
58+
-- | Chain sync `mustReplayTimeout` upper bound.
5559
--
56-
-- o So that a net running Praos can better cope with streaks of empty slots.
57-
-- (See @intersectmbo/ouroboros-network#2245@.)
58-
data ChainSyncTimeout = ChainSyncTimeout
59-
{ canAwaitTimeout :: Maybe DiffTime
60-
, intersectTimeout :: Maybe DiffTime
61-
, mustReplyTimeout :: Maybe DiffTime
62-
, idleTimeout :: Maybe DiffTime
63-
}
60+
maxChainSyncTimeout :: DiffTime
61+
maxChainSyncTimeout = 269
62+
6463

6564
-- | Time Limits
6665
--
@@ -69,26 +68,35 @@ data ChainSyncTimeout = ChainSyncTimeout
6968
-- > 'TokNext TokMustReply' the given 'mustReplyTimeout'
7069
-- > 'TokIntersect' the given 'intersectTimeout'
7170
timeLimitsChainSync :: forall header point tip.
72-
ChainSyncTimeout
71+
StdGen
7372
-> ProtocolTimeLimits (ChainSync header point tip)
74-
timeLimitsChainSync csTimeouts = ProtocolTimeLimits stateToLimit
73+
timeLimitsChainSync rnd = ProtocolTimeLimits stateToLimit
7574
where
76-
ChainSyncTimeout
77-
{ canAwaitTimeout
78-
, intersectTimeout
79-
, mustReplyTimeout
80-
, idleTimeout
81-
} = csTimeouts
82-
8375
stateToLimit :: forall (st :: ChainSync header point tip).
8476
ActiveState st => StateToken st -> Maybe DiffTime
85-
stateToLimit SingIdle = idleTimeout
86-
stateToLimit (SingNext SingCanAwait) = canAwaitTimeout
87-
stateToLimit (SingNext SingMustReply) = mustReplyTimeout
88-
stateToLimit SingIntersect = intersectTimeout
89-
stateToLimit a@SingDone = notActiveState a
90-
91-
-- | Codec for chain sync that encodes/decodes headers
77+
stateToLimit SingIdle = Just 3673
78+
stateToLimit SingIntersect = shortWait
79+
stateToLimit (SingNext SingCanAwait) = shortWait
80+
stateToLimit (SingNext SingMustReply) =
81+
-- We draw from a range for which streaks of empty slots ranges
82+
-- from 0.0001% up to 1% probability.
83+
-- t = T_s [log (1-Y) / log (1-f)]
84+
-- Y = [0.99, 0.999...]
85+
-- T_s = slot length of 1s.
86+
-- f = 0.05
87+
-- The timeout is randomly picked per state to avoid all peers go down at
88+
-- the same time in case of a long streak of empty slots, and thus to
89+
-- avoid global synchronisation. The timeout is picked uniformly from
90+
-- the interval 135 - 269, which corresponds to 99.9% to
91+
-- 99.9999% thresholds.
92+
let timeout :: DiffTime
93+
timeout = realToFrac . fst
94+
. randomR ( realToFrac minChainSyncTimeout :: Double
95+
, realToFrac maxChainSyncTimeout :: Double
96+
)
97+
$ rnd
98+
in Just timeout
99+
stateToLimit a@SingDone = notActiveState a
92100
--
93101
-- NOTE: See 'wrapCBORinCBOR' and 'unwrapCBORinCBOR' if you want to use this
94102
-- with a header type that has annotations.

0 commit comments

Comments
 (0)