Skip to content

Commit bf03196

Browse files
karknukarknu
committed
Space out requests for the same tx.
Reduce resource consumption by waiting at least 200ms before issueing the next request for the same tx.
1 parent 7e182e6 commit bf03196

8 files changed

Lines changed: 254 additions & 122 deletions

File tree

‎cardano-diffusion/tests/lib/Test/Cardano/Network/Diffusion/Testnet.hs‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1090,10 +1090,14 @@ prop_check_inflight_ratio bi ds@(DiffusionScript simArgs _ _) =
10901090
txDecisionPolicy = saTxDecisionPolicy simArgs
10911091

10921092
in tabulate "Max observeed ratio of inflight multiplicity by the max stipulated by the policy"
1093-
(map (\m -> "has " ++ show m ++ " in flight - ratio: "
1094-
++ show @(Ratio Int) (fromIntegral m / fromIntegral (txInflightMultiplicity txDecisionPolicy))
1095-
)
1096-
(Map.elems inflightTxsMap))
1093+
(let maxAllowed = txInflightMultiplicity txDecisionPolicy
1094+
inflightCounts = map inFlightCount (Map.elems inflightTxsMap)
1095+
in map (\m -> "has " ++ show m ++ " in flight - ratio: "
1096+
++ if maxAllowed > 0
1097+
then show @(Ratio Int) (m % maxAllowed)
1098+
else "n/a"
1099+
)
1100+
inflightCounts)
10971101
True
10981102

10991103
-- | This test coverage of InboundGovernor transitions.

‎ouroboros-network/bench/Main.hs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,8 @@ main =
4646
)
4747
(\as ->
4848
bench "makeDecisions: 100 x 10"
49-
$ let run (policy, state) =
50-
Tx.makeDecisions policy state (peerTxStates state)
49+
$ let run (now, policy, state) =
50+
Tx.makeDecisions now policy state (peerTxStates state)
5151
in nf (map run) as
5252
)
5353
, env (do let a = TX.mkDecisionContexts 131 100 100
@@ -62,8 +62,8 @@ main =
6262
)
6363
(\as ->
6464
bench "makeDecisions: 100 x 100"
65-
$ let run (policy, state) =
66-
Tx.makeDecisions policy state (peerTxStates state)
65+
$ let run (now, policy, state) =
66+
Tx.makeDecisions now policy state (peerTxStates state)
6767
in nf (map run) as
6868
)
6969
, env (do let a = TX.mkDecisionContexts 361 100 1_000
@@ -78,8 +78,8 @@ main =
7878
)
7979
(\as ->
8080
bench "makeDecisions: 100 x 1000"
81-
$ let run (policy, state) =
82-
Tx.makeDecisions policy state (peerTxStates state)
81+
$ let run (now, policy, state) =
82+
Tx.makeDecisions now policy state (peerTxStates state)
8383
in nf (map run) as
8484
)
8585
{-
@@ -91,8 +91,8 @@ main =
9191
)
9292
(\a ->
9393
bench "makeDecisions: random"
94-
$ let run (policy, state) =
95-
Tx.makeDecisions policy state (peerTxStates state)
94+
$ let run (now, policy, state) =
95+
Tx.makeDecisions now policy state (peerTxStates state)
9696
in nf (map run) a
9797
)
9898
-}

‎ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Decision.hs‎

Lines changed: 31 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Decision
1818

1919
import Control.Arrow ((>>>))
2020
import Control.Exception (assert)
21+
import Control.Monad.Class.MonadTime.SI (addTime, Time)
2122

2223
import Data.Bifunctor (second)
2324
import Data.Hashable
@@ -46,7 +47,9 @@ makeDecisions
4647
, Ord txid
4748
, Hashable peeraddr
4849
)
49-
=> TxDecisionPolicy
50+
=> Time
51+
-- ^ current time
52+
-> TxDecisionPolicy
5053
-- ^ decision policy
5154
-> SharedTxState peeraddr txid tx
5255
-- ^ decision context
@@ -60,11 +63,11 @@ makeDecisions
6063
-> ( SharedTxState peeraddr txid tx
6164
, Map peeraddr (TxDecision txid tx)
6265
)
63-
makeDecisions policy st =
66+
makeDecisions now policy st =
6467
let (salt, rng') = random (peerRng st)
6568
st' = st { peerRng = rng' }
6669
in fn
67-
. pickTxsToDownload policy st'
70+
. pickTxsToDownload now policy st'
6871
. orderByRejections salt
6972
where
7073
fn :: forall a.
@@ -93,7 +96,7 @@ orderByRejections salt =
9396
-- | Internal state of `pickTxsToDownload` computation.
9497
--
9598
data St peeraddr txid tx =
96-
St { stInflight :: !(Map txid Int),
99+
St { stInflight :: !(Map txid InFlightState),
97100
-- ^ `txid`s in-flight.
98101

99102
stAcknowledged :: !(Map txid Int),
@@ -123,7 +126,9 @@ pickTxsToDownload
123126
( Ord peeraddr
124127
, Ord txid
125128
)
126-
=> TxDecisionPolicy
129+
=> Time
130+
-- ^ current time
131+
-> TxDecisionPolicy
127132
-- ^ decision policy
128133
-> SharedTxState peeraddr txid tx
129134
-- ^ shared state
@@ -133,8 +138,9 @@ pickTxsToDownload
133138
, [(peeraddr, TxDecision txid tx)]
134139
)
135140

136-
pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer,
137-
txInflightMultiplicity }
141+
pickTxsToDownload now policy@TxDecisionPolicy { txsSizeInflightPerPeer,
142+
txInflightMultiplicity,
143+
interTxSpace }
138144
sharedState@SharedTxState { peerTxStates,
139145
inflightTxs,
140146
bufferedTxs,
@@ -180,7 +186,8 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer,
180186
-- does not allow to short circuit the fold, unlike
181187
-- `foldWithState`.
182188
foldWithState
183-
(\(txid, (txSize, inflightMultiplicity)) sizeInflight ->
189+
(\(txid, (txSize, inflightSt)) sizeInflight ->
190+
let inflightMultiplicity = inFlightCount inflightSt in
184191
if -- note that we pick `txid`'s as long the `s` is
185192
-- smaller or equal to `txsSizeInflightPerPeer`.
186193
sizeInflight <= txsSizeInflightPerPeer
@@ -196,7 +203,7 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer,
196203
-- merge `availableTxIds` with `stInflight`, so we don't
197204
-- need to lookup into `stInflight` on every `txid` which
198205
-- is in `availableTxIds`.
199-
Map.merge (Map.mapMaybeMissing \_txid -> Just . (,0))
206+
Map.merge (Map.mapMaybeMissing \_txid -> Just . (, mempty))
200207
Map.dropMissing
201208
(Map.zipWithMatched \_txid -> (,))
202209

@@ -233,13 +240,14 @@ pickTxsToDownload policy@TxDecisionPolicy { txsSizeInflightPerPeer,
233240

234241
stAcknowledged' = Map.unionWith (+) stAcknowledged txIdsToAck
235242

236-
stInflightDelta :: Map txid Int
237-
stInflightDelta = Map.fromSet (\_ -> 1) txsToRequest
243+
stInflightDelta :: Map txid InFlightState
244+
stInflightDelta = Map.fromSet (\_ -> InFlightState 1 $ addTime interTxSpace now)
245+
txsToRequest
238246
-- note: this is right since every `txid`
239247
-- could be picked at most once
240248

241-
stInflight' :: Map txid Int
242-
stInflight' = Map.unionWith (+) stInflightDelta stInflight
249+
stInflight' :: Map txid InFlightState
250+
stInflight' = Map.unionWith (<>) stInflightDelta stInflight
243251

244252
stInSubmissionToMempoolTxs' = stInSubmissionToMempoolTxs
245253
<> Set.fromList (map fst listOfTxsToMempool)
@@ -346,10 +354,12 @@ filterActivePeers
346354
:: forall peeraddr txid tx.
347355
Ord txid
348356
=> HasCallStack
349-
=> TxDecisionPolicy
357+
=> Time
358+
-> TxDecisionPolicy
350359
-> SharedTxState peeraddr txid tx
351360
-> Map peeraddr (PeerTxState txid tx)
352361
filterActivePeers
362+
now
353363
policy@TxDecisionPolicy {
354364
maxUnacknowledgedTxIds,
355365
txsSizeInflightPerPeer,
@@ -362,7 +372,13 @@ filterActivePeers
362372
inSubmissionToMempoolTxs
363373
} = Map.filter gn peerTxStates
364374
where
365-
unrequestable = Map.keysSet (Map.filter (>= txInflightMultiplicity) inflightTxs)
375+
376+
unrequestableFilter :: InFlightState -> Bool
377+
unrequestableFilter InFlightState{inFlightCount, inFlightNextReq} =
378+
inFlightCount >= txInflightMultiplicity || inFlightNextReq > now
379+
380+
unrequestable :: Set txid
381+
unrequestable = Map.keysSet (Map.filter unrequestableFilter inflightTxs)
366382
<> Map.keysSet bufferedTxs
367383

368384
gn :: PeerTxState txid tx -> Bool

‎ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Policy.hs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,9 +59,12 @@ data TxDecisionPolicy = TxDecisionPolicy {
5959
scoreRate :: !Double,
6060
-- ^ rate at which "rejected" TXs drain. Unit: TX/seconds.
6161

62-
scoreMax :: !Double
62+
scoreMax :: !Double,
6363
-- ^ Maximum number of "rejections". Unit: seconds
6464

65+
interTxSpace :: !DiffTime
66+
-- ^ space between requests for the same TX.
67+
6568
}
6669
deriving Show
6770

@@ -78,5 +81,6 @@ defaultTxDecisionPolicy =
7881
txInflightMultiplicity = 2,
7982
bufferedTxsMinLifetime = 2,
8083
scoreRate = 0.1,
81-
scoreMax = 15 * 60
84+
scoreMax = 15 * 60,
85+
interTxSpace = 0.2
8286
}

‎ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs‎

Lines changed: 70 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Registry
1818
) where
1919

2020
import Control.Concurrent.Class.MonadMVar.Strict
21+
import Control.Concurrent.Class.MonadSTM qualified as Lazy
2122
import Control.Concurrent.Class.MonadSTM.Strict
2223
import Control.Concurrent.Class.MonadSTM.TSem
2324
import Control.Monad.Class.MonadAsync
@@ -240,8 +241,8 @@ withPeer tracer
240241

241242
purgeInflightTxs m txid = Map.alter fn txid m
242243
where
243-
fn (Just n) | n > 1 = Just $! pred n
244-
fn _ = Nothing
244+
fn (Just a ) | inFlightCount a > 1 = Just $! a { inFlightCount = inFlightCount a - 1 }
245+
fn _ = Nothing
245246

246247
--
247248
-- PeerTxAPI
@@ -435,28 +436,32 @@ drainRejectionThread tracer policy sharedStateVar = do
435436
threadDelay 1
436437

437438
!now <- getMonotonicTime
438-
st'' <- atomically $ do
439+
st''' <- atomically $ do
439440
st <- readTVar sharedStateVar
440441
let ptss = if now > nextDrain then Map.map (updateRejects policy now 0) (peerTxStates st)
441442
else peerTxStates st
442443
st' = tickTimedTxs now st
443444
{ peerTxStates = ptss }
444-
writeTVar sharedStateVar st'
445-
return st'
446-
traceWith tracer (TraceSharedTxState "drainRejectionThread" st'')
445+
st'' = st' { inflightTxs = Map.filter (filterStaleReq now) (inflightTxs st')}
446+
writeTVar sharedStateVar st''
447+
return st''
448+
traceWith tracer (TraceSharedTxState "drainRejectionThread" st''')
447449

448450
if now > nextDrain
449451
then go $ addTime drainInterval now
450452
else go nextDrain
451453

454+
filterStaleReq :: Time -> InFlightState -> Bool
455+
filterStaleReq now e = inFlightCount e > 0 || inFlightNextReq e > now
456+
452457

453458
decisionLogicThread
454459
:: forall m peeraddr txid tx.
455460
( MonadDelay m
456461
, MonadMVar m
457-
, MonadSTM m
458462
, MonadMask m
459463
, MonadFork m
464+
, MonadTimer m
460465
, Ord peeraddr
461466
, Ord txid
462467
, Hashable peeraddr
@@ -477,26 +482,64 @@ decisionLogicThread tracer counterTracer policy txChannelsVar sharedStateVar = d
477482
-- if there are too many inbound connections.
478483
threadDelay _DECISION_LOOP_DELAY
479484

480-
(decisions, st) <- atomically do
485+
now <- getMonotonicTime
486+
nextDelay <- atomically $ do
487+
sharedTxState <- readTVar sharedStateVar
488+
return $ nextDecisionDelay now sharedTxState
489+
delayVar <- registerDelay nextDelay
490+
res_m <- atomically do
481491
sharedTxState <- readTVar sharedStateVar
482-
let activePeers = filterActivePeers policy sharedTxState
483-
484-
-- block until at least one peer is active
485-
check (not (Map.null activePeers))
486-
487-
let (sharedState, decisions) = makeDecisions policy sharedTxState activePeers
488-
writeTVar sharedStateVar sharedState
489-
return (decisions, sharedState)
490-
traceWith tracer (TraceSharedTxState "decisionLogicThread" st)
491-
traceWith tracer (TraceTxDecisions decisions)
492-
TxChannels { txChannelMap } <- readMVar txChannelsVar
493-
traverse_
494-
(\(mvar, d) -> modifyMVarWithDefault_ mvar d (\d' -> pure (d' <> d)))
495-
(Map.intersectionWith (,)
496-
txChannelMap
497-
decisions)
498-
traceWith counterTracer (mkTxSubmissionCounters st)
499-
go
492+
let activePeers = filterActivePeers now policy sharedTxState
493+
timerExpired <- Lazy.readTVar delayVar
494+
495+
-- block until at least one peer is active or the timer expires
496+
if not (Map.null activePeers)
497+
then do
498+
let (sharedState, decisions) = makeDecisions now policy sharedTxState activePeers
499+
writeTVar sharedStateVar sharedState
500+
return $ Just (decisions, sharedState)
501+
else if timerExpired
502+
then return Nothing
503+
else retry
504+
505+
case res_m of
506+
Nothing -> go
507+
Just (decisions, st) -> do
508+
traceWith tracer (TraceSharedTxState "decisionLogicThread" st)
509+
traceWith tracer (TraceTxDecisions decisions)
510+
TxChannels { txChannelMap } <- readMVar txChannelsVar
511+
traverse_
512+
(\(mvar, d) -> modifyMVarWithDefault_ mvar d (\d' -> pure (d' <> d)))
513+
(Map.intersectionWith (,)
514+
txChannelMap
515+
decisions)
516+
traceWith counterTracer (mkTxSubmissionCounters st)
517+
go
518+
519+
nextDecisionDelay
520+
:: Time
521+
-> SharedTxState peeraddr txid tx
522+
-> DiffTime
523+
nextDecisionDelay now SharedTxState { inflightTxs } =
524+
fromMaybe maxDelay (diffTimeNow <$> nextWake)
525+
where
526+
-- If there are no outstanding TXs we wait for a long time
527+
-- or until an STM value changes.
528+
maxDelay :: DiffTime
529+
maxDelay = 120
530+
531+
nextWake :: Maybe Time
532+
nextWake =
533+
Foldable.foldl' step Nothing inflightTxs
534+
535+
step :: Maybe Time -> InFlightState -> Maybe Time
536+
step acc InFlightState { inFlightNextReq } =
537+
if inFlightNextReq <= now
538+
then acc
539+
else Just $ maybe inFlightNextReq (min inFlightNextReq) acc
540+
541+
diffTimeNow :: Time -> DiffTime
542+
diffTimeNow t = t `diffTime` now
500543

501544
-- Variant of modifyMVar_ that puts a default value if the MVar is empty.
502545
modifyMVarWithDefault_ :: StrictMVar m a -> a -> (a -> m a) -> m ()
@@ -519,6 +562,7 @@ decisionLogicThreads
519562
, MonadMask m
520563
, MonadAsync m
521564
, MonadFork m
565+
, MonadTimer m
522566
, Ord peeraddr
523567
, Ord txid
524568
, Hashable peeraddr

‎ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -467,11 +467,10 @@ collectTxsImpl txSize peeraddr requestedTxIdsMap receivedTxs
467467
Map.merge
468468
(Map.mapMaybeMissing \_ x -> Just x)
469469
(Map.mapMaybeMissing \_ _ -> assert False Nothing)
470-
(Map.zipWithMaybeMatched \_ x y -> assert (x >= y)
471-
let z = x - y in
472-
if z > 0
473-
then Just z
474-
else Nothing)
470+
(Map.zipWithMaybeMatched \_ x y ->
471+
assert (inFlightCount x >= y)
472+
let cnt' = inFlightCount x - y in
473+
Just $ x { inFlightCount = cnt' })
475474
(inflightTxs st)
476475
(Map.fromSet (const 1) requestedTxIds)
477476

0 commit comments

Comments
 (0)