Merge branch 'ulan/run-718' into 'master'

RUN-718: Add a `scaling_factor` field to `SubnetAvailableMemory`

Currently the subnet available memory is divided evenly between
execution threads in order to enforce the subnet limit
deterministically.

This MR adds a new `scaling_factor` field to `SubnetAvailableMemory`.
The field keeps track of the division factor such that an execution
thread can estimate the global available memory from the per-thread
available memory.

Using this new field, the MR fixes the subnet memory reservation by
scaling it consistently with the subnet available memory. 

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

rs/execution_environment/src/execution_environment.rs

@@ -1435,6 +1435,13 @@ impl ExecutionEnvironment {
             log: &self.log,
             time,
         };
+        // This function is called on an execution thread with a scaled
+        // available memory. We also need to scale the subnet reservation in
+        // order to be consistent with the scaling of the available memory.
+        let scaled_subnet_memory_reservation = NumBytes::new(
+            self.config.subnet_memory_reservation.get()
+                / round_limits.subnet_available_memory.get_scaling_factor() as u64,
+        );
         execute_response(
             canister,
             response,
@@ -1444,7 +1451,7 @@ impl ExecutionEnvironment {
             round,
             round_limits,
             subnet_size,
-            self.config.subnet_memory_reservation,
+            scaled_subnet_memory_reservation,
         )
     }
 

rs/execution_environment/tests/execution_test.rs

@@ -794,6 +794,7 @@ fn max_canister_memory_respected_even_when_no_memory_allocation_is_set() {
 #[test]
 fn subnet_memory_reservation_works() {
     let subnet_config = SubnetConfig::new(SubnetType::Application);
+    let num_cores = subnet_config.scheduler_config.scheduler_cores as u64;
     let env = StateMachine::new_with_config(StateMachineConfig::new(
         subnet_config,
         HypervisorConfig {
@@ -828,7 +829,7 @@ fn subnet_memory_reservation_works() {
         .build();
 
     // The update call grow and the response callback both grow memory by
-    // roughly 50MB.
+    // roughly `50MB / num_cores`.
     let a = wasm()
         .stable_grow(800)
         .call_with_cycles(
@@ -839,8 +840,8 @@ fn subnet_memory_reservation_works() {
                 .on_reject(wasm().reject_code().reject_message().reject())
                 .on_reply(
                     wasm()
-                        .stable_grow(800)
-                        .stable64_fill(0, 0, 1000)
+                        .stable_grow(800 / num_cores as u32)
+                        .stable64_fill(0, 0, 1000 / num_cores)
                         .instruction_counter_is_at_least(1_000_000)
                         .message_payload()
                         .append_and_reply(),
@@ -856,6 +857,75 @@ fn subnet_memory_reservation_works() {
     }
 }
 
+#[test]
+fn subnet_memory_reservation_scales_with_number_of_cores() {
+    let subnet_config = SubnetConfig::new(SubnetType::Application);
+    let num_cores = subnet_config.scheduler_config.scheduler_cores as u64;
+    assert!(num_cores > 1);
+    let env = StateMachine::new_with_config(StateMachineConfig::new(
+        subnet_config,
+        HypervisorConfig {
+            subnet_memory_capacity: NumBytes::from(120 * 1024 * 1024),
+            subnet_memory_reservation: NumBytes::from(50 * 1024 * 1024),
+            ..Default::default()
+        },
+    ));
+
+    let a_id = env
+        .install_canister_with_cycles(
+            UNIVERSAL_CANISTER_WASM.into(),
+            vec![],
+            Some(CanisterSettingsArgsBuilder::new().build()),
+            INITIAL_CYCLES_BALANCE,
+        )
+        .unwrap();
+
+    let b_id = env
+        .install_canister_with_cycles(
+            UNIVERSAL_CANISTER_WASM.into(),
+            vec![],
+            Some(CanisterSettingsArgsBuilder::new().build()),
+            INITIAL_CYCLES_BALANCE,
+        )
+        .unwrap();
+
+    let b = wasm()
+        .accept_cycles(Cycles::from(1_000u128))
+        .message_payload()
+        .append_and_reply()
+        .build();
+
+    // The update call grow and the response callback both grow memory by
+    // roughly 50MB. It should fail because there are at least two threads
+    // and each threads gets `50MB / num_cores` reservation.
+    let a = wasm()
+        .stable_grow(800)
+        .call_with_cycles(
+            b_id.get(),
+            "update",
+            call_args()
+                .other_side(b)
+                .on_reject(wasm().reject_code().reject_message().reject())
+                .on_reply(
+                    wasm()
+                        .stable_grow(800)
+                        .stable64_fill(0, 0, 1000)
+                        .instruction_counter_is_at_least(1_000_000)
+                        .message_payload()
+                        .append_and_reply(),
+                ),
+            Cycles::from(2000u128),
+        )
+        .build();
+
+    let err = env.execute_ingress(a_id, "update", a).unwrap_err();
+    assert_eq!(
+        err.description(),
+        format!("Canister {} trapped: stable memory out of bounds", a_id)
+    );
+    assert_eq!(err.code(), ErrorCode::CanisterTrapped);
+}
+
 #[test]
 fn canister_with_memory_allocation_does_not_fail_when_growing_wasm_memory() {
     let subnet_config = SubnetConfig::new(SubnetType::Application);

rs/interfaces/src/execution_environment.rs

@@ -172,6 +172,10 @@ pub struct SubnetAvailableMemory {
     message_memory: i64,
     /// The memory available for Wasm custom sections.
     wasm_custom_sections_memory: i64,
+    /// Specifies the factor by which the subnet available memory was scaled
+    /// using the division operator. It is useful for approximating the global
+    /// available memory from the per-thread available memory.
+    scaling_factor: i64,
 }
 
 impl SubnetAvailableMemory {
@@ -184,6 +188,9 @@ impl SubnetAvailableMemory {
             execution_memory,
             message_memory,
             wasm_custom_sections_memory,
+            // The newly created value is not scaled (divided), which
+            // corresponds to the scaling factor of 1.
+            scaling_factor: 1,
         }
     }
 
@@ -203,6 +210,17 @@ impl SubnetAvailableMemory {
         self.wasm_custom_sections_memory
     }
 
+    /// Returns the scaling factor that specifies by how much the initial
+    /// available memory was scaled using the division operator.
+    ///
+    /// It is useful for approximating the global available memory from the
+    /// per-thread available memory. Note that the approximation may be off in
+    /// both directions because there is no way to deterministically know how
+    /// much other threads have allocated.
+    pub fn get_scaling_factor(&self) -> i64 {
+        self.scaling_factor
+    }
+
     /// Try to use some memory capacity and fail if not enough is available
     ///
     /// `self.execution_memory`, `self.message_memory` and `self.wasm_custom_sections_memory`
@@ -289,6 +307,7 @@ impl ops::Div<i64> for SubnetAvailableMemory {
             execution_memory: self.execution_memory / rhs,
             message_memory: self.message_memory / rhs,
             wasm_custom_sections_memory: self.wasm_custom_sections_memory / rhs,
+            scaling_factor: self.scaling_factor * rhs,
         }
     }
 }