Plug rotateDynamo into BlockFetchConsensusInterface

Author: Niols

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs

@@ -43,7 +43,6 @@ import           Data.Function (on)
 import           Data.Functor ((<&>))
 import           Data.Hashable (Hashable)
 import           Data.List.NonEmpty (NonEmpty)
-import           Data.Map.Strict (Map)
 import           Data.Maybe (isJust, mapMaybe)
 import           Data.Proxy
 import qualified Data.Text as Text
@@ -65,7 +64,7 @@ import qualified Ouroboros.Consensus.MiniProtocol.BlockFetch.ClientInterface as
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
                      (ChainSyncClientHandle (..),
                      ChainSyncClientHandleCollection (..), ChainSyncState (..),
-                     newChainSyncClientHandleCollection, viewChainSyncState)
+                     newChainSyncClientHandleCollection)
 import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
                      (SomeHeaderInFutureCheck)
 import           Ouroboros.Consensus.Node.Genesis (GenesisNodeKernelArgs (..),
@@ -386,9 +385,6 @@ initInternalState NodeKernelArgs { tracers, chainDB, registry, cfg
 
     fetchClientRegistry <- newFetchClientRegistry
 
-    let getCandidates :: STM m (Map (ConnectionId addrNTN) (AnchoredFragment (Header blk)))
-        getCandidates = viewChainSyncState (cschcMap varChainSyncHandles) csCandidate
-
     slotForgeTimeOracle <- BlockFetchClientInterface.initSlotForgeTimeOracle cfg chainDB
     let readFetchMode = BlockFetchClientInterface.readFetchModeDefault
           btime
@@ -399,7 +395,7 @@ initInternalState NodeKernelArgs { tracers, chainDB, registry, cfg
         blockFetchInterface = BlockFetchClientInterface.mkBlockFetchConsensusInterface
           (configBlock cfg)
           (BlockFetchClientInterface.defaultChainDbView chainDB)
-          getCandidates
+          varChainSyncHandles
           blockFetchSize
           slotForgeTimeOracle
           readFetchMode

ouroboros-consensus-diffusion/test/consensus-test/Test/Consensus/PeerSimulator/BlockFetch.hs

@@ -23,21 +23,20 @@ import           Control.Monad.Class.MonadTime
 import           Control.Monad.Class.MonadTimer.SI (MonadTimer)
 import           Control.Tracer (Tracer, nullTracer, traceWith)
 import           Data.Functor.Contravariant ((>$<))
-import           Data.Map.Strict (Map)
 import           Network.TypedProtocol.Codec (AnyMessage, PeerHasAgency (..),
                      PeerRole)
 import           Ouroboros.Consensus.Block (HasHeader)
 import           Ouroboros.Consensus.Block.Abstract (Header, Point (..))
 import qualified Ouroboros.Consensus.MiniProtocol.BlockFetch.ClientInterface as BlockFetchClientInterface
+import           Ouroboros.Consensus.MiniProtocol.ChainSync.Client
+                     (ChainSyncClientHandleCollection)
 import           Ouroboros.Consensus.Node.ProtocolInfo
                      (NumCoreNodes (NumCoreNodes))
 import           Ouroboros.Consensus.Storage.ChainDB.API
 import           Ouroboros.Consensus.Util (ShowProxy)
 import           Ouroboros.Consensus.Util.IOLike (DiffTime,
-                     Exception (fromException), IOLike, STM, atomically, retry,
-                     try)
+                     Exception (fromException), IOLike, atomically, retry, try)
 import           Ouroboros.Consensus.Util.ResourceRegistry
-import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import           Ouroboros.Network.BlockFetch (BlockFetchConfiguration (..),
                      FetchClientRegistry, FetchMode (..), blockFetchLogic,
                      bracketFetchClient, bracketKeepAliveClient)
@@ -78,17 +77,17 @@ startBlockFetchLogic ::
   -> Tracer m (TraceEvent TestBlock)
   -> ChainDB m TestBlock
   -> FetchClientRegistry PeerId (Header TestBlock) TestBlock m
-  -> STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
+  -> ChainSyncClientHandleCollection PeerId m TestBlock
   -> m ()
-startBlockFetchLogic registry tracer chainDb fetchClientRegistry getCandidates = do
+startBlockFetchLogic registry tracer chainDb fetchClientRegistry csHandlesCol = do
     let slotForgeTime :: BlockFetchClientInterface.SlotForgeTimeOracle m blk
         slotForgeTime _ = pure dawnOfTime
 
         blockFetchConsensusInterface =
           BlockFetchClientInterface.mkBlockFetchConsensusInterface
             (TestBlockConfig $ NumCoreNodes 0) -- Only needed when minting blocks
             (BlockFetchClientInterface.defaultChainDbView chainDb)
-            getCandidates
+            csHandlesCol
             -- The size of headers in bytes is irrelevant because our tests
             -- do not serialize the blocks.
             (\_hdr -> 1000)

ouroboros-consensus-diffusion/test/consensus-test/Test/Consensus/PeerSimulator/Run.hs

@@ -336,9 +336,7 @@ startNode ::
   LiveInterval TestBlock m ->
   m ()
 startNode schedulerConfig genesisTest interval = do
-  let
-      handles = psrHandles lrPeerSim
-      getCandidates = viewChainSyncState (cschcMap handles) CSClient.csCandidate
+  let handles = psrHandles lrPeerSim
   fetchClientRegistry <- newFetchClientRegistry
   let chainDbView = CSClient.defaultChainDbView lnChainDb
       activePeers = Map.toList $ Map.restrictKeys (psrPeers lrPeerSim) (lirActive liveResult)
@@ -385,7 +383,7 @@ startNode schedulerConfig genesisTest interval = do
   -- The block fetch logic needs to be started after the block fetch clients
   -- otherwise, an internal assertion fails because getCandidates yields more
   -- peer fragments than registered clients.
-  BlockFetch.startBlockFetchLogic lrRegistry lrTracer lnChainDb fetchClientRegistry getCandidates
+  BlockFetch.startBlockFetchLogic lrRegistry lrTracer lnChainDb fetchClientRegistry handles
 
   for_ lrLoEVar $ \ var -> do
       forkLinkedWatcher lrRegistry "LoE updater background" $

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/BlockFetch/ClientInterface.hs

@@ -26,6 +26,10 @@ import qualified Ouroboros.Consensus.HardFork.Abstract as History
 import qualified Ouroboros.Consensus.HardFork.History as History
 import           Ouroboros.Consensus.Ledger.Abstract
 import           Ouroboros.Consensus.Ledger.Extended
+import           Ouroboros.Consensus.Ledger.SupportsProtocol
+                     (LedgerSupportsProtocol)
+import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client as CSClient
+import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client.Jumping as Jumping
 import           Ouroboros.Consensus.Storage.ChainDB.API (ChainDB)
 import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
 import           Ouroboros.Consensus.Storage.ChainDB.API.Types.InvalidBlockPunishment
@@ -171,21 +175,25 @@ mkBlockFetchConsensusInterface ::
      forall m peer blk.
      ( IOLike m
      , BlockSupportsDiffusionPipelining blk
-     , BlockSupportsProtocol blk
+     , Ord peer
+     , LedgerSupportsProtocol blk
      )
   => BlockConfig blk
   -> ChainDbView m blk
-  -> STM m (Map peer (AnchoredFragment (Header blk)))
+  -> CSClient.ChainSyncClientHandleCollection peer m blk
   -> (Header blk -> SizeInBytes)
   -> SlotForgeTimeOracle m blk
      -- ^ Slot forge time, see 'headerForgeUTCTime' and 'blockForgeUTCTime'.
   -> STM m FetchMode
      -- ^ See 'readFetchMode'.
   -> BlockFetchConsensusInterface peer (Header blk) blk m
 mkBlockFetchConsensusInterface
-  bcfg chainDB getCandidates blockFetchSize slotForgeTime readFetchMode =
+  bcfg chainDB csHandlesCol blockFetchSize slotForgeTime readFetchMode =
     BlockFetchConsensusInterface {..}
   where
+    getCandidates :: STM m (Map peer (AnchoredFragment (Header blk)))
+    getCandidates = CSClient.viewChainSyncState (CSClient.cschcMap csHandlesCol) CSClient.csCandidate
+
     blockMatchesHeader :: Header blk -> blk -> Bool
     blockMatchesHeader = Block.blockMatchesHeader
 
@@ -344,3 +352,6 @@ mkBlockFetchConsensusInterface
     blockForgeUTCTime  = slotForgeTime . blockRealPoint  . unFromConsensus
 
     lastChainSelStarvation = getLastTimeStarved chainDB
+
+    demoteCSJDynamo :: peer -> m ()
+    demoteCSJDynamo = void . atomically . Jumping.rotateDynamo csHandlesCol

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ChainSync/Client/Jumping.hs

@@ -775,20 +775,20 @@ rotateDynamo ::
     LedgerSupportsProtocol blk,
     MonadSTM m
   ) =>
-  Context m peer blk ->
+  ChainSyncClientHandleCollection peer m blk ->
   peer ->
   STM m (Maybe (peer, ChainSyncClientHandle m blk))
-rotateDynamo context peer = do
-  handles <- cschcMap (handlesCol context)
+rotateDynamo handlesCol peer = do
+  handles <- cschcMap handlesCol
   case handles Map.!? peer of
     Nothing ->
       -- Do not re-elect a dynamo if the peer has been disconnected.
-      getDynamo (handlesCol context)
+      getDynamo handlesCol
     Just oldDynHandle ->
       readTVar (cschJumping oldDynHandle) >>= \case
         Dynamo{} -> do
-          cschcRotateHandle (handlesCol context) peer
-          peerStates <- cschcSeq (handlesCol context)
+          cschcRotateHandle handlesCol peer
+          peerStates <- cschcSeq handlesCol
           mEngaged <- findNonDisengaged peerStates
           case mEngaged of
             Nothing ->
@@ -806,7 +806,7 @@ rotateDynamo context peer = do
                 pure $ Just (newDynamoId, newDynHandle)
         _ ->
           -- Do not re-elect a dynamo if the peer is not the dynamo.
-          getDynamo (handlesCol context)
+          getDynamo handlesCol
 
 -- | Choose an unspecified new non-idling dynamo and demote all other peers to
 -- jumpers.