Merge branch 'rumenov/rejjf' into 'master'

chore: move the initialization of some variables only when they are needed and not immediately when the replica is started.

 

See merge request dfinity-lab/public/ic!12590

Author: rumenov

rs/replay/src/player.rs

@@ -249,10 +249,10 @@ impl Player {
         }
 
         let subnet_type = get_subnet_type(
-            registry.as_ref(),
+            &log,
             subnet_id,
             registry.get_latest_version(),
-            &log,
+            registry.as_ref(),
         );
         let metrics_registry = MetricsRegistry::new();
         let subnet_config = SubnetConfig::new(subnet_type);

rs/replica/src/main.rs

@@ -5,18 +5,16 @@ use ic_config::Config;
 use ic_crypto_sha::Sha256;
 use ic_crypto_tls_interfaces::TlsHandshake;
 use ic_http_endpoints_metrics::MetricsHttpEndpoint;
-use ic_interfaces_registry::{LocalStoreCertifiedTimeReader, RegistryClient};
 use ic_logger::{info, new_replica_logger_from_config};
 use ic_metrics::MetricsRegistry;
 use ic_onchain_observability_server::spawn_onchain_observability_grpc_server_and_register_metrics;
-use ic_registry_client_helpers::subnet::SubnetRegistry;
 use ic_replica::setup;
 use ic_sys::PAGE_SIZE;
 use ic_types::consensus::CatchUpPackage;
 use ic_types::{replica_version::REPLICA_BINARY_HASH, PrincipalId, ReplicaVersion, SubnetId};
 use nix::unistd::{setpgid, Pid};
 use static_assertions::assert_eq_size;
-use std::{env, io, path::PathBuf, str::FromStr, sync::Arc, time::Duration};
+use std::{env, fs, io, path::PathBuf, str::FromStr, sync::Arc, time::Duration};
 use tokio::signal::unix::{signal, SignalKind};
 
 #[cfg(target_os = "linux")]
@@ -30,7 +28,6 @@ use jemallocator::Jemalloc;
 #[cfg(target_os = "linux")]
 static ALLOC: Jemalloc = Jemalloc;
 
-use ic_registry_local_store::LocalStoreImpl;
 #[cfg(feature = "profiler")]
 use pprof::{protos::Message, ProfilerGuard};
 #[cfg(feature = "profiler")]
@@ -43,15 +40,15 @@ use std::io::Write;
 /// Determine sha256 hash of the current replica binary
 ///
 /// Returns tuple (path of the replica binary, hex encoded sha256 of binary)
-fn get_replica_binary_hash() -> std::result::Result<(PathBuf, String), String> {
+fn get_replica_binary_hash() -> Result<(PathBuf, String), String> {
     let mut hasher = Sha256::new();
     let replica_binary_path = env::current_exe()
         .map_err(|e| format!("Failed to determine replica binary path: {:?}", e))?;
 
-    let mut binary_file = std::fs::File::open(&replica_binary_path)
+    let mut binary_file = fs::File::open(&replica_binary_path)
         .map_err(|e| format!("Failed to open replica binary to calculate hash: {:?}", e))?;
 
-    std::io::copy(&mut binary_file, &mut hasher)
+    io::copy(&mut binary_file, &mut hasher)
         .map_err(|e| format!("Failed to calculate hash for replica binary: {:?}", e))?;
 
     Ok((replica_binary_path, hex::encode(hasher.finish())))
@@ -60,10 +57,8 @@ fn get_replica_binary_hash() -> std::result::Result<(PathBuf, String), String> {
 fn main() -> io::Result<()> {
     // We do not support 32 bits architectures and probably never will.
     assert_eq_size!(usize, u64);
-
     // Ensure that the hardcoded constant matches the OS page size.
     assert_eq!(ic_sys::sysconf_page_size(), PAGE_SIZE);
-
     // At this point we need to setup a new process group. This is
     // done to ensure all our children processes belong to the same
     // process group (as policy wise in production we restrict setpgid
@@ -75,7 +70,7 @@ fn main() -> io::Result<()> {
         eprintln!("Failed to setup a new process group for replica.");
         // This is a generic exit error. At this point sandboxing is
         // not turned on so we can do a simple exit with cleanup.
-        return Err(std::io::Error::new(std::io::ErrorKind::Other, err));
+        return Err(io::Error::new(io::ErrorKind::Other, err));
     }
 
     #[cfg(feature = "profiler")]
@@ -139,6 +134,11 @@ fn main() -> io::Result<()> {
         e.print().expect("Failed to print CLI argument error.");
     }
 
+    // We abort the whole program with a core dump if a single thread panics.
+    // This way we can capture all the context if a critical error
+    // happens.
+    abort_on_panic();
+
     let config_source = setup::get_config_source(&replica_args);
     // Setup temp directory for the configuration.
     let tmpdir = tempfile::Builder::new()
@@ -148,12 +148,14 @@ fn main() -> io::Result<()> {
     let config = Config::load_with_tmpdir(config_source, tmpdir.path().to_path_buf());
 
     let (logger, async_log_guard) = new_replica_logger_from_config(&config.logger);
-
     let metrics_registry = MetricsRegistry::global();
-
     #[cfg(target_os = "linux")]
     metrics_registry.register(jemalloc_metrics::JemallocMetrics::new());
 
+    let cup = setup::get_catch_up_package(&replica_args, &logger)
+        .as_ref()
+        .map(|c| CatchUpPackage::try_from(c).expect("deserializing CUP failed"));
+
     // Set the replica verison and report as metric
     setup::set_replica_version(&replica_args, &logger);
     {
@@ -172,17 +174,13 @@ fn main() -> io::Result<()> {
     }
 
     let (registry, crypto) = setup::setup_crypto_registry(
-        config.clone(),
+        &config,
         rt_main.handle().clone(),
-        Some(&metrics_registry),
+        &metrics_registry,
         logger.clone(),
     );
 
     let node_id = crypto.get_node_id();
-    let cup_proto = setup::get_catch_up_package(&replica_args, &logger);
-    let cup = cup_proto
-        .as_ref()
-        .map(|c| CatchUpPackage::try_from(c).expect("deserializing CUP failed"));
 
     let subnet_id = match &replica_args {
         Ok(args) => {
@@ -198,23 +196,6 @@ fn main() -> io::Result<()> {
         Err(_) => setup::get_subnet_id(node_id, registry.as_ref(), cup.as_ref(), &logger),
     };
 
-    let subnet_type = setup::get_subnet_type(
-        registry.as_ref(),
-        subnet_id,
-        registry.get_latest_version(),
-        &logger,
-    );
-
-    // Read the root subnet id from registry
-    let root_subnet_id = registry
-        .get_root_subnet_id(
-            cup.as_ref()
-                .map(|c| c.content.registry_version())
-                .unwrap_or_else(|| registry.get_latest_version()),
-        )
-        .expect("cannot read from registry")
-        .expect("cannot find root subnet id");
-
     // Set node_id and subnet_id in the logging context
     let mut context = logger.get_context();
     context.node_id = format!("{}", node_id.get());
@@ -228,13 +209,6 @@ fn main() -> io::Result<()> {
         let _ = REPLICA_BINARY_HASH.set(hash);
     }
 
-    // We abort the whole program with a core dump if a single thread panics.
-    // This way we can capture all the context if a critical error
-    // happens.
-    abort_on_panic();
-
-    setup::create_consensus_pool_dir(&config);
-
     let crypto = Arc::new(crypto);
     let _metrics_endpoint = MetricsHttpEndpoint::new(
         rt_http.handle().clone(),
@@ -245,10 +219,6 @@ fn main() -> io::Result<()> {
         &logger.inner_logger.root,
     );
 
-    let registry_certified_time_reader: Arc<dyn LocalStoreCertifiedTimeReader> = Arc::new(
-        LocalStoreImpl::new(config.registry_client.local_store.clone()),
-    );
-
     info!(logger, "Constructing IC stack");
     let (_, _, _p2p_thread_joiner, _, _xnet_endpoint) =
         ic_replica::setup_ic_stack::construct_ic_stack(
@@ -260,12 +230,9 @@ fn main() -> io::Result<()> {
             config.clone(),
             node_id,
             subnet_id,
-            subnet_type,
-            root_subnet_id,
             registry,
             crypto,
             cup,
-            registry_certified_time_reader,
         )?;
 
     info!(logger, "Constructed IC stack");

rs/replica/src/setup.rs

@@ -11,10 +11,9 @@ use ic_registry_client_helpers::subnet::{SubnetListRegistry, SubnetRegistry};
 use ic_registry_local_store::LocalStoreImpl;
 use ic_registry_subnet_type::SubnetType;
 use ic_types::{
-    consensus::catchup::CatchUpPackage,
-    {NodeId, RegistryVersion, ReplicaVersion, SubnetId},
+    consensus::catchup::CatchUpPackage, NodeId, RegistryVersion, ReplicaVersion, SubnetId,
 };
-use std::{convert::TryFrom, env, path::PathBuf, sync::Arc};
+use std::{env, path::PathBuf, sync::Arc};
 
 /// Parse command-line args into `ReplicaArgs`
 pub fn parse_args() -> Result<ReplicaArgs, clap::Error> {
@@ -129,10 +128,10 @@ pub fn get_subnet_id(
 
 /// Return the subnet type of the given subnet.
 pub fn get_subnet_type(
-    registry: &dyn RegistryClient,
+    logger: &ReplicaLogger,
     subnet_id: SubnetId,
     registry_version: RegistryVersion,
-    logger: &ReplicaLogger,
+    registry: &dyn RegistryClient,
 ) -> SubnetType {
     loop {
         match registry.get_subnet_record(subnet_id, registry_version) {
@@ -181,26 +180,20 @@ pub fn get_config_source(replica_args: &Result<ReplicaArgs, clap::Error>) -> Con
     }
 }
 
-/// Create the consensus pool directory (if none exists)
-pub fn create_consensus_pool_dir(config: &Config) {
-    std::fs::create_dir_all(&config.artifact_pool.consensus_pool_path).unwrap_or_else(|err| {
-        panic!(
-            "Failed to create consensus pool directory {}: {}",
-            config.artifact_pool.consensus_pool_path.display(),
-            err
-        )
-    });
-}
-
 pub fn setup_crypto_registry(
-    config: Config,
+    config: &Config,
     tokio_runtime_handle: tokio::runtime::Handle,
-    metrics_registry: Option<&MetricsRegistry>,
+    metrics_registry: &MetricsRegistry,
     logger: ReplicaLogger,
 ) -> (std::sync::Arc<RegistryClientImpl>, CryptoComponent) {
-    let data_provider = Arc::new(LocalStoreImpl::new(config.registry_client.local_store));
+    let data_provider = Arc::new(LocalStoreImpl::new(
+        config.registry_client.local_store.clone(),
+    ));
 
-    let registry = Arc::new(RegistryClientImpl::new(data_provider, metrics_registry));
+    let registry = Arc::new(RegistryClientImpl::new(
+        data_provider,
+        Some(metrics_registry),
+    ));
 
     // The registry must be initialized before setting up the crypto component
     if let Err(e) = registry.fetch_and_start_polling() {
@@ -282,14 +275,14 @@ pub fn setup_crypto_provider(
     tokio_runtime_handle: tokio::runtime::Handle,
     registry: Arc<dyn RegistryClient>,
     replica_logger: ReplicaLogger,
-    metrics_registry: Option<&MetricsRegistry>,
+    metrics_registry: &MetricsRegistry,
 ) -> CryptoComponent {
     CryptoConfig::check_dir_has_required_permissions(&config.crypto_root).unwrap();
     CryptoComponent::new(
         config,
         Some(tokio_runtime_handle),
         registry,
         replica_logger,
-        metrics_registry,
+        Some(metrics_registry),
     )
 }

rs/replica/src/setup_ic_stack.rs

@@ -1,3 +1,4 @@
+use crate::setup::get_subnet_type;
 use ic_btc_adapter_client::{setup_bitcoin_adapter_clients, BitcoinAdapterClients};
 use ic_btc_consensus::BitcoinPayloadBuilder;
 use ic_config::{artifact_pool::ArtifactPoolConfig, subnet_config::SubnetConfig, Config};
@@ -16,7 +17,8 @@ use ic_logger::{info, ReplicaLogger};
 use ic_messaging::MessageRoutingImpl;
 use ic_metrics::MetricsRegistry;
 use ic_pprof::Pprof;
-use ic_registry_subnet_type::SubnetType;
+use ic_registry_client_helpers::subnet::SubnetRegistry;
+use ic_registry_local_store::LocalStoreImpl;
 use ic_replica_setup_ic_network::{
     create_networking_stack, init_artifact_pools, P2PStateSyncClient,
 };
@@ -27,6 +29,17 @@ use ic_xnet_endpoint::{XNetEndpoint, XNetEndpointConfig};
 use ic_xnet_payload_builder::XNetPayloadBuilderImpl;
 use std::sync::Arc;
 
+/// Create the consensus pool directory (if none exists)
+fn create_consensus_pool_dir(config: &Config) {
+    std::fs::create_dir_all(&config.artifact_pool.consensus_pool_path).unwrap_or_else(|err| {
+        panic!(
+            "Failed to create consensus pool directory {}: {}",
+            config.artifact_pool.consensus_pool_path.display(),
+            err
+        )
+    });
+}
+
 #[allow(clippy::too_many_arguments, clippy::type_complexity)]
 pub fn construct_ic_stack(
     log: &ReplicaLogger,
@@ -37,12 +50,9 @@ pub fn construct_ic_stack(
     config: Config,
     node_id: NodeId,
     subnet_id: SubnetId,
-    subnet_type: SubnetType,
-    root_subnet_id: SubnetId,
     registry: Arc<dyn RegistryClient + Send + Sync>,
     crypto: Arc<CryptoComponent>,
     catch_up_package: Option<CatchUpPackage>,
-    local_store_time_reader: Arc<dyn LocalStoreCertifiedTimeReader>,
 ) -> std::io::Result<(
     // TODO: remove this return value since it is used only in tests
     Arc<StateManagerImpl>,
@@ -54,13 +64,10 @@ pub fn construct_ic_stack(
     XNetEndpoint,
 )> {
     // ---------- ARTIFACT POOLS DEPS FOLLOW ----------
+    create_consensus_pool_dir(&config);
     // Determine the correct catch-up package.
     let catch_up_package = {
         use ic_types::consensus::HasHeight;
-        let make_registry_cup = || {
-            ic_consensus::dkg::make_registry_cup(&*registry, subnet_id, None)
-                .expect("Couldn't create a registry CUP")
-        };
         match catch_up_package {
             // The replica was started on a CUP persisted by the orchestrator.
             Some(cup_from_orc) => {
@@ -79,6 +86,10 @@ pub fn construct_ic_stack(
             // This case is only possible if the replica is started without an orchestrator which
             // is currently only possible in the local development mode with `dfx`.
             None => {
+                let make_registry_cup = || {
+                    ic_consensus::dkg::make_registry_cup(&*registry, subnet_id, None)
+                        .expect("Couldn't create a registry CUP")
+                };
                 let registry_cup = CatchUpPackage::try_from(&make_registry_cup())
                     .expect("deserializing CUP failed");
                 info!(
@@ -90,6 +101,11 @@ pub fn construct_ic_stack(
             }
         }
     };
+    let root_subnet_id = registry
+        .get_root_subnet_id(catch_up_package.content.registry_version())
+        .expect("cannot read from registry")
+        .expect("cannot find root subnet id");
+
     let artifact_pool_config = ArtifactPoolConfig::from(config.artifact_pool.clone());
     let artifact_pools = init_artifact_pools(
         node_id,
@@ -99,8 +115,14 @@ pub fn construct_ic_stack(
         log.clone(),
         catch_up_package,
     );
-
     // ---------- REPLICATED STATE DEPS FOLLOW ----------
+    let subnet_type = get_subnet_type(
+        log,
+        subnet_id,
+        registry.get_latest_version(),
+        registry.as_ref(),
+    );
+
     let consensus_pool_cache = artifact_pools.consensus_pool.read().unwrap().get_cache();
     let verifier = Arc::new(VerifierImpl::new(crypto.clone()));
     let state_manager = Arc::new(StateManagerImpl::new(
@@ -219,6 +241,9 @@ pub fn construct_ic_stack(
     // ---------- CONSENSUS AND P2P DEPS FOLLOW ----------
     let state_sync = StateSync::new(state_manager.clone(), log.clone());
     let sev_handshake = Arc::new(Sev::new(node_id, registry.clone()));
+    let local_store_cert_time_reader: Arc<dyn LocalStoreCertifiedTimeReader> = Arc::new(
+        LocalStoreImpl::new(config.registry_client.local_store.clone()),
+    );
     let (ingress_ingestion_service, p2p_runner) = create_networking_stack(
         metrics_registry,
         log.clone(),
@@ -245,7 +270,7 @@ pub fn construct_ic_stack(
         execution_services.ingress_history_reader,
         artifact_pools,
         cycles_account_manager,
-        local_store_time_reader,
+        local_store_cert_time_reader,
         canister_http_adapter_client,
         config.nns_registry_replicator.poll_delay_duration_ms,
     );