Full rework of the BlockFetch logic for bulk sync mode

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

@@ -1,24 +1,32 @@
 {-# 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
 import Ouroboros.Network.SizeInBytes (SizeInBytes)
 
 
+-- REVIEW: “Bulk Sync” is really not bulk anymore, so maybe we should rename
+-- this? Just “Sync”? Maybe “Genesis Sync”?
 data FetchMode =
        -- | Use this mode when we are catching up on the chain but are stil
        -- well behind. In this mode the fetch logic will optimise for
@@ -149,7 +157,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 +177,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,13 @@ clientBlockFetch sockAddrs maxSlotNo = withIOManager $ \iocp -> do
                       blockFetchPolicy
                       registry
                       (BlockFetchConfiguration {
-                        bfcMaxConcurrencyBulkSync = 1,
                         bfcMaxConcurrencyDeadline = 2,
                         bfcMaxRequestsInflight    = 10,
                         bfcDecisionLoopInterval   = 0.01,
-                        bfcSalt                   = 0
+                        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
@@ -121,6 +124,7 @@ library
                        nothunks,
                        psqueues          >=0.2.3 && <0.3,
                        random,
+                       transformers,
 
                        cardano-prelude,
                        cardano-slotting,
@@ -337,6 +341,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

@@ -44,6 +44,7 @@ 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 +56,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 +65,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 +135,13 @@ blockFetchExample0 decisionTracer clientStateTracer clientMsgTracer
           (sampleBlockFetchPolicy1 headerForgeUTCTime blockHeap currentChainHeaders candidateChainHeaders)
           registry
           (BlockFetchConfiguration {
-            bfcMaxConcurrencyBulkSync = 1,
             bfcMaxConcurrencyDeadline = 2,
             bfcMaxRequestsInflight    = 10,
             bfcDecisionLoopInterval   = 0.01,
-            bfcSalt                   = 0
+            bfcSalt                   = 0,
+            bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+              { gbfcBulkSyncGracePeriod = 10 -- seconds
+              }
           })
         >> return ()
 
@@ -172,8 +175,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
@@ -243,11 +245,13 @@ blockFetchExample1 decisionTracer clientStateTracer clientMsgTracer
           (sampleBlockFetchPolicy1 headerForgeUTCTime blockHeap currentChainHeaders candidateChainHeaders)
           registry
           (BlockFetchConfiguration {
-            bfcMaxConcurrencyBulkSync = 1,
             bfcMaxConcurrencyDeadline = 2,
             bfcMaxRequestsInflight    = 10,
             bfcDecisionLoopInterval   = 0.01,
-            bfcSalt                   = 0
+            bfcSalt                   = 0,
+            bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+              { gbfcBulkSyncGracePeriod = 10 -- seconds
+              }
           })
         >> return ()
 
@@ -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 =

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

@@ -52,6 +52,7 @@ import Ouroboros.Network.NodeToNode.Version (isPipeliningEnabled)
 import Ouroboros.Network.Protocol.BlockFetch.Type (BlockFetch)
 
 import Ouroboros.Network.Testing.Utils
+import Ouroboros.Network.BlockFetch.Decision.Trace (TraceDecisionEvent)
 
 
 --
@@ -206,8 +207,7 @@ chainPoints = map (castPoint . blockPoint)
             . AnchoredFragment.toOldestFirst
 
 data Example1TraceEvent =
-     TraceFetchDecision       [TraceLabelPeer Int
-                                (FetchDecision [Point BlockHeader])]
+     TraceFetchDecision       (TraceDecisionEvent Int BlockHeader)
    | TraceFetchClientState    (TraceLabelPeer Int
                                 (TraceFetchClientState BlockHeader))
    | TraceFetchClientSendRecv (TraceLabelPeer Int
@@ -373,7 +373,7 @@ tracePropertyClientStateSanity es =
     saneStateValues PeerFetchInFlight {..} status =
         -- Here we know the fixed dummy block size so we know exactly what
         -- the bytes in flight should be.
-           Set.size peerFetchBlocksInFlight * 2000
+           Map.size peerFetchBlocksInFlight * 2000
         == fromIntegral peerFetchBytesInFlight
 
      && case status of
@@ -382,7 +382,7 @@ tracePropertyClientStateSanity es =
           _                      -> False -- not used in this test
 
      && if peerFetchReqsInFlight == 0
-           then Set.null peerFetchBlocksInFlight
+           then Map.null peerFetchBlocksInFlight
            else True
 
 

ouroboros-network/sim-tests-lib/Test/Ouroboros/Network/Diffusion/Node.hs

@@ -38,7 +38,7 @@ import Control.Monad.Class.MonadSay
 import Control.Monad.Class.MonadST (MonadST)
 import Control.Monad.Class.MonadThrow (MonadEvaluate, MonadMask, MonadThrow,
            SomeException)
-import Control.Monad.Class.MonadTime.SI (DiffTime, MonadTime)
+import Control.Monad.Class.MonadTime.SI (DiffTime, MonadTime, Time (..))
 import Control.Monad.Class.MonadTimer.SI (MonadDelay, MonadTimer)
 import Control.Monad.Fix (MonadFix)
 import Control.Tracer (Tracer (..), nullTracer)
@@ -66,6 +66,7 @@ import Ouroboros.Network.AnchoredFragment qualified as AF
 import Ouroboros.Network.Block (MaxSlotNo (..), maxSlotNoFromWithOrigin,
            pointSlot)
 import Ouroboros.Network.BlockFetch
+import Ouroboros.Network.BlockFetch.ConsensusInterface (ChainSelStarvation(..))
 import Ouroboros.Network.ConnectionManager.Types (DataFlow (..))
 import Ouroboros.Network.Diffusion qualified as Diff
 import Ouroboros.Network.Diffusion.P2P qualified as Diff.P2P
@@ -290,11 +291,13 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
         (blockFetchPolicy nodeKernel)
         (nkFetchClientRegistry nodeKernel)
         (BlockFetchConfiguration {
-          bfcMaxConcurrencyBulkSync = 1,
           bfcMaxConcurrencyDeadline = 2,
           bfcMaxRequestsInflight    = 10,
           bfcDecisionLoopInterval   = 0.01,
-          bfcSalt                   = 0
+          bfcSalt                   = 0,
+          bfcGenesisBFConfig        = GenesisBlockFetchConfiguration
+            { gbfcBulkSyncGracePeriod = 10 -- seconds
+            }
         })
 
     blockFetchPolicy :: NodeKernel BlockHeader Block s m
@@ -323,7 +326,10 @@ run blockGeneratorArgs limits ni na tracersExtra tracerBlockFetch =
           blockMatchesHeader     = \_ _ -> True,
 
           headerForgeUTCTime,
-          blockForgeUTCTime      = headerForgeUTCTime . fmap blockHeader
+          blockForgeUTCTime      = headerForgeUTCTime . fmap blockHeader,
+
+          readChainSelStarvation = pure (ChainSelStarvationEndedAt (Time 0)),
+          demoteCSJDynamo = \_ -> pure ()
         }
       where
         plausibleCandidateChain cur candidate =

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

@@ -84,6 +84,7 @@ module Ouroboros.Network.BlockFetch
   ( blockFetchLogic
   , BlockFetchConfiguration (..)
   , BlockFetchConsensusInterface (..)
+  , GenesisBlockFetchConfiguration (..)
     -- ** Tracer types
   , FetchDecision
   , TraceFetchClientState (..)
@@ -121,16 +122,14 @@ import Ouroboros.Network.BlockFetch.ConsensusInterface
            (BlockFetchConsensusInterface (..), FromConsensus (..),
            WhetherReceivingTentativeBlocks (..))
 import Ouroboros.Network.BlockFetch.State
+import Ouroboros.Network.BlockFetch.Decision.Trace (TraceDecisionEvent)
 
 
 
 -- | Configuration for FetchDecisionPolicy.
 -- Should be determined by external local node config.
 data BlockFetchConfiguration =
      BlockFetchConfiguration {
-         -- | Maximum concurrent downloads during bulk syncing.
-         bfcMaxConcurrencyBulkSync :: !Word,
-
          -- | Maximum concurrent downloads during deadline syncing.
          bfcMaxConcurrencyDeadline :: !Word,
 
@@ -141,10 +140,22 @@ data BlockFetchConfiguration =
          bfcDecisionLoopInterval   :: !DiffTime,
 
          -- | Salt used when comparing peers
-         bfcSalt                   :: !Int
+         bfcSalt                   :: !Int,
+
+         -- | Genesis-specific parameters
+         bfcGenesisBFConfig        :: !GenesisBlockFetchConfiguration
      }
      deriving (Show)
 
+-- | BlockFetch configuration parameters specific to Genesis.
+data GenesisBlockFetchConfiguration =
+     GenesisBlockFetchConfiguration
+      { -- | Grace period when starting to talk to a peer in bulk sync mode
+        -- during which it is fine if the chain selection gets starved.
+        gbfcBulkSyncGracePeriod    :: !DiffTime
+      }
+      deriving (Show)
+
 -- | Execute the block fetch logic. It monitors the current chain and candidate
 -- chains. It decided which block bodies to fetch and manages the process of
 -- fetching them, including making alternative decisions based on timeouts and
@@ -161,7 +172,7 @@ blockFetchLogic :: forall addr header block m.
                    , Ord addr
                    , Hashable addr
                    )
-                => Tracer m [TraceLabelPeer addr (FetchDecision [Point header])]
+                => Tracer m (TraceDecisionEvent addr header)
                 -> Tracer m (TraceLabelPeer addr (TraceFetchClientState header))
                 -> BlockFetchConsensusInterface addr header block m
                 -> FetchClientRegistry addr header block m
@@ -179,6 +190,7 @@ blockFetchLogic decisionTracer clientStateTracer
       fetchDecisionPolicy
       fetchTriggerVariables
       fetchNonTriggerVariables
+      demoteCSJDynamo
   where
     mkFetchClientPolicy :: WhetherReceivingTentativeBlocks -> STM m (FetchClientPolicy header block m)
     mkFetchClientPolicy receivingTentativeBlocks = do
@@ -194,10 +206,10 @@ blockFetchLogic decisionTracer clientStateTracer
     fetchDecisionPolicy =
       FetchDecisionPolicy {
         maxInFlightReqsPerPeer   = bfcMaxRequestsInflight,
-        maxConcurrencyBulkSync   = bfcMaxConcurrencyBulkSync,
         maxConcurrencyDeadline   = bfcMaxConcurrencyDeadline,
         decisionLoopInterval     = bfcDecisionLoopInterval,
         peerSalt                 = bfcSalt,
+        bulkSyncGracePeriod      = gbfcBulkSyncGracePeriod bfcGenesisBFConfig,
 
         plausibleCandidateChain,
         compareCandidateChains,
@@ -219,5 +231,6 @@ blockFetchLogic decisionTracer clientStateTracer
         readStatePeerStateVars    = readFetchClientsStateVars registry,
         readStatePeerGSVs         = readPeerGSVs registry,
         readStateFetchMode        = readFetchMode,
-        readStateFetchedMaxSlotNo = readFetchedMaxSlotNo
+        readStateFetchedMaxSlotNo = readFetchedMaxSlotNo,
+        readStateChainSelStarvation = readChainSelStarvation
       }