Full rework of the BlockFetch logic for bulk sync mode

ouroboros-network-api/src/Ouroboros/Network/BlockFetch/ConsensusInterface.hs

@@ -1,18 +1,24 @@
 {-# LANGUAGE DeriveFunctor #-}
 {-# LANGUAGE RankNTypes    #-}
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DeriveGeneric #-}
 
 module Ouroboros.Network.BlockFetch.ConsensusInterface
   ( FetchMode (..)
   , BlockFetchConsensusInterface (..)
   , WhetherReceivingTentativeBlocks (..)
   , FromConsensus (..)
+  , ChainSelStarvation (..)
   ) where
 
 import Control.Monad.Class.MonadSTM
 import Control.Monad.Class.MonadTime (UTCTime)
+import Control.Monad.Class.MonadTime.SI (Time)
 
 import Data.Map.Strict (Map)
 import GHC.Stack (HasCallStack)
+import GHC.Generics (Generic)
+import NoThunks.Class (NoThunks)
 
 import Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import Ouroboros.Network.Block
@@ -149,7 +155,17 @@ data BlockFetchConsensusInterface peer header block m =
        -- PRECONDITION: Same as 'headerForgeUTCTime'.
        --
        -- WARNING: Same as 'headerForgeUTCTime'.
-       blockForgeUTCTime  :: FromConsensus block -> STM m UTCTime
+       blockForgeUTCTime  :: FromConsensus block -> STM m UTCTime,
+
+       -- | Information on the ChainSel starvation status; whether it is ongoing
+       -- or has ended recently. Needed by the bulk sync decision logic.
+       readChainSelStarvation :: STM m ChainSelStarvation,
+
+       -- | Action to inform CSJ that the given peer has not been performing
+       -- adequately with respect to BlockFetch, and that it should be demoted
+       -- from the dynamo role. Can be set to @const (pure ())@ in all other
+       -- scenarios.
+       demoteCSJDynamo :: peer -> m ()
      }
 
 
@@ -159,6 +175,16 @@ data WhetherReceivingTentativeBlocks
   = ReceivingTentativeBlocks
   | NotReceivingTentativeBlocks
 
+-- | Whether ChainSel is starved or has been recently.
+--
+-- The bulk sync fetch decision logic needs to decide whether the current
+-- focused peer has starved ChainSel recently. This datatype is used to
+-- represent this piece of information.
+data ChainSelStarvation
+  = ChainSelStarvationOngoing
+  | ChainSelStarvationEndedAt Time
+  deriving (Eq, Show, NoThunks, Generic)
+
 {-------------------------------------------------------------------------------
   Syntactic indicator of key precondition about Consensus time conversions
 -------------------------------------------------------------------------------}

ouroboros-network/demo/chain-sync.hs

@@ -31,6 +31,7 @@ import Control.Concurrent.Async
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Exception
 import Control.Monad (when)
+import Control.Monad.Class.MonadTime.SI (Time (..))
 import Control.Tracer
 
 import System.Directory
@@ -75,6 +76,7 @@ import Ouroboros.Network.Protocol.BlockFetch.Type qualified as BlockFetch
 import Ouroboros.Network.BlockFetch
 import Ouroboros.Network.BlockFetch.Client
 import Ouroboros.Network.BlockFetch.ClientRegistry (FetchClientRegistry (..))
+import Ouroboros.Network.BlockFetch.ConsensusInterface (ChainSelStarvation(..))
 import Ouroboros.Network.DeltaQ (defaultGSV)
 
 
@@ -440,7 +442,10 @@ clientBlockFetch sockAddrs maxSlotNo = withIOManager $ \iocp -> do
               blockMatchesHeader     = \_ _ -> True,
 
               headerForgeUTCTime,
-              blockForgeUTCTime      = headerForgeUTCTime . fmap blockHeader
+              blockForgeUTCTime      = headerForgeUTCTime . fmap blockHeader,
+
+              readChainSelStarvation = pure (ChainSelStarvationEndedAt (Time 0)),
+              demoteCSJDynamo = \_ -> pure ()
             }
           where
             plausibleCandidateChain cur candidate =
@@ -506,11 +511,14 @@ clientBlockFetch sockAddrs maxSlotNo = withIOManager $ \iocp -> do
                       blockFetchPolicy
                       registry
                       (BlockFetchConfiguration {
-                        bfcMaxConcurrencyBulkSync = 1,
                         bfcMaxConcurrencyDeadline = 2,
                         bfcMaxRequestsInflight    = 10,
-                        bfcDecisionLoopInterval   = 0.01,
-                        bfcSalt                   = 0
+                        bfcDecisionLoopIntervalBulkSync = 0.04,
+                        bfcDecisionLoopIntervalDeadline = 0.01,
+                        bfcSalt                   = 0,
+                        bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+                          { gbfcBulkSyncGracePeriod = 10 -- seconds
+                          }
                         })
                  >> return ()
 

ouroboros-network/ouroboros-network.cabal

@@ -33,6 +33,9 @@ library
                        Ouroboros.Network.BlockFetch.ClientRegistry
                        Ouroboros.Network.BlockFetch.ClientState
                        Ouroboros.Network.BlockFetch.Decision
+                       Ouroboros.Network.BlockFetch.Decision.BulkSync
+                       Ouroboros.Network.BlockFetch.Decision.Deadline
+                       Ouroboros.Network.BlockFetch.Decision.Trace
                        Ouroboros.Network.BlockFetch.DeltaQ
                        Ouroboros.Network.BlockFetch.State
                        Ouroboros.Network.DeltaQ
@@ -113,6 +116,7 @@ library
                        cborg             >=0.2.1 && <0.3,
                        containers,
                        deepseq,
+                       dlist,
                        dns,
                        hashable,
                        iproute,
@@ -121,6 +125,7 @@ library
                        nothunks,
                        psqueues          >=0.2.3 && <0.3,
                        random,
+                       transformers,
 
                        cardano-prelude,
                        cardano-slotting,
@@ -337,6 +342,7 @@ executable demo-chain-sync
 
                        contra-tracer,
 
+                       si-timers,
                        typed-protocols,
                        strict-stm,
                        ouroboros-network-api,

ouroboros-network/sim-tests-lib/Ouroboros/Network/BlockFetch/Examples.hs

@@ -40,10 +40,12 @@ import Ouroboros.Network.Block
 import Network.TypedProtocol.Core
 import Network.TypedProtocol.Pipelined
 
+import qualified Ouroboros.Network.AnchoredFragment as AF
 import Ouroboros.Network.ControlMessage (ControlMessageSTM)
 
 import Ouroboros.Network.BlockFetch
 import Ouroboros.Network.BlockFetch.Client
+import Ouroboros.Network.BlockFetch.ConsensusInterface (ChainSelStarvation(..))
 import Ouroboros.Network.Channel
 import Ouroboros.Network.DeltaQ
 import Ouroboros.Network.Driver
@@ -55,6 +57,7 @@ import Ouroboros.Network.Protocol.BlockFetch.Type
 import Ouroboros.Network.Util.ShowProxy
 
 import Ouroboros.Network.Mock.ConcreteBlock
+import Ouroboros.Network.BlockFetch.Decision.Trace (TraceDecisionEvent)
 
 
 -- | Run a single block fetch protocol until the chain is downloaded.
@@ -63,8 +66,7 @@ blockFetchExample0 :: forall m.
                       (MonadSTM m, MonadST m, MonadAsync m, MonadDelay m,
                        MonadFork m, MonadTime m, MonadTimer m, MonadMask m,
                        MonadThrow (STM m))
-                   => Tracer m [TraceLabelPeer Int
-                                 (FetchDecision [Point BlockHeader])]
+                   => Tracer m (TraceDecisionEvent Int BlockHeader)
                    -> Tracer m (TraceLabelPeer Int
                                  (TraceFetchClientState BlockHeader))
                    -> Tracer m (TraceLabelPeer Int
@@ -134,11 +136,14 @@ blockFetchExample0 decisionTracer clientStateTracer clientMsgTracer
           (sampleBlockFetchPolicy1 headerForgeUTCTime blockHeap currentChainHeaders candidateChainHeaders)
           registry
           (BlockFetchConfiguration {
-            bfcMaxConcurrencyBulkSync = 1,
             bfcMaxConcurrencyDeadline = 2,
             bfcMaxRequestsInflight    = 10,
-            bfcDecisionLoopInterval   = 0.01,
-            bfcSalt                   = 0
+            bfcDecisionLoopIntervalBulkSync = 0.04,
+            bfcDecisionLoopIntervalDeadline = 0.01,
+            bfcSalt                   = 0,
+            bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+              { gbfcBulkSyncGracePeriod = 10 -- seconds
+              }
           })
         >> return ()
 
@@ -172,8 +177,7 @@ blockFetchExample1 :: forall m.
                       (MonadSTM m, MonadST m, MonadAsync m, MonadDelay m,
                        MonadFork m, MonadTime m, MonadTimer m, MonadMask m,
                        MonadThrow (STM m))
-                   => Tracer m [TraceLabelPeer Int
-                                 (FetchDecision [Point BlockHeader])]
+                   => Tracer m (TraceDecisionEvent Int BlockHeader)
                    -> Tracer m (TraceLabelPeer Int
                                  (TraceFetchClientState BlockHeader))
                    -> Tracer m (TraceLabelPeer Int
@@ -211,7 +215,7 @@ blockFetchExample1 decisionTracer clientStateTracer clientMsgTracer
     driverAsync <- async $ do
       threadId <- myThreadId
       labelThread threadId "block-fetch-driver"
-      driver blockHeap
+      downloadTimer
 
     -- Order of shutdown here is important for this example: must kill off the
     -- fetch thread before the peer threads.
@@ -243,25 +247,25 @@ blockFetchExample1 decisionTracer clientStateTracer clientMsgTracer
           (sampleBlockFetchPolicy1 headerForgeUTCTime blockHeap currentChainHeaders candidateChainHeaders)
           registry
           (BlockFetchConfiguration {
-            bfcMaxConcurrencyBulkSync = 1,
             bfcMaxConcurrencyDeadline = 2,
             bfcMaxRequestsInflight    = 10,
-            bfcDecisionLoopInterval   = 0.01,
-            bfcSalt                   = 0
+            bfcDecisionLoopIntervalBulkSync = 0.04,
+            bfcDecisionLoopIntervalDeadline = 0.01,
+            bfcSalt                   = 0,
+            bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+              { gbfcBulkSyncGracePeriod = 10 -- seconds
+              }
           })
         >> return ()
 
     headerForgeUTCTime (FromConsensus x) =
         pure $ convertSlotToTimeForTestsAssumingNoHardFork (blockSlot x)
 
-    driver :: TestFetchedBlockHeap m Block -> m ()
-    driver blockHeap = do
-      atomically $ do
-        heap <- getTestFetchedBlocks blockHeap
-        check $
-          all (\c -> AnchoredFragment.headPoint c `Set.member` heap)
-              candidateChains
-
+    -- | Terminates after 1 second per block in the candidate chains.
+    downloadTimer :: m ()
+    downloadTimer =
+      let totalBlocks = sum $ map AF.length candidateChains
+       in threadDelay (fromIntegral totalBlocks)
 
 --
 -- Sample block fetch configurations
@@ -293,7 +297,10 @@ sampleBlockFetchPolicy1 headerFieldsForgeUTCTime blockHeap currentChain candidat
       blockMatchesHeader     = \_ _ -> True,
 
       headerForgeUTCTime     = headerFieldsForgeUTCTime,
-      blockForgeUTCTime      = headerFieldsForgeUTCTime
+      blockForgeUTCTime      = headerFieldsForgeUTCTime,
+
+      readChainSelStarvation = pure (ChainSelStarvationEndedAt (Time 0)),
+      demoteCSJDynamo = \_ -> pure ()
       }
   where
     plausibleCandidateChain cur candidate =