Integrate [ouroboros-network #4795](IntersectMBO/ouroboros-network#4795)

IntersectMBO · Mar 27, 2024 · 449d7cf · 449d7cf
1 parent 5a6176b
commit 449d7cf
Show file tree

Hide file tree

Showing 4 changed files with 34 additions and 26 deletions.
diff --git a/...sus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs b/...sus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs
@@ -48,6 +48,7 @@ import           Control.Monad.Class.MonadTimer.SI (MonadTimer)
 import           Control.Tracer
 import           Data.ByteString.Lazy (ByteString)
 import qualified Data.ByteString.Lazy as BSL
+import           Data.Hashable (Hashable)
 import           Data.Int (Int64)
 import           Data.Map.Strict (Map)
 import           Data.Void (Void)
@@ -110,8 +111,7 @@ import           Ouroboros.Network.Protocol.PeerSharing.Codec
                      codecPeerSharingId, timeLimitsPeerSharing)
 import           Ouroboros.Network.Protocol.PeerSharing.Server
                      (PeerSharingServer, peerSharingServerPeer)
-import           Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharing,
-                     PeerSharingAmount)
+import           Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharing)
 import           Ouroboros.Network.Protocol.TxSubmission2.Client
 import           Ouroboros.Network.Protocol.TxSubmission2.Codec
 import           Ouroboros.Network.Protocol.TxSubmission2.Server
@@ -204,16 +204,14 @@ mkHandlers ::
      , HasTxId (GenTx blk)
      , LedgerSupportsProtocol blk
      , Ord addrNTN
+     , Hashable addrNTN
      )
   => NodeKernelArgs m addrNTN addrNTC blk
   -> NodeKernel     m addrNTN addrNTC blk
-  -> (PeerSharingAmount -> m [addrNTN])
-  -- ^ Peer Sharing result computation callback
   -> Handlers       m addrNTN           blk
 mkHandlers
       NodeKernelArgs {chainSyncFutureCheck, keepAliveRng, miniProtocolParameters}
-      NodeKernel {getChainDB, getMempool, getTopLevelConfig, getTracers = tracers}
-      computePeers =
+      NodeKernel {getChainDB, getMempool, getTopLevelConfig, getTracers = tracers, getPeerSharingAPI} =
     Handlers {
         hChainSyncClient = \peer _isBigLedgerpeer dynEnv ->
           CsClient.chainSyncClient
@@ -257,7 +255,7 @@ mkHandlers
       , hKeepAliveClient = \_version -> keepAliveClient (Node.keepAliveClientTracer tracers) keepAliveRng
       , hKeepAliveServer = \_version _peer -> keepAliveServer
       , hPeerSharingClient = \_version controlMessageSTM _peer -> peerSharingClient controlMessageSTM
-      , hPeerSharingServer = \_version _peer -> peerSharingServer computePeers
+      , hPeerSharingServer = \_version _peer -> peerSharingServer getPeerSharingAPI
       }
 
 {-------------------------------------------------------------------------------

diff --git a/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node.hs b/ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node.hs
@@ -133,15 +133,14 @@ import           Ouroboros.Network.PeerSelection.PeerSharing (PeerSharing)
 import           Ouroboros.Network.PeerSelection.PeerSharing.Codec
                      (decodeRemoteAddress, encodeRemoteAddress)
 import           Ouroboros.Network.Protocol.Limits (shortWait)
-import           Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharingAmount)
 import           Ouroboros.Network.RethrowPolicy
 import qualified SafeWildCards
 import           System.Exit (ExitCode (..))
 import           System.FilePath ((</>))
 import           System.FS.API (SomeHasFS (..))
 import           System.FS.API.Types
 import           System.FS.IO (ioHasFS)
-import           System.Random (StdGen, newStdGen, randomIO, randomRIO)
+import           System.Random (StdGen, newStdGen, randomIO, randomRIO, split)
 
 {-------------------------------------------------------------------------------
   The arguments to the Consensus Layer node functionality
@@ -503,8 +502,6 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
       -> (NodeToNodeVersion -> addrNTN -> CBOR.Encoding)
       -> (NodeToNodeVersion -> forall s . CBOR.Decoder s addrNTN)
       -> BlockNodeToNodeVersion blk
-      -> (PeerSharingAmount -> m [addrNTN])
-      -- ^ Peer Sharing result computation callback
       -> NTN.Apps m
           addrNTN
           ByteString
@@ -514,7 +511,7 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
           ByteString
           NodeToNodeInitiatorResult
           ()
-    mkNodeToNodeApps nodeKernelArgs nodeKernel peerMetrics encAddrNTN decAddrNTN version computePeers =
+    mkNodeToNodeApps nodeKernelArgs nodeKernel peerMetrics encAddrNTN decAddrNTN version =
         NTN.mkApps
           nodeKernel
           rnTraceNTN
@@ -523,7 +520,7 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
           llrnChainSyncTimeout
           llrnChainSyncLoPBucketConfig
           (reportMetric Diffusion.peerMetricsConfiguration peerMetrics)
-          (NTN.mkHandlers nodeKernelArgs nodeKernel computePeers)
+          (NTN.mkHandlers nodeKernelArgs nodeKernel)
 
     mkNodeToClientApps
       :: NodeKernelArgs m addrNTN (ConnectionId addrNTC) blk
@@ -542,8 +539,6 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
       :: NetworkP2PMode p2p
       -> MiniProtocolParameters
       -> (   BlockNodeToNodeVersion blk
-          -- Peer Sharing result computation callback
-          -> (PeerSharingAmount -> m [addrNTN])
           -> NTN.Apps
                m
                addrNTN
@@ -606,16 +601,16 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
                       -- Initiator side won't start responder side of Peer
                       -- Sharing protocol so we give a dummy implementation
                       -- here.
-                      $ ntnApps blockVersion (error "impossible happened!"))
+                      $ ntnApps blockVersion)
                 | (version, blockVersion) <- Map.toList llrnNodeToNodeVersions
                 ],
-            Diffusion.daApplicationInitiatorResponderMode = \computePeers ->
+            Diffusion.daApplicationInitiatorResponderMode =
               combineVersions
                 [ simpleSingletonVersions
                     version
                     llrnVersionDataNTN
                     (NTN.initiatorAndResponder miniProtocolParams version rnPeerSharing
-                      $ ntnApps blockVersion computePeers)
+                      $ ntnApps blockVersion)
                 | (version, blockVersion) <- Map.toList llrnNodeToNodeVersions
                 ],
             Diffusion.daLocalResponderApplication =
@@ -746,7 +741,7 @@ mkNodeKernelArgs
   registry
   bfcSalt
   gsmAntiThunderingHerd
-  keepAliveRng
+  rng
   cfg
   tracers
   btime
@@ -757,6 +752,7 @@ mkNodeKernelArgs
   gsmMarkerFileView
   getUseBootstrapPeers
   = do
+    let (kaRng, psRng) = split rng
     return NodeKernelArgs
       { tracers
       , registry
@@ -769,14 +765,15 @@ mkNodeKernelArgs
       , mempoolCapacityOverride = NoMempoolCapacityBytesOverride
       , miniProtocolParameters  = defaultMiniProtocolParameters
       , blockFetchConfiguration = defaultBlockFetchConfiguration
-      , keepAliveRng
       , gsmArgs = GsmNodeKernelArgs {
           gsmAntiThunderingHerd
         , gsmDurationUntilTooOld
         , gsmMarkerFileView
         , gsmMinCaughtUpDuration = maxCaughtUpAge
         }
       , getUseBootstrapPeers
+      , keepAliveRng = kaRng
+      , peerSharingRng = psRng
       }
   where
     defaultBlockFetchConfiguration :: BlockFetchConfiguration

diff --git a/...s-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs b/...s-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs
@@ -90,8 +90,10 @@ import           Ouroboros.Network.NodeToNode (ConnectionId,
 import           Ouroboros.Network.PeerSelection.Bootstrap (UseBootstrapPeers)
 import           Ouroboros.Network.PeerSelection.LedgerPeers.Type
                      (LedgerStateJudgement (..))
-import           Ouroboros.Network.PeerSharing (PeerSharingRegistry,
-                     newPeerSharingRegistry)
+import           Ouroboros.Network.PeerSharing (PeerSharingAPI,
+                     PeerSharingRegistry, newPeerSharingAPI,
+                     newPeerSharingRegistry, ps_POLICY_PEER_SHARE_MAX_PEERS,
+                     ps_POLICY_PEER_SHARE_STICKY_TIME)
 import           Ouroboros.Network.TxSubmission.Inbound
                      (TxSubmissionMempoolWriter)
 import qualified Ouroboros.Network.TxSubmission.Inbound as Inbound
@@ -144,7 +146,9 @@ data NodeKernel m addrNTN addrNTC blk = NodeKernel {
       --
       -- When set with the empty list '[]' block forging will be disabled.
       --
-    , setBlockForging         :: [BlockForging m blk] -> m ()
+    , setBlockForging        :: [BlockForging m blk] -> m ()
+
+    , getPeerSharingAPI      :: PeerSharingAPI addrNTN StdGen m
     }
 
 -- | Arguments required when initializing a node
@@ -163,6 +167,7 @@ data NodeKernelArgs m addrNTN addrNTC blk = NodeKernelArgs {
     , keepAliveRng            :: StdGen
     , gsmArgs                 :: GsmNodeKernelArgs m blk
     , getUseBootstrapPeers    :: STM m UseBootstrapPeers
+    , peerSharingRng          :: StdGen
     }
 
 initNodeKernel ::
@@ -180,6 +185,7 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
                                    , chainDB, initChainDB
                                    , blockFetchConfiguration
                                    , gsmArgs
+                                   , peerSharingRng
                                    } = do
     -- using a lazy 'TVar', 'BlockForging' does not have a 'NoThunks' instance.
     blockForgingVar :: LazySTM.TMVar m [BlockForging m blk] <- LazySTM.newTMVarIO []
@@ -240,6 +246,10 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
           TooOld      -> GSM.enterPreSyncing gsm
           YoungEnough -> GSM.enterCaughtUp   gsm
 
+    peerSharingAPI <- newPeerSharingAPI peerSharingRng
+                                        ps_POLICY_PEER_SHARE_STICKY_TIME
+                                        ps_POLICY_PEER_SHARE_MAX_PEERS
+
     void $ forkLinkedThread registry "NodeKernel.blockForging" $
                             blockForgingController st (LazySTM.takeTMVar blockForgingVar)
 
@@ -265,6 +275,7 @@ initNodeKernel args@NodeKernelArgs { registry, cfg, tracers
       , getPeerSharingRegistry  = peerSharingRegistry
       , getTracers              = tracers
       , setBlockForging         = \a -> atomically . LazySTM.putTMVar blockForgingVar $! a
+      , getPeerSharingAPI      = peerSharingAPI
       }
   where
     blockForgingController :: InternalState m remotePeer localPeer blk

diff --git a/ouroboros-consensus-diffusion/src/unstable-diffusion-testlib/Test/ThreadNet/Network.hs b/ouroboros-consensus-diffusion/src/unstable-diffusion-testlib/Test/ThreadNet/Network.hs
@@ -115,7 +115,7 @@ import           Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharing)
 import           Ouroboros.Network.Protocol.TxSubmission2.Type
 import qualified System.FS.Sim.MockFS as Mock
 import           System.FS.Sim.MockFS (MockFS)
-import           System.Random (mkStdGen)
+import           System.Random (mkStdGen, split)
 import           Test.ThreadNet.TxGen
 import           Test.ThreadNet.Util.NodeJoinPlan
 import           Test.ThreadNet.Util.NodeRestarts
@@ -969,8 +969,9 @@ runThreadNetwork systemTime ThreadNetworkArgs
         , hfbtMaxClockRewind = secondsToNominalDiffTime 0
         }
 
-      let kaRng = case seed of
+      let rng = case seed of
                     Seed s -> mkStdGen s
+          (kaRng, psRng) = split rng
       let nodeKernelArgs = NodeKernelArgs
             { tracers
             , registry
@@ -985,6 +986,7 @@ runThreadNetwork systemTime ThreadNetworkArgs
             , blockFetchSize          = estimateBlockSize
             , mempoolCapacityOverride = NoMempoolCapacityBytesOverride
             , keepAliveRng            = kaRng
+            , peerSharingRng          = psRng
             , miniProtocolParameters  = MiniProtocolParameters {
                   chainSyncPipeliningHighMark = 4,
                   chainSyncPipeliningLowMark  = 2,
@@ -1040,7 +1042,7 @@ runThreadNetwork systemTime ThreadNetworkArgs
                   -- The purpose of this test is not testing protocols, so
                   -- returning constant empty list is fine if we have thorough
                   -- tests about the peer sharing protocol itself.
-                  (NTN.mkHandlers nodeKernelArgs nodeKernel (\_ -> return []))
+                  (NTN.mkHandlers nodeKernelArgs nodeKernel)
 
       -- In practice, a robust wallet/user can persistently add a transaction
       -- until it appears on the chain. This thread adds robustness for the