diff --git a/rs/config/src/execution_environment.rs b/rs/config/src/execution_environment.rs index 986fb1180f00..4b56889be0c9 100644 --- a/rs/config/src/execution_environment.rs +++ b/rs/config/src/execution_environment.rs @@ -46,21 +46,6 @@ const SUBNET_MEMORY_THRESHOLD: NumBytes = NumBytes::new(750 * GIB); /// IC protocol requires storing copies of the canister state. const SUBNET_MEMORY_CAPACITY: NumBytes = NumBytes::new(2 * TIB); -/// This is the upper limit on how much memory can be used by all guaranteed -/// response canister messages on a given subnet. -/// -/// Guaranteed response message memory usage is calculated as the total size of -/// enqueued guaranteed responses; plus the maximum allowed response size per -/// reserved guaranteed response slot. -const SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY: NumBytes = NumBytes::new(15 * GIB); - -/// The limit on how much memory may be used by all guaranteed response messages -/// on a given subnet at the end of a round. -/// -/// During the round, the best-effort message memory usage may exceed the limit, -/// but the constraint is restored at the end of the round by shedding messages. -const SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY: NumBytes = NumBytes::new(5 * GIB); - /// This is the upper limit on how much memory can be used by the ingress /// history on a given subnet. It is lower than the subnet message memory /// capacity because here we count actual memory consumption as opposed to @@ -110,7 +95,7 @@ pub const STOP_CANISTER_TIMEOUT_DURATION: Duration = Duration::from_secs(5 * 60) /// potential fragmentation. This limit should be larger than the maximum /// canister memory size to guarantee that a message that overwrites the whole /// memory can succeed. -pub(crate) const SUBNET_HEAP_DELTA_CAPACITY: NumBytes = NumBytes::new(140 * GIB); +pub const SUBNET_HEAP_DELTA_CAPACITY: NumBytes = NumBytes::new(140 * GIB); /// The maximum number of instructions for inspect_message calls. const MAX_INSTRUCTIONS_FOR_MESSAGE_ACCEPTANCE_CALLS: NumInstructions = @@ -244,14 +229,6 @@ pub struct Config { /// the subnet. pub subnet_memory_capacity: NumBytes, - /// The maximum amount of logical storage available to guaranteed response - /// canister messages across the whole subnet. - pub guaranteed_response_message_memory_capacity: NumBytes, - - /// The maximum amount of logical storage available to best-effort canister - /// messages across the whole subnet. - pub best_effort_message_memory_capacity: NumBytes, - /// The maximum amount of logical storage available to the ingress history /// across the whole subnet. pub ingress_history_memory_capacity: NumBytes, @@ -413,9 +390,6 @@ impl Default for Config { MAX_INSTRUCTIONS_FOR_MESSAGE_ACCEPTANCE_CALLS, subnet_memory_threshold: SUBNET_MEMORY_THRESHOLD, subnet_memory_capacity: SUBNET_MEMORY_CAPACITY, - guaranteed_response_message_memory_capacity: - SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY, - best_effort_message_memory_capacity: SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY, ingress_history_memory_capacity: INGRESS_HISTORY_MEMORY_CAPACITY, subnet_wasm_custom_sections_memory_capacity: SUBNET_WASM_CUSTOM_SECTIONS_MEMORY_CAPACITY, diff --git a/rs/execution_environment/src/execution_environment.rs b/rs/execution_environment/src/execution_environment.rs index ff1adee2dede..6560105363db 100644 --- a/rs/execution_environment/src/execution_environment.rs +++ b/rs/execution_environment/src/execution_environment.rs @@ -532,8 +532,9 @@ impl ExecutionEnvironment { self.subnet_memory_capacity(state.resource_limits()).get() as i64 - self.config.subnet_memory_reservation.get() as i64 - memory_taken.execution().get() as i64, - self.config - .guaranteed_response_message_memory_capacity + state + .metadata + .guaranteed_response_message_memory_capacity() .get() as i64 - memory_taken.guaranteed_response_messages().get() as i64, self.config @@ -572,8 +573,9 @@ impl ExecutionEnvironment { &self, state: &ReplicatedState, ) -> i64 { - self.config - .guaranteed_response_message_memory_capacity + state + .metadata + .guaranteed_response_message_memory_capacity() .get() as i64 - state.guaranteed_response_message_memory_taken().get() as i64 } @@ -3469,8 +3471,7 @@ impl ExecutionEnvironment { // Letting the canister grow arbitrarily when executing the // query is fine as we do not persist state modifications. - let subnet_available_memory = - full_subnet_memory_capacity(&self.config, state.resource_limits()); + let subnet_available_memory = full_subnet_memory_capacity(&self.config, &state); let execution_parameters = self.execution_parameters( canister_state, instruction_limits, @@ -4855,13 +4856,17 @@ impl CompilationCostHandling { /// Returns the subnet's configured memory capacity (ignoring current usage). pub(crate) fn full_subnet_memory_capacity( config: &ExecutionConfig, - resource_limits: ResourceLimits, + state: &ReplicatedState, ) -> SubnetAvailableMemory { SubnetAvailableMemory::new_scaled( - resource_limits + state + .resource_limits() .maximum_state_size_or(config.subnet_memory_capacity) .get() as i64, - config.guaranteed_response_message_memory_capacity.get() as i64, + state + .metadata + .guaranteed_response_message_memory_capacity() + .get() as i64, config.subnet_wasm_custom_sections_memory_capacity.get() as i64, NonZeroU64::new(1).expect("scaling_factor must be non zero"), ) diff --git a/rs/execution_environment/src/query_handler.rs b/rs/execution_environment/src/query_handler.rs index 7ee0a27feab5..26fb47cb3800 100644 --- a/rs/execution_environment/src/query_handler.rs +++ b/rs/execution_environment/src/query_handler.rs @@ -329,8 +329,7 @@ impl InternalHttpQueryHandler { // Letting the canister grow arbitrarily when executing the // query is fine as we do not persist state modifications. - let subnet_available_memory = - full_subnet_memory_capacity(&self.config, state.get_ref().resource_limits()); + let subnet_available_memory = full_subnet_memory_capacity(&self.config, state.get_ref()); // Letting the canister use the full subnet memory reservation // is fine as we do not persist state modifications. let subnet_memory_reservation = self.config.subnet_memory_reservation; diff --git a/rs/execution_environment/src/scheduler/test_utilities.rs b/rs/execution_environment/src/scheduler/test_utilities.rs index 9d3af0028654..6c7064c929fe 100644 --- a/rs/execution_environment/src/scheduler/test_utilities.rs +++ b/rs/execution_environment/src/scheduler/test_utilities.rs @@ -40,7 +40,9 @@ use ic_replicated_state::{ CanisterState, ExecutionState, ExportedFunctions, InputQueueType, Memory, NumWasmPages, OutputRequest, ReplicatedState, canister_state::execution_state::{self, WasmExecutionMode, WasmMetadata}, - metadata_state::testing::{NetworkTopologyTesting, SystemMetadataTesting}, + metadata_state::testing::{ + NetworkTopologyTesting, SystemMetadataTesting, heap_delta_capacity_for_message_memory, + }, metrics::ReplicatedStateMetrics, num_bytes_try_from, page_map::TestPageAllocatorFileDescriptorImpl, @@ -813,7 +815,7 @@ pub(crate) struct SchedulerTestBuilder { hypervisor_config: HypervisorConfig, initial_canister_cycles: Cycles, subnet_memory_capacity: u64, - subnet_guaranteed_response_message_memory: u64, + maximum_state_delta: Option, subnet_callback_soft_limit: usize, canister_guaranteed_callback_quota: usize, registry_settings: RegistryExecutionSettings, @@ -845,9 +847,7 @@ impl Default for SchedulerTestBuilder { hypervisor_config, initial_canister_cycles: Cycles::new(1_000_000_000_000_000_000), subnet_memory_capacity: config.subnet_memory_capacity.get(), - subnet_guaranteed_response_message_memory: config - .guaranteed_response_message_memory_capacity - .get(), + maximum_state_delta: None, subnet_callback_soft_limit: config.subnet_callback_soft_limit, canister_guaranteed_callback_quota: config.canister_guaranteed_callback_quota, registry_settings: test_registry_settings(), @@ -891,7 +891,9 @@ impl SchedulerTestBuilder { subnet_guaranteed_response_message_memory: u64, ) -> Self { Self { - subnet_guaranteed_response_message_memory, + maximum_state_delta: Some(heap_delta_capacity_for_message_memory(NumBytes::new( + subnet_guaranteed_response_message_memory, + ))), ..self } } @@ -981,6 +983,10 @@ impl SchedulerTestBuilder { }).unwrap(); let mut state = ReplicatedState::new(self.own_subnet_id, self.subnet_type); + { + let own_subnet_info = std::sync::Arc::make_mut(&mut state.metadata.own_subnet_info); + own_subnet_info.resource_limits.maximum_state_delta = self.maximum_state_delta; + } let mut registry_settings = self.registry_settings; @@ -1060,9 +1066,6 @@ impl SchedulerTestBuilder { let config = ic_config::execution_environment::Config { allocatable_compute_capacity_in_percent: self.allocatable_compute_capacity_in_percent, subnet_memory_capacity: NumBytes::new(self.subnet_memory_capacity), - guaranteed_response_message_memory_capacity: NumBytes::new( - self.subnet_guaranteed_response_message_memory, - ), // Keep it simple, no memory reservation for responses. subnet_memory_reservation: NumBytes::new(0), subnet_callback_soft_limit: self.subnet_callback_soft_limit, diff --git a/rs/messaging/BUILD.bazel b/rs/messaging/BUILD.bazel index 2d817da1b494..991ac46c7bcb 100644 --- a/rs/messaging/BUILD.bazel +++ b/rs/messaging/BUILD.bazel @@ -110,6 +110,7 @@ rust_ic_test_suite_with_extra_srcs( "//rs/protobuf", "//rs/registry/keys", "//rs/registry/proto_data_provider", + "//rs/registry/resource_limits", "//rs/registry/routing_table", "//rs/registry/subnet_type", "//rs/replicated_state", diff --git a/rs/messaging/src/message_routing.rs b/rs/messaging/src/message_routing.rs index 3db6082cb9fa..0cf04d761c4d 100644 --- a/rs/messaging/src/message_routing.rs +++ b/rs/messaging/src/message_routing.rs @@ -663,7 +663,6 @@ impl BatchProcessorImpl { ))); let stream_handler = Box::new(routing::stream_handler::StreamHandlerImpl::new( subnet_id, - hypervisor_config.clone(), metrics_registry, &metrics, Arc::clone(&time_in_stream_metrics), @@ -696,7 +695,6 @@ impl BatchProcessorImpl { scheduler, demux, stream_builder, - hypervisor_config.clone(), log.clone(), metrics.clone(), )); diff --git a/rs/messaging/src/routing/stream_handler.rs b/rs/messaging/src/routing/stream_handler.rs index e1a20554c5d9..30843555061b 100644 --- a/rs/messaging/src/routing/stream_handler.rs +++ b/rs/messaging/src/routing/stream_handler.rs @@ -2,8 +2,6 @@ use crate::message_routing::{ CRITICAL_ERROR_ILLEGAL_ENGINE_MESSAGE, CRITICAL_ERROR_INDUCT_RESPONSE_FAILED, LatencyMetrics, MessageRoutingMetrics, }; -use ic_base_types::NumBytes; -use ic_config::execution_environment::Config as HypervisorConfig; use ic_error_types::RejectCode; use ic_interfaces::messaging::{ LABEL_VALUE_CANISTER_METHOD_NOT_FOUND, LABEL_VALUE_CANISTER_NOT_FOUND, @@ -202,9 +200,6 @@ pub(crate) trait StreamHandler: Send { pub(crate) struct StreamHandlerImpl { subnet_id: SubnetId, - /// The memory allocated for guaranteed response messages on the subnet. - guaranteed_response_message_memory_capacity: NumBytes, - metrics: StreamHandlerMetrics, /// Per-destination-subnet histogram of wall time spent by messages in the /// stream before they are garbage collected. @@ -219,7 +214,6 @@ pub(crate) struct StreamHandlerImpl { impl StreamHandlerImpl { pub(crate) fn new( subnet_id: SubnetId, - hypervisor_config: HypervisorConfig, metrics_registry: &MetricsRegistry, message_routing_metrics: &MessageRoutingMetrics, time_in_stream_metrics: Arc>, @@ -227,8 +221,6 @@ impl StreamHandlerImpl { ) -> Self { Self { subnet_id, - guaranteed_response_message_memory_capacity: hypervisor_config - .guaranteed_response_message_memory_capacity, metrics: StreamHandlerMetrics::new(metrics_registry, message_routing_metrics), time_in_stream_metrics, time_in_backlog_metrics: RefCell::new(LatencyMetrics::new_time_in_backlog( @@ -1258,7 +1250,10 @@ impl StreamHandlerImpl { /// difference between the subnet's guaranteed response message memory capacity /// and its current usage. fn available_guaranteed_response_memory(&self, state: &ReplicatedState) -> i64 { - self.guaranteed_response_message_memory_capacity.get() as i64 + state + .metadata + .guaranteed_response_message_memory_capacity() + .get() as i64 - state.guaranteed_response_message_memory_taken().get() as i64 } diff --git a/rs/messaging/src/routing/stream_handler/tests.rs b/rs/messaging/src/routing/stream_handler/tests.rs index 9094aad6a723..b94256be1b17 100644 --- a/rs/messaging/src/routing/stream_handler/tests.rs +++ b/rs/messaging/src/routing/stream_handler/tests.rs @@ -2,13 +2,13 @@ use super::*; use crate::message_routing::{LABEL_REMOTE, METRIC_TIME_IN_BACKLOG, METRIC_TIME_IN_STREAM}; use MessageBuilder::*; use assert_matches::assert_matches; -use ic_base_types::NumSeconds; +use ic_base_types::{NumBytes, NumSeconds}; use ic_certification_version::{CURRENT_CERTIFICATION_VERSION, CertificationVersion}; -use ic_config::execution_environment::Config as HypervisorConfig; use ic_interfaces::messaging::LABEL_VALUE_CANISTER_NOT_FOUND; use ic_metrics::MetricsRegistry; use ic_registry_routing_table::{CanisterIdRange, CanisterIdRanges, RoutingTable}; use ic_registry_subnet_type::SubnetType; +use ic_replicated_state::metadata_state::testing::heap_delta_capacity_for_message_memory; use ic_replicated_state::{ CanisterStatus, ReplicatedState, Stream, SubnetTopology, metadata_state::{ @@ -377,78 +377,32 @@ fn induct_loopback_stream_success() { /// `StreamHandlerImpl::induct_loopback_stream()`. #[test] fn induct_loopback_stream_with_subnet_message_memory_limit() { - // A stream handler with a subnet message memory limit that only allows up to 3 reservations. - induct_loopback_stream_with_memory_limit_impl(HypervisorConfig { - guaranteed_response_message_memory_capacity: NumBytes::new( - MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2, - ), - ..Default::default() - }); -} - -/// Tests that wasm custom sections memory capacity does not affect -/// `StreamHandlerImpl::induct_loopback_stream()`. -#[test] -fn induct_loopback_stream_with_zero_subnet_wasm_custom_sections_limit() { - // A stream handler with a subnet message memory limit that only allows up to 3 reservations - // and no allowance for wasm custom sections. - induct_loopback_stream_with_memory_limit_impl(HypervisorConfig { - guaranteed_response_message_memory_capacity: NumBytes::new( - MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2, - ), - subnet_wasm_custom_sections_memory_capacity: NumBytes::new(0), - ..Default::default() - }); -} - -/// Tests that subnet memory limit is ignored by -/// `StreamHandlerImpl::induct_loopback_stream()` for system subnets. -#[test] -fn system_subnet_induct_loopback_stream_ignores_subnet_memory_limit() { - // A stream handler with a subnet memory limit that only allows up to 3 reservations. - induct_loopback_stream_ignores_memory_limit_impl(HypervisorConfig { - subnet_memory_capacity: NumBytes::new(MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2), - ..Default::default() - }); + // A subnet message memory limit that only allows up to 3 reservations. + induct_loopback_stream_with_memory_limit_impl(Some(heap_delta_capacity_for_message_memory( + NumBytes::new(MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2), + ))); } /// Tests that subnet message memory limit is ignored by /// `StreamHandlerImpl::induct_loopback_stream()` for system subnets. #[test] fn system_subnet_induct_loopback_stream_ignores_subnet_message_memory_limit() { - // A stream handler with a subnet message memory limit that only allows up to 3 reservations. - induct_loopback_stream_ignores_memory_limit_impl(HypervisorConfig { - guaranteed_response_message_memory_capacity: NumBytes::new( - MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2, - ), - ..Default::default() - }); -} - -/// Tests that subnet wasm custom sections memory limit is ignored by -/// `StreamHandlerImpl::induct_loopback_stream()` for system subnets. -#[test] -fn system_subnet_induct_loopback_stream_ignores_subnet_wasm_custom_sections_memory_limit() { - // A stream handler with a subnet message memory limit that only allows up to 3 reservations. - induct_loopback_stream_ignores_memory_limit_impl(HypervisorConfig { - guaranteed_response_message_memory_capacity: NumBytes::new( - MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2, - ), - subnet_wasm_custom_sections_memory_capacity: NumBytes::new(0), - ..Default::default() - }); + // A subnet message memory limit that only allows up to 3 reservations. + induct_loopback_stream_ignores_memory_limit_impl(Some(heap_delta_capacity_for_message_memory( + NumBytes::new(MAX_RESPONSE_COUNT_BYTES as u64 * 7 / 2), + ))); } /// Common initial state setup for `StreamHandlerImpl::induct_loopback_stream()` /// memory limit tests. fn with_induct_loopback_stream_setup( - config: HypervisorConfig, + maximum_state_delta: Option, subnet_type: SubnetType, certification_version: CertificationVersion, test_impl: impl FnOnce(StreamHandlerImpl, ReplicatedState, MetricsFixture), ) { with_local_test_setup_and_config( - config, + maximum_state_delta, subnet_type, certification_version, btreemap![LOCAL_SUBNET => StreamConfig { @@ -474,9 +428,9 @@ fn with_induct_loopback_stream_setup( /// loopback requests and a loopback stream containing said requests. Tries to /// induct the loopback stream and expects the first request to be inducted; and /// the second request to fail to be inducted due to lack of memory. -fn induct_loopback_stream_with_memory_limit_impl(config: HypervisorConfig) { +fn induct_loopback_stream_with_memory_limit_impl(maximum_state_delta: Option) { with_induct_loopback_stream_setup( - config, + maximum_state_delta, SubnetType::Application, CURRENT_CERTIFICATION_VERSION, |stream_handler, state, metrics| { @@ -534,9 +488,9 @@ fn induct_loopback_stream_with_memory_limit_impl(config: HypervisorConfig) { /// loopback requests and a loopback stream containing said requests. Tries to /// induct the loopback stream and expects both requests to be inducted /// successfully. -fn induct_loopback_stream_ignores_memory_limit_impl(config: HypervisorConfig) { +fn induct_loopback_stream_ignores_memory_limit_impl(maximum_state_delta: Option) { with_induct_loopback_stream_setup( - config, + maximum_state_delta, SubnetType::System, CURRENT_CERTIFICATION_VERSION, |stream_handler, state, metrics| { @@ -2630,13 +2584,10 @@ fn induct_stream_slices_with_messages_from_migrating_canister() { /// guaranteed response memory for one request. fn induct_stream_slices_with_memory_limit_impl(subnet_type: SubnetType) { with_test_setup_and_config( - // A config with only enough subnet message memory for one request + epsilon. - HypervisorConfig { - guaranteed_response_message_memory_capacity: NumBytes::new( - MAX_RESPONSE_COUNT_BYTES as u64 * 15 / 10, - ), - ..Default::default() - }, + // A subnet message memory limit with only enough for one request + epsilon. + Some(heap_delta_capacity_for_message_memory(NumBytes::new( + MAX_RESPONSE_COUNT_BYTES as u64 * 15 / 10, + ))), subnet_type, CURRENT_CERTIFICATION_VERSION, // An empty outgoing stream. @@ -3771,7 +3722,7 @@ fn with_test_setup( ), ) { with_test_setup_and_config( - HypervisorConfig::default(), + None, SubnetType::Application, CURRENT_CERTIFICATION_VERSION, stream_configs, @@ -3787,7 +3738,7 @@ fn with_test_setup( /// API been used to arrive at it, i.e. responses and (reject) responses generated from these requests /// can be successfully inducted into the state. Same for the generated stream slices. fn with_test_setup_and_config( - hypervisor_config: HypervisorConfig, + maximum_state_delta: Option, subnet_type: SubnetType, certification_version: CertificationVersion, stream_configs: BTreeMap>>, @@ -3803,10 +3754,12 @@ fn with_test_setup_and_config( // Generate an empty `ReplicatedState` for `LOCAL_SUBNET`. let mut state = ReplicatedState::new(LOCAL_SUBNET, subnet_type); state.metadata.certification_version = certification_version; + let mut own_subnet_info = (*state.metadata.own_subnet_info).clone(); + own_subnet_info.resource_limits.maximum_state_delta = maximum_state_delta; + state.metadata.own_subnet_info = Arc::new(own_subnet_info); let metrics_registry = MetricsRegistry::new(); let stream_handler = StreamHandlerImpl::new( LOCAL_SUBNET, - hypervisor_config, &metrics_registry, &MessageRoutingMetrics::new(&metrics_registry), Arc::new(Mutex::new(LatencyMetrics::new_time_in_stream( @@ -3986,14 +3939,14 @@ fn with_local_test_setup( /// Generates a local test setup, i.e. without incoming stream slices. /// For details see `with_test_setup_and_config()`. fn with_local_test_setup_and_config( - hypervisor_config: HypervisorConfig, + maximum_state_delta: Option, subnet_type: SubnetType, certification_version: CertificationVersion, stream_configs: BTreeMap>>, test_impl: impl FnOnce(StreamHandlerImpl, ReplicatedState, MetricsFixture), ) { with_test_setup_and_config( - hypervisor_config, + maximum_state_delta, subnet_type, certification_version, stream_configs, diff --git a/rs/messaging/src/state_machine.rs b/rs/messaging/src/state_machine.rs index d49c68d4dca3..f7f1025ee1ab 100644 --- a/rs/messaging/src/state_machine.rs +++ b/rs/messaging/src/state_machine.rs @@ -3,7 +3,6 @@ use crate::routing::demux::Demux; use crate::routing::stream_builder::{ StreamBuilder, generate_reject_responses_for_deleted_subnets, }; -use ic_config::execution_environment::Config as HypervisorConfig; use ic_interfaces::execution_environment::{ ExecutionRoundSummary, ExecutionRoundType, RegistryExecutionSettings, Scheduler, }; @@ -12,7 +11,7 @@ use ic_logger::{ReplicaLogger, error, fatal}; use ic_query_stats::deliver_query_stats; use ic_replicated_state::{NetworkTopology, OwnSubnetInfo, ReplicatedState}; use ic_types::batch::{Batch, BatchContent}; -use ic_types::{ExecutionRound, NumBytes, SubnetId}; +use ic_types::{ExecutionRound, SubnetId}; use std::sync::Arc; #[cfg(test)] @@ -39,7 +38,6 @@ pub(crate) struct StateMachineImpl { scheduler: Box>, demux: Box, stream_builder: Box, - best_effort_message_memory_capacity: NumBytes, log: ReplicaLogger, metrics: MessageRoutingMetrics, } @@ -49,7 +47,6 @@ impl StateMachineImpl { scheduler: Box>, demux: Box, stream_builder: Box, - hypervisor_config: HypervisorConfig, log: ReplicaLogger, metrics: MessageRoutingMetrics, ) -> Self { @@ -57,8 +54,6 @@ impl StateMachineImpl { scheduler, demux, stream_builder, - best_effort_message_memory_capacity: hypervisor_config - .best_effort_message_memory_capacity, log, metrics, } @@ -272,10 +267,11 @@ impl StateMachine for StateMachineImpl { // Shed enough messages to stay below the best-effort message memory limit. let shed_messages_timer = self.metrics.start_phase_timer(PHASE_SHED_MESSAGES); - state_after_stream_builder.enforce_best_effort_message_limit( - self.best_effort_message_memory_capacity, - &self.metrics, - ); + let best_effort_message_memory_capacity = state_after_stream_builder + .metadata + .best_effort_message_memory_capacity(); + state_after_stream_builder + .enforce_best_effort_message_limit(best_effort_message_memory_capacity, &self.metrics); #[cfg(debug_assertions)] state_after_stream_builder.assert_balance_with_messages(balance_before_routing); shed_messages_timer.observe_duration(); diff --git a/rs/messaging/src/state_machine/tests.rs b/rs/messaging/src/state_machine/tests.rs index 43f5c294b8c6..264675962906 100644 --- a/rs/messaging/src/state_machine/tests.rs +++ b/rs/messaging/src/state_machine/tests.rs @@ -175,7 +175,6 @@ fn state_machine_populates_network_topology() { fixture.scheduler, fixture.demux, fixture.stream_builder, - Default::default(), log, fixture.metrics, )); @@ -208,7 +207,6 @@ fn test_delivered_batch(provided_batch: Batch) -> ReplicatedState { fixture.scheduler, fixture.demux, fixture.stream_builder, - Default::default(), log, fixture.metrics, )); @@ -638,7 +636,6 @@ fn state_machine_handles_messages_to_deleted_subnet() { scheduler, demux, stream_builder, - Default::default(), log, message_routing_metrics, )); @@ -829,7 +826,6 @@ fn test_online_split(new_subnet_id: SubnetId, other_subnet_id: SubnetId) -> Repl fixture.scheduler, fixture.demux, fixture.stream_builder, - Default::default(), log, fixture.metrics, )); @@ -938,7 +934,6 @@ fn test_batch_time_impl( fixture.scheduler, fixture.demux, fixture.stream_builder, - Default::default(), log, fixture.metrics, ); diff --git a/rs/messaging/tests/common/mod.rs b/rs/messaging/tests/common/mod.rs index 91a36ff439d6..4e49cdc97874 100644 --- a/rs/messaging/tests/common/mod.rs +++ b/rs/messaging/tests/common/mod.rs @@ -3,10 +3,12 @@ use ic_base_types::{CanisterId, PrincipalId}; use ic_config::execution_environment::Config as HypervisorConfig; use ic_config::subnet_config::{CyclesAccountManagerConfig, SchedulerConfig, SubnetConfig}; use ic_management_canister_types_private::CanisterStatusType; +use ic_registry_resource_limits::ResourceLimits; +use ic_replicated_state::metadata_state::testing::heap_delta_capacity_for_message_memory; use ic_replicated_state::testing::CanisterQueuesTesting; use ic_state_machine_tests::{StateMachine, StateMachineConfig, SubmitIngressError, UserError}; use ic_types::{ - SubnetId, + NumBytes, SubnetId, ingress::{IngressState, IngressStatus, WasmResult}, messages::{MessageId, RequestOrResponse}, }; @@ -241,16 +243,14 @@ impl TestSubnetConfig { }, cycles_account_manager_config: CyclesAccountManagerConfig::application_subnet(), }, - HypervisorConfig { - guaranteed_response_message_memory_capacity: self - .guaranteed_response_message_memory_capacity - .into(), - best_effort_message_memory_capacity: self - .best_effort_message_memory_capacity - .into(), - ..HypervisorConfig::default() - }, + HypervisorConfig::default(), ) + .with_resource_limits(ResourceLimits { + maximum_state_size: None, + maximum_state_delta: Some(heap_delta_capacity_for_message_memory(NumBytes::from( + self.guaranteed_response_message_memory_capacity, + ))), + }) } } diff --git a/rs/registry/resource_limits/src/lib.rs b/rs/registry/resource_limits/src/lib.rs index 06baf6537ecf..5995acb433e0 100644 --- a/rs/registry/resource_limits/src/lib.rs +++ b/rs/registry/resource_limits/src/lib.rs @@ -35,6 +35,15 @@ impl ResourceLimits { .filter(|maximum_state_size| maximum_state_size.get() != 0) .unwrap_or(default) } + + /// Returns the subnet heap delta capacity. + /// + /// This is `maximum_state_delta` if not `0`, otherwise the provided `default`. + pub fn maximum_state_delta_or(&self, default: NumBytes) -> NumBytes { + self.maximum_state_delta + .filter(|maximum_state_delta| maximum_state_delta.get() != 0) + .unwrap_or(default) + } } impl From for pb::ResourceLimits { diff --git a/rs/replicated_state/src/metadata_state.rs b/rs/replicated_state/src/metadata_state.rs index 6461f1f81e63..c239d839e85e 100644 --- a/rs/replicated_state/src/metadata_state.rs +++ b/rs/replicated_state/src/metadata_state.rs @@ -551,6 +551,39 @@ impl SubnetMetrics { } } +const GIB: u64 = 1024 * 1024 * 1024; + +/// The upper limit on the total size of all guaranteed response messages on a +/// subnet. +/// +/// Guaranteed response message memory usage is the total size of enqueued +/// guaranteed responses plus the maximum allowed response size per reserved +/// guaranteed response slot. +const SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY: NumBytes = NumBytes::new(15 * GIB); + +/// The limit on the total size of all best-effort messages on a subnet, +/// restored at the end of each round by shedding messages. +const SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY: NumBytes = NumBytes::new(5 * GIB); + +/// Message memory is capped at `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` of the +/// subnet's heap delta capacity. +const MESSAGE_MEMORY_HEAP_DELTA_DIVISOR: u64 = 3; + +/// Caps `configured_default_capacity` at `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` +/// of `heap_delta_capacity`. +/// +/// Message memory is held in RAM alongside the heap delta pages, so this bounds +/// message memory relative to a subnet's heap delta capacity (which is itself +/// sized relative to available RAM). +fn message_memory_capacity( + configured_default_capacity: NumBytes, + heap_delta_capacity: NumBytes, +) -> NumBytes { + configured_default_capacity.min(NumBytes::new( + heap_delta_capacity.get() / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR, + )) +} + impl SystemMetadata { /// Creates a new empty system metadata state. pub fn new(own_subnet_id: SubnetId, own_subnet_type: SubnetType) -> Self { @@ -614,6 +647,32 @@ impl SystemMetadata { .get_reference_subnet_size(&self.own_subnet_id) } + /// Returns the subnet's guaranteed response message memory capacity, capped + /// relative to the subnet's heap delta capacity. + pub fn guaranteed_response_message_memory_capacity(&self) -> NumBytes { + message_memory_capacity( + SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY, + self.heap_delta_capacity(), + ) + } + + /// Returns the subnet's best-effort message memory capacity, capped relative + /// to the subnet's heap delta capacity. + pub fn best_effort_message_memory_capacity(&self) -> NumBytes { + message_memory_capacity( + SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY, + self.heap_delta_capacity(), + ) + } + + /// The effective heap delta capacity: the registry override if set, else the + /// protocol default. + fn heap_delta_capacity(&self) -> NumBytes { + self.own_subnet_info + .resource_limits + .maximum_state_delta_or(ic_config::execution_environment::SUBNET_HEAP_DELTA_CAPACITY) + } + /// One-off initialization: populate `canister_allocation_ranges` with the only /// `[N * 2^20, (N+1) * 2^20 - 1]` range fully hosted by the subnet as per the /// routing table; and initialize `last_generated_canister_id` based on @@ -2081,6 +2140,15 @@ impl UnflushedCheckpointOps { pub mod testing { use super::*; + /// Test helper for configuring a subnet's `maximum_state_delta`: returns the + /// heap delta capacity required to allow `message_memory` of message memory, + /// i.e. the inverse of the cap applied by `message_memory_capacity`. Lives + /// here so the message-memory/heap-delta factor stays confined to this + /// module rather than being duplicated across tests. + pub fn heap_delta_capacity_for_message_memory(message_memory: NumBytes) -> NumBytes { + NumBytes::new(message_memory.get() * MESSAGE_MEMORY_HEAP_DELTA_DIVISOR) + } + /// Exposes `SystemMetadata` internals for use in tests. pub trait SystemMetadataTesting { fn modify_network_topology(&mut self, f: impl FnOnce(&mut NetworkTopology)); diff --git a/rs/state_machine_tests/src/lib.rs b/rs/state_machine_tests/src/lib.rs index afeb76c6b170..09fe2a913339 100644 --- a/rs/state_machine_tests/src/lib.rs +++ b/rs/state_machine_tests/src/lib.rs @@ -682,6 +682,7 @@ fn into_cbor(r: &R) -> Vec { pub struct StateMachineConfig { subnet_config: SubnetConfig, hypervisor_config: HypervisorConfig, + resource_limits: ResourceLimits, } impl StateMachineConfig { @@ -689,8 +690,14 @@ impl StateMachineConfig { Self { subnet_config, hypervisor_config, + resource_limits: ResourceLimits::default(), } } + + pub fn with_resource_limits(mut self, resource_limits: ResourceLimits) -> Self { + self.resource_limits = resource_limits; + self + } } /// Struct mocking consensus time required for instantiating `IngressManager` @@ -1254,7 +1261,6 @@ pub struct StateMachine { chain_key_payload_builder: Arc, remove_old_states: bool, cycles_account_manager: Arc, - hypervisor_config: HypervisorConfig, } impl Default for StateMachine { @@ -2438,7 +2444,6 @@ impl StateMachine { chain_key_payload_builder, remove_old_states, cycles_account_manager: execution_services.cycles_account_manager, - hypervisor_config, } } @@ -5329,9 +5334,9 @@ impl StateMachine { } } }); - let mut available_guaranteed_response_memory = self - .hypervisor_config - .guaranteed_response_message_memory_capacity + let mut available_guaranteed_response_memory = replicated_state + .metadata + .guaranteed_response_message_memory_capacity() .get() as i64 - replicated_state .guaranteed_response_message_memory_taken() @@ -5601,6 +5606,7 @@ fn multi_subnet_setup( registry_data_provider: Arc, ) -> Arc { StateMachineBuilder::new() + .with_resource_limits(config.resource_limits) .with_config(Some(config)) .with_subnet_seed([subnet_seed; 32]) .with_subnet_type(subnet_type) diff --git a/rs/test_utilities/execution_environment/src/lib.rs b/rs/test_utilities/execution_environment/src/lib.rs index 34a5b0399948..fad5ea485392 100644 --- a/rs/test_utilities/execution_environment/src/lib.rs +++ b/rs/test_utilities/execution_environment/src/lib.rs @@ -51,6 +51,7 @@ use ic_registry_routing_table::{ }; use ic_registry_subnet_features::SubnetFeatures; use ic_registry_subnet_type::SubnetType; +use ic_replicated_state::metadata_state::testing::heap_delta_capacity_for_message_memory; use ic_replicated_state::{ CallContext, CanisterState, ExecutionState, ExecutionTask, InputQueueType, NetworkTopology, PageIndex, ReplicatedState, SubnetTopology, @@ -2614,9 +2615,9 @@ impl ExecutionTestBuilder { mut self, subnet_guaranteed_response_message_memory: u64, ) -> Self { - self.execution_config - .guaranteed_response_message_memory_capacity = - NumBytes::from(subnet_guaranteed_response_message_memory); + self.resource_limits.maximum_state_delta = Some(heap_delta_capacity_for_message_memory( + NumBytes::from(subnet_guaranteed_response_message_memory), + )); self } @@ -2983,6 +2984,7 @@ impl ExecutionTestBuilder { own_subnet_info.subnet_features = SubnetFeatures::from_str(&self.subnet_features).unwrap(); } + own_subnet_info.resource_limits = self.resource_limits; let metrics_registry = MetricsRegistry::new();