chore: group the creation of bouncers similar to artifact pools (dfinity#3474)

Author: rumenov

rs/artifact_pool/src/idkg_pool.rs

@@ -12,17 +12,14 @@ use crate::{
     IntoInner,
 };
 use ic_config::artifact_pool::{ArtifactPoolConfig, PersistentPoolBackend};
+use ic_interfaces::idkg::{
+    IDkgChangeAction, IDkgChangeSet, IDkgPool, IDkgPoolSection, IDkgPoolSectionOp,
+    IDkgPoolSectionOps, MutableIDkgPoolSection,
+};
 use ic_interfaces::p2p::consensus::{
     ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, MutablePool, UnvalidatedArtifact,
     ValidatedPoolReader,
 };
-use ic_interfaces::{
-    idkg::{
-        IDkgChangeAction, IDkgChangeSet, IDkgPool, IDkgPoolSection, IDkgPoolSectionOp,
-        IDkgPoolSectionOps, MutableIDkgPoolSection,
-    },
-    time_source::TimeSource,
-};
 use ic_logger::{info, warn, ReplicaLogger};
 use ic_metrics::MetricsRegistry;
 use ic_types::consensus::{
@@ -380,11 +377,7 @@ impl IDkgPoolImpl {
     }
 
     // Populates the unvalidated pool with the initial dealings from the CUP.
-    pub fn add_initial_dealings(
-        &mut self,
-        catch_up_package: &CatchUpPackage,
-        time_source: &dyn TimeSource,
-    ) {
+    pub fn add_initial_dealings(&mut self, catch_up_package: &CatchUpPackage) {
         let block = catch_up_package.content.block.get_value();
 
         let mut initial_dealings = Vec::new();
@@ -413,7 +406,7 @@ impl IDkgPoolImpl {
             self.insert(UnvalidatedArtifact {
                 message: IDkgMessage::Dealing(signed_dealing.clone()),
                 peer_id: signed_dealing.dealer_id(),
-                timestamp: time_source.get_relative_time(),
+                timestamp: block.context.time,
             })
         }
     }

rs/replica/setup_ic_network/src/lib.rs

@@ -28,10 +28,10 @@ use ic_https_outcalls_consensus::{
 use ic_ingress_manager::{bouncer::IngressBouncer, IngressManager, RandomStateKind};
 use ic_interfaces::{
     batch_payload::BatchPayloadBuilder,
+    consensus_pool::{ConsensusBlockCache, ConsensusPoolCache},
     execution_environment::IngressHistoryReader,
     messaging::{MessageRouting, XNetPayloadBuilder},
-    p2p::artifact_manager::JoinGuard,
-    p2p::state_sync::StateSyncClient,
+    p2p::{artifact_manager::JoinGuard, state_sync::StateSyncClient},
     self_validating_payload::SelfValidatingPayloadBuilder,
     time_source::{SysTimeSource, TimeSource},
 };
@@ -85,6 +85,105 @@ struct ArtifactPools {
     canister_http_pool: Arc<RwLock<CanisterHttpPoolImpl>>,
 }
 
+impl ArtifactPools {
+    fn new(
+        log: &ReplicaLogger,
+        metrics_registry: &MetricsRegistry,
+        node_id: NodeId,
+        config: ArtifactPoolConfig,
+        catch_up_package: &CatchUpPackage,
+    ) -> Self {
+        let ingress_pool = Arc::new(RwLock::new(IngressPoolImpl::new(
+            node_id,
+            config.clone(),
+            metrics_registry.clone(),
+            log.clone(),
+        )));
+
+        let mut idkg_pool = IDkgPoolImpl::new(
+            config.clone(),
+            log.clone(),
+            metrics_registry.clone(),
+            Box::new(idkg::IDkgStatsImpl::new(metrics_registry.clone())),
+        );
+        idkg_pool.add_initial_dealings(catch_up_package);
+        let idkg_pool = Arc::new(RwLock::new(idkg_pool));
+
+        let certification_pool = Arc::new(RwLock::new(CertificationPoolImpl::new(
+            node_id,
+            config,
+            log.clone(),
+            metrics_registry.clone(),
+        )));
+        let dkg_pool = Arc::new(RwLock::new(DkgPoolImpl::new(
+            metrics_registry.clone(),
+            log.clone(),
+        )));
+        let canister_http_pool = Arc::new(RwLock::new(CanisterHttpPoolImpl::new(
+            metrics_registry.clone(),
+            log.clone(),
+        )));
+        Self {
+            ingress_pool,
+            certification_pool,
+            dkg_pool,
+            idkg_pool,
+            canister_http_pool,
+        }
+    }
+}
+
+struct Bouncers {
+    ingress: Arc<IngressBouncer>,
+    consensus: Arc<ConsensusBouncer>,
+    certifier: Arc<CertifierBouncer>,
+    dkg: Arc<DkgBouncer>,
+    idkg: Arc<idkg::IDkgBouncer>,
+    https_outcalls: Arc<CanisterHttpGossipImpl>,
+}
+
+impl Bouncers {
+    fn new(
+        log: &ReplicaLogger,
+        metrics_registry: &MetricsRegistry,
+        subnet_id: SubnetId,
+        time_source: Arc<dyn TimeSource>,
+        message_router: Arc<dyn MessageRouting>,
+        consensus_pool_cache: Arc<dyn ConsensusPoolCache>,
+        consensus_block_cache: Arc<dyn ConsensusBlockCache>,
+        state_reader: Arc<dyn StateReader<State = ReplicatedState>>,
+    ) -> Self {
+        let ingress = Arc::new(IngressBouncer::new(time_source.clone()));
+        let consensus = Arc::new(ConsensusBouncer::new(metrics_registry, message_router));
+        let dkg = Arc::new(DkgBouncer::new(metrics_registry));
+        let certifier = Arc::new(CertifierBouncer::new(
+            metrics_registry,
+            consensus_pool_cache.clone(),
+        ));
+        let idkg = Arc::new(idkg::IDkgBouncer::new(
+            metrics_registry,
+            subnet_id,
+            consensus_block_cache,
+            state_reader.clone(),
+        ));
+
+        let https_outcalls = Arc::new(CanisterHttpGossipImpl::new(
+            consensus_pool_cache.clone(),
+            state_reader.clone(),
+            log.clone(),
+        ));
+
+        Self {
+            ingress,
+            consensus,
+            dkg,
+            idkg,
+            certifier,
+            https_outcalls,
+        }
+    }
+}
+
 pub type CanisterHttpAdapterClient =
     Box<dyn NonBlockingChannel<CanisterHttpRequest, Response = CanisterHttpResponse> + Send>;
 
@@ -138,7 +237,7 @@ pub fn setup_consensus_and_p2p(
         node_id,
         subnet_id,
         artifact_pool_config,
-        catch_up_package,
+        &catch_up_package,
         Arc::clone(&consensus_crypto) as Arc<_>,
         Arc::clone(&certifier_crypto) as Arc<_>,
         Arc::clone(&ingress_sig_crypto) as Arc<_>,
@@ -219,7 +318,7 @@ fn start_consensus(
     node_id: NodeId,
     subnet_id: SubnetId,
     artifact_pool_config: ArtifactPoolConfig,
-    catch_up_package: CatchUpPackage,
+    catch_up_package: &CatchUpPackage,
     // ConsensusCrypto is an extension of the Crypto trait and we can
     // not downcast traits.
     consensus_crypto: Arc<dyn ConsensusCrypto>,
@@ -245,26 +344,34 @@ fn start_consensus(
     Vec<Box<dyn JoinGuard>>,
     AbortableBroadcastChannelBuilder,
 ) {
+    let artifact_pools = ArtifactPools::new(
+        log,
+        metrics_registry,
+        node_id,
+        artifact_pool_config,
+        catch_up_package,
+    );
     let time_source = Arc::new(SysTimeSource::new());
+    let consensus_pool_cache = consensus_pool.read().unwrap().get_cache();
+    let consensus_block_cache = consensus_pool.read().unwrap().get_block_cache();
+    let bouncers = Bouncers::new(
+        log,
+        metrics_registry,
+        subnet_id,
+        time_source.clone(),
+        message_router.clone(),
+        consensus_pool_cache.clone(),
+        consensus_block_cache,
+        state_reader.clone(),
+    );
+
     let mut new_p2p_consensus: ic_consensus_manager::AbortableBroadcastChannelBuilder =
         ic_consensus_manager::AbortableBroadcastChannelBuilder::new(
             log.clone(),
             rt_handle.clone(),
             metrics_registry.clone(),
         );
 
-    let artifact_pools = init_artifact_pools(
-        node_id,
-        artifact_pool_config,
-        metrics_registry,
-        log,
-        catch_up_package,
-        time_source.as_ref(),
-    );
-
-    let mut join_handles = vec![];
-
-    let consensus_pool_cache = consensus_pool.read().unwrap().get_cache();
     let consensus_time = consensus_pool.read().unwrap().get_consensus_time();
     let replica_config = ReplicaConfig { node_id, subnet_id };
     let ingress_manager = Arc::new(IngressManager::new(
@@ -302,6 +409,8 @@ fn start_consensus(
         &PoolReader::new(&*consensus_pool.read().unwrap()),
     )));
 
+    let mut join_handles = vec![];
+
     {
         let consensus_impl = ConsensusImpl::new(
             replica_config.clone(),
@@ -327,14 +436,13 @@ fn start_consensus(
 
         let consensus_pool = Arc::clone(&consensus_pool);
 
-        let bouncer = Arc::new(ConsensusBouncer::new(metrics_registry, message_router));
         let (outbound_tx, inbound_rx) = if HASHES_IN_BLOCKS_FEATURE_ENABLED {
             let assembler = ic_artifact_downloader::FetchStrippedConsensusArtifact::new(
                 log.clone(),
                 rt_handle.clone(),
                 consensus_pool.clone(),
                 artifact_pools.ingress_pool.clone(),
-                bouncer,
+                bouncers.consensus,
                 metrics_registry.clone(),
                 node_id,
             );
@@ -344,7 +452,7 @@ fn start_consensus(
                 log.clone(),
                 rt_handle.clone(),
                 consensus_pool.clone(),
-                bouncer,
+                bouncers.consensus,
                 metrics_registry.clone(),
             );
             new_p2p_consensus.abortable_broadcast_channel(assembler, SLOT_TABLE_NO_LIMIT)
@@ -366,12 +474,11 @@ fn start_consensus(
     let user_ingress_tx = {
         #[allow(clippy::disallowed_methods)]
         let (user_ingress_tx, user_ingress_rx) = unbounded_channel();
-        let bouncer = Arc::new(IngressBouncer::new(time_source.clone()));
         let assembler = ic_artifact_downloader::FetchArtifact::new(
             log.clone(),
             rt_handle.clone(),
             artifact_pools.ingress_pool.clone(),
-            bouncer,
+            bouncers.ingress,
             metrics_registry.clone(),
         );
 
@@ -402,12 +509,11 @@ fn start_consensus(
             log.clone(),
             max_certified_height_tx,
         );
-        let bouncer = CertifierBouncer::new(metrics_registry, Arc::clone(&consensus_pool_cache));
         let assembler = ic_artifact_downloader::FetchArtifact::new(
             log.clone(),
             rt_handle.clone(),
             artifact_pools.certification_pool.clone(),
-            Arc::new(bouncer),
+            bouncers.certifier,
             metrics_registry.clone(),
         );
 
@@ -426,12 +532,11 @@ fn start_consensus(
     };
 
     {
-        let bouncer = Arc::new(DkgBouncer::new(metrics_registry));
         let assembler = ic_artifact_downloader::FetchArtifact::new(
             log.clone(),
             rt_handle.clone(),
             artifact_pools.dkg_pool.clone(),
-            bouncer,
+            bouncers.dkg,
             metrics_registry.clone(),
         );
 
@@ -469,17 +574,11 @@ fn start_consensus(
             finalized.payload.as_ref().is_summary(),
             finalized.payload.as_ref().as_idkg().is_some(),
         );
-        let bouncer = Arc::new(idkg::IDkgBouncer::new(
-            metrics_registry,
-            subnet_id,
-            consensus_pool.read().unwrap().get_block_cache(),
-            Arc::clone(&state_reader),
-        ));
         let assembler = ic_artifact_downloader::FetchArtifact::new(
             log.clone(),
             rt_handle.clone(),
             artifact_pools.idkg_pool.clone(),
-            bouncer,
+            bouncers.idkg,
             metrics_registry.clone(),
         );
 
@@ -506,16 +605,11 @@ fn start_consensus(
     };
 
     {
-        let bouncer = Arc::new(CanisterHttpGossipImpl::new(
-            Arc::clone(&consensus_pool_cache),
-            Arc::clone(&state_reader),
-            log.clone(),
-        ));
         let assembler = ic_artifact_downloader::FetchArtifact::new(
             log.clone(),
             rt_handle.clone(),
             artifact_pools.canister_http_pool.clone(),
-            bouncer,
+            bouncers.https_outcalls,
             metrics_registry.clone(),
         );
 
@@ -549,50 +643,3 @@ fn start_consensus(
         new_p2p_consensus,
     )
 }
-
-fn init_artifact_pools(
-    node_id: NodeId,
-    config: ArtifactPoolConfig,
-    metrics_registry: &MetricsRegistry,
-    log: &ReplicaLogger,
-    catch_up_package: CatchUpPackage,
-    time_source: &dyn TimeSource,
-) -> ArtifactPools {
-    let ingress_pool = Arc::new(RwLock::new(IngressPoolImpl::new(
-        node_id,
-        config.clone(),
-        metrics_registry.clone(),
-        log.clone(),
-    )));
-
-    let mut idkg_pool = IDkgPoolImpl::new(
-        config.clone(),
-        log.clone(),
-        metrics_registry.clone(),
-        Box::new(idkg::IDkgStatsImpl::new(metrics_registry.clone())),
-    );
-    idkg_pool.add_initial_dealings(&catch_up_package, time_source);
-    let idkg_pool = Arc::new(RwLock::new(idkg_pool));
-
-    let certification_pool = Arc::new(RwLock::new(CertificationPoolImpl::new(
-        node_id,
-        config,
-        log.clone(),
-        metrics_registry.clone(),
-    )));
-    let dkg_pool = Arc::new(RwLock::new(DkgPoolImpl::new(
-        metrics_registry.clone(),
-        log.clone(),
-    )));
-    let canister_http_pool = Arc::new(RwLock::new(CanisterHttpPoolImpl::new(
-        metrics_registry.clone(),
-        log.clone(),
-    )));
-    ArtifactPools {
-        ingress_pool,
-        certification_pool,
-        dkg_pool,
-        idkg_pool,
-        canister_http_pool,
-    }
-}