diff --git a/rs/messaging/src/message_routing.rs b/rs/messaging/src/message_routing.rs index f3ca2c383dc6..3db6082cb9fa 100644 --- a/rs/messaging/src/message_routing.rs +++ b/rs/messaging/src/message_routing.rs @@ -55,6 +55,7 @@ use prometheus::{ Gauge, Histogram, HistogramTimer, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, }; +use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::convert::{AsRef, TryFrom}; use std::net::{Ipv4Addr, Ipv6Addr}; @@ -582,14 +583,10 @@ trait BatchProcessor: Send { } /// Implementation of [`BatchProcessor`]. -struct BatchProcessorImpl -where - RegistryClient_: RegistryClient, -{ +struct BatchProcessorImpl { state_manager: Arc>, state_machine: Box, - registry: Arc, - bitcoin_config: BitcoinConfig, + registry_reader: RegistryReader, metrics: MessageRoutingMetrics, log: ReplicaLogger, #[allow(dead_code)] @@ -704,18 +701,68 @@ impl BatchProcessorImpl { metrics.clone(), )); + let registry_reader = RegistryReader::new( + registry, + hypervisor_config.bitcoin, + metrics.clone(), + log.clone(), + ); + Self { state_manager, state_machine, - registry, - bitcoin_config: hypervisor_config.bitcoin, + registry_reader, metrics, log, malicious_flags, } } +} + +/// The registry-derived values read at a given registry version, cached by +/// [`RegistryReader`] so that repeated reads at the same version are cheap. +struct CachedRegistryData { + registry_version: RegistryVersion, + network_topology: Arc, + own_subnet_info: Arc, + registry_execution_settings: Arc, +} + +/// Reads the registry contents required by `BatchProcessorImpl::process_batch()`. +/// +/// Caches the values read at the most recent registry version, so that a +/// subsequent read at the same version returns `Arc` clones of the cached +/// values instead of reading the registry again. +struct RegistryReader { + registry: Arc, + bitcoin_config: BitcoinConfig, + /// Cache holding the values read at the most recent registry version. + cache: RefCell>, + metrics: MessageRoutingMetrics, + log: ReplicaLogger, +} + +impl RegistryReader { + fn new( + registry: Arc, + bitcoin_config: BitcoinConfig, + metrics: MessageRoutingMetrics, + log: ReplicaLogger, + ) -> Self { + Self { + registry, + bitcoin_config, + cache: RefCell::new(None), + metrics, + log, + } + } /// Reads registry contents required by `BatchProcessorImpl::process_batch()`. + /// + /// Returns `Arc` clones of the cached values if they were previously read at + /// the same `registry_version`; otherwise reads the registry, caches the + /// results and returns them. // /// # Warning /// If the registry is unavailable, this method loops until it becomes @@ -724,10 +771,25 @@ impl BatchProcessorImpl { &self, registry_version: RegistryVersion, own_subnet_id: SubnetId, - ) -> (NetworkTopology, OwnSubnetInfo, RegistryExecutionSettings) { - loop { + ) -> ( + Arc, + Arc, + Arc, + ) { + // Return the cached values if they were read at the requested version. + if let Some(cached) = self.cache.borrow().as_ref() + && cached.registry_version == registry_version + { + return ( + Arc::clone(&cached.network_topology), + Arc::clone(&cached.own_subnet_info), + Arc::clone(&cached.registry_execution_settings), + ); + } + + let (network_topology, own_subnet_info, registry_execution_settings) = loop { match self.try_to_read_registry(registry_version, own_subnet_id) { - Ok(result) => return result, + Ok(result) => break result, Err(ReadRegistryError::Persistent(error_message)) => { // Increment the critical error counter in case of a persistent error. self.metrics.critical_error_failed_to_read_registry.inc(); @@ -749,11 +811,28 @@ impl BatchProcessorImpl { } } sleep(std::time::Duration::from_millis(100)); - } + }; + + let network_topology = Arc::new(network_topology); + let own_subnet_info = Arc::new(own_subnet_info); + let registry_execution_settings = Arc::new(registry_execution_settings); + + *self.cache.borrow_mut() = Some(CachedRegistryData { + registry_version, + network_topology: Arc::clone(&network_topology), + own_subnet_info: Arc::clone(&own_subnet_info), + registry_execution_settings: Arc::clone(®istry_execution_settings), + }); + + ( + network_topology, + own_subnet_info, + registry_execution_settings, + ) } - /// Loads the `NetworkTopology`, `SubnetFeatures`, execution settings and - /// own subnet node public keys from the registry. + /// Loads the `NetworkTopology`, `OwnSubnetInfo` and execution settings from the + /// registry. /// /// All of the above are required for deterministic processing, so if any /// entry is missing or cannot be decoded; or reading the registry fails; the @@ -1373,11 +1452,10 @@ impl BatchProcessor for BatchProcessorImpl, + own_subnet_info: Arc, registry_settings: &RegistryExecutionSettings, ) -> ReplicatedState { - state.metadata.network_topology = Arc::new(network_topology); - state.metadata.own_subnet_info = Arc::new(own_subnet_info); + state.metadata.network_topology = network_topology; + state.metadata.own_subnet_info = own_subnet_info; state.put_canister_state( CanisterStateBuilder::new() .with_canister_id(canister_test_id(1)) @@ -668,8 +668,12 @@ fn make_batch_processor( let batch_processor = BatchProcessorImpl { state_manager: state_manager.clone(), state_machine: Box::new(FakeStateMachine(registry_settings.clone())), - registry, - bitcoin_config: BitcoinConfig::default(), + registry_reader: RegistryReader::new( + registry, + BitcoinConfig::default(), + metrics.clone(), + log.clone(), + ), metrics: metrics.clone(), log, malicious_flags: MaliciousFlags::default(), @@ -685,7 +689,9 @@ fn try_to_read_registry( own_subnet_id: SubnetId, ) -> Result<(NetworkTopology, OwnSubnetInfo, RegistryExecutionSettings), ReadRegistryError> { let (batch_processor, _, _, _) = make_batch_processor(registry.clone(), log); - batch_processor.try_to_read_registry(registry.get_latest_version(), own_subnet_id) + batch_processor + .registry_reader + .try_to_read_registry(registry.get_latest_version(), own_subnet_id) } /// Tests that `BatchProcessorImpl::try_to_read_registry()` returns `Ok(_)`; and checks that the @@ -873,6 +879,7 @@ fn try_read_registry_succeeds_with_fully_specified_registry_records() { let (batch_processor, metrics, state_manager, registry_settings) = make_batch_processor(fixture.registry.clone(), log); let (network_topology, own_subnet_info, registry_execution_settings) = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id) .unwrap(); let OwnSubnetInfo { @@ -1095,6 +1102,7 @@ fn try_read_registry_succeeds_with_minimal_registry_records() { let (batch_processor, metrics, _, _) = make_batch_processor(fixture.registry.clone(), log.clone()); let result = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id); assert_matches!(result, Ok(_)); @@ -1156,6 +1164,81 @@ fn try_read_registry_succeeds_with_minimal_registry_records() { }); } +/// Tests that `RegistryReader::read_registry()` caches the values read at the most +/// recent registry version: a second read at the same version returns `Arc` clones +/// of the cached values, while a read at a different version reads the registry anew. +#[test] +fn read_registry_caches_values_by_registry_version() { + with_test_replica_logger(|log| { + use Integrity::*; + + let own_subnet_id = subnet_test_id(13); + let own_subnet_record = SubnetRecord { + max_number_of_canisters: 784, + ..Default::default() + }; + let own_transcript = dummy_transcript_for_tests(); + let nns_subnet_id = subnet_test_id(42); + + let input = TestRecords { + subnet_ids: Valid([own_subnet_id]), + subnet_records: [Valid(&own_subnet_record)], + ni_dkg_transcripts: [Valid(Some(&own_transcript))], + nns_subnet_id: Valid(nns_subnet_id), + chain_key_enabled_subnets: &BTreeMap::default(), + provisional_whitelist: Missing, + routing_table: Missing, + canister_migrations: Missing, + node_public_keys: &BTreeMap::default(), + api_boundary_node_records: &BTreeMap::default(), + node_records: &BTreeMap::default(), + }; + + let fixture = RegistryFixture::new(); + fixture.write_test_records(&input).unwrap(); + let version_1 = fixture.registry.get_latest_version(); + + let (batch_processor, _, _, _) = + make_batch_processor(fixture.registry.clone(), log.clone()); + + // Two reads at the same registry version return `Arc` clones of the same + // (cached) values. + let (nt_1, osi_1, res_1) = batch_processor + .registry_reader + .read_registry(version_1, own_subnet_id); + let (nt_2, osi_2, res_2) = batch_processor + .registry_reader + .read_registry(version_1, own_subnet_id); + assert!(Arc::ptr_eq(&nt_1, &nt_2)); + assert!(Arc::ptr_eq(&osi_1, &osi_2)); + assert!(Arc::ptr_eq(&res_1, &res_2)); + + // Writing the same records again bumps the registry to a new version. + fixture.write_test_records(&input).unwrap(); + let version_2 = fixture.registry.get_latest_version(); + assert_ne!(version_1, version_2); + + // A read at a different version bypasses the cache and returns freshly read + // values: distinct `Arc` instances, but equal in content. + let (nt_3, osi_3, res_3) = batch_processor + .registry_reader + .read_registry(version_2, own_subnet_id); + assert!(!Arc::ptr_eq(&nt_1, &nt_3)); + assert!(!Arc::ptr_eq(&osi_1, &osi_3)); + assert!(!Arc::ptr_eq(&res_1, &res_3)); + assert_eq!(nt_1, nt_3); + assert_eq!(osi_1, osi_3); + + // The freshly read values are now the cached ones. + let (nt_4, osi_4, res_4) = batch_processor + .registry_reader + .read_registry(version_2, own_subnet_id); + assert!(Arc::ptr_eq(&nt_3, &nt_4)); + assert!(Arc::ptr_eq(&osi_3, &osi_4)); + assert!(Arc::ptr_eq(&res_3, &res_4)); + }); +} + /// Tests that `BatchProcessorImpl::try_to_read_registry()` returns `Err(_)` if records in the /// registry hold corrupted data. Corrupted data is simulated by writing a string of prime numbers /// (almost anything would do) rather than an actual struct into registry. @@ -1416,6 +1499,7 @@ fn try_read_registry_can_skip_missing_or_invalid_node_public_keys() { let (batch_processor, metrics, _, _) = make_batch_processor(fixture.registry.clone(), log.clone()); let res = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id); assert_matches!(res, Ok(_)); @@ -1562,6 +1646,7 @@ fn try_read_registry_can_skip_missing_or_invalid_fields_of_api_boundary_nodes() let (batch_processor, metrics, _, _) = make_batch_processor(fixture.registry.clone(), log.clone()); let res = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id); assert_matches!(res, Ok(_)); @@ -1639,6 +1724,7 @@ fn try_read_registry_succeeds_and_populates_subnet_admins() { let (batch_processor, _, _, _) = make_batch_processor(fixture.registry.clone(), log.clone()); let network_topology = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id) .unwrap() .0; @@ -1718,6 +1804,7 @@ fn try_read_registry_succeeds_and_resets_subnet_admins() { let (batch_processor, metrics, _, _) = make_batch_processor(fixture.registry.clone(), log.clone()); let network_topology = batch_processor + .registry_reader .try_to_read_registry(fixture.registry.get_latest_version(), own_subnet_id) .unwrap() .0; @@ -2091,13 +2178,17 @@ fn check_critical_error_counter_is_not_incremented_for_transient_error() { // Try reading the registry at the next version; should return `Err(_)`. assert_matches!( - batch_processor.try_to_read_registry(next_registry_version, own_subnet_id), + batch_processor + .registry_reader + .try_to_read_registry(next_registry_version, own_subnet_id), Err(ReadRegistryError::Transient(_)) ); // Write minimal records to the registry, reading the registry should now work. fixture.write_test_records(&minimal_input).unwrap(); assert_matches!( - batch_processor.try_to_read_registry(next_registry_version, own_subnet_id), + batch_processor + .registry_reader + .try_to_read_registry(next_registry_version, own_subnet_id), Ok(_) ); @@ -2108,7 +2199,9 @@ fn check_critical_error_counter_is_not_incremented_for_transient_error() { // Spawn a thread trying to read from the registry at the next version; this will fail // until we update the registry. let handle = std::thread::spawn(move || { - batch_processor.read_registry(next_registry_version, own_subnet_id) + batch_processor + .registry_reader + .read_registry(next_registry_version, own_subnet_id) }); // Wait 150ms, then update the registry and join the thread above. std::thread::sleep(Duration::from_millis(150)); @@ -2146,11 +2239,15 @@ fn reading_mainnet_registry_succeeds() { let (batch_processor, _, _, _) = make_batch_processor(registry, log); assert_matches!( - batch_processor.try_to_read_registry(registry_version, mainnet_nns_subnet()), + batch_processor + .registry_reader + .try_to_read_registry(registry_version, mainnet_nns_subnet()), Ok(_) ); assert_matches!( - batch_processor.try_to_read_registry(registry_version, mainnet_app_subnet()), + batch_processor + .registry_reader + .try_to_read_registry(registry_version, mainnet_app_subnet()), Ok(_) ); }); diff --git a/rs/messaging/src/state_machine.rs b/rs/messaging/src/state_machine.rs index 047c1b7086e6..d49c68d4dca3 100644 --- a/rs/messaging/src/state_machine.rs +++ b/rs/messaging/src/state_machine.rs @@ -30,8 +30,8 @@ pub(crate) trait StateMachine: Send { &self, state: ReplicatedState, batch: Batch, - network_topology: NetworkTopology, - own_subnet_info: OwnSubnetInfo, + network_topology: Arc, + own_subnet_info: Arc, registry_settings: &RegistryExecutionSettings, ) -> ReplicatedState; } @@ -101,8 +101,8 @@ impl StateMachine for StateMachineImpl { &self, mut state: ReplicatedState, batch: Batch, - network_topology: NetworkTopology, - own_subnet_info: OwnSubnetInfo, + network_topology: Arc, + own_subnet_info: Arc, registry_settings: &RegistryExecutionSettings, ) -> ReplicatedState { let time_out_messages_timer = self.metrics.start_phase_timer(PHASE_TIME_OUT_MESSAGES); @@ -119,8 +119,8 @@ impl StateMachine for StateMachineImpl { ) } - state.metadata.network_topology = Arc::new(network_topology); - state.metadata.own_subnet_info = Arc::new(own_subnet_info); + state.metadata.network_topology = network_topology; + state.metadata.own_subnet_info = own_subnet_info; if let Err(message) = state.metadata.init_allocation_ranges_if_empty() { self.metrics .observe_no_canister_allocation_range(&self.log, message); diff --git a/rs/messaging/src/state_machine/tests.rs b/rs/messaging/src/state_machine/tests.rs index e71f059e8081..43f5c294b8c6 100644 --- a/rs/messaging/src/state_machine/tests.rs +++ b/rs/messaging/src/state_machine/tests.rs @@ -63,7 +63,7 @@ struct StateMachineTestFixture { demux: Box, stream_builder: Box, initial_state: ReplicatedState, - network_topology: NetworkTopology, + network_topology: Arc, metrics: MessageRoutingMetrics, metrics_registry: MetricsRegistry, } @@ -151,7 +151,7 @@ fn test_fixture(provided_batch: &Batch) -> StateMachineTestFixture { demux, stream_builder, initial_state, - network_topology, + network_topology: Arc::new(network_topology), metrics, metrics_registry, } @@ -181,8 +181,8 @@ fn state_machine_populates_network_topology() { )); assert_ne!( - fixture.initial_state.metadata.network_topology.as_ref(), - &fixture.network_topology + fixture.initial_state.metadata.network_topology, + fixture.network_topology ); let state = state_machine.execute_round( @@ -193,10 +193,7 @@ fn state_machine_populates_network_topology() { &test_registry_settings(), ); - assert_eq!( - state.metadata.network_topology.as_ref(), - &fixture.network_topology - ); + assert_eq!(state.metadata.network_topology, fixture.network_topology); }); } @@ -649,7 +646,7 @@ fn state_machine_handles_messages_to_deleted_subnet() { let mut state = state_machine.execute_round( initial_state, provided_batch, - network_topology, + Arc::new(network_topology), Default::default(), &test_registry_settings(), ); @@ -800,7 +797,7 @@ fn split_fixture() -> StateMachineTestFixture { demux, stream_builder, initial_state, - network_topology, + network_topology: Arc::new(network_topology), metrics, metrics_registry, }