feat(canister_logging): add log_memory_store to system_state

rs/embedders/src/wasmtime_embedder/system_api/sandbox_safe_system_state.rs

@@ -513,10 +513,23 @@ impl SystemStateModifications {
             system_state.global_timer = new_global_timer;
         }
 
+        // TODO(DSM-11): cleanup population logic after migration is done.
+        // We need to copy existing canister_log to log_memory_store in order
+        // not to loose any log records until the migration is complete.
+        if system_state.log_memory_store.is_empty() && !system_state.canister_log.is_empty() {
+            system_state
+                .log_memory_store
+                .append_delta_log(&mut system_state.canister_log.clone());
+        }
+
         // Append delta log to the total canister log.
+        let mut canister_log_copy = self.canister_log.clone();
         system_state
             .canister_log
             .append_delta_log(&mut self.canister_log);
+        system_state
+            .log_memory_store
+            .append_delta_log(&mut canister_log_copy);
 
         // Bump the canister version after all changes have been applied.
         if self.should_bump_canister_version {

rs/execution_environment/src/canister_manager.rs

@@ -562,8 +562,11 @@ impl CanisterManager {
         if let Some(log_visibility) = settings.log_visibility() {
             canister.system_state.log_visibility = log_visibility.clone();
         }
-        if let Some(_log_memory_limit) = settings.log_memory_limit() {
-            // TODO: populate log_memory_store with the new limit.
+        if let Some(log_memory_limit) = settings.log_memory_limit() {
+            canister
+                .system_state
+                .log_memory_store
+                .set_byte_capacity(log_memory_limit.get() as usize);
         }
         if let Some(wasm_memory_limit) = settings.wasm_memory_limit() {
             canister.system_state.wasm_memory_limit = Some(wasm_memory_limit);
@@ -1099,7 +1102,7 @@ impl CanisterManager {
         let freeze_threshold = canister.system_state.freeze_threshold;
         let reserved_cycles_limit = canister.system_state.reserved_balance_limit();
         let log_visibility = canister.system_state.log_visibility.clone();
-        let log_memory_limit = canister.system_state.canister_log.byte_capacity();
+        let log_memory_limit = canister.system_state.log_memory_store.byte_capacity();
         let wasm_memory_limit = canister.system_state.wasm_memory_limit;
         let wasm_memory_threshold = canister.system_state.wasm_memory_threshold;
 

rs/replicated_state/src/canister_state/system_state.rs

@@ -1,15 +1,15 @@
 mod call_context_manager;
+pub mod log_memory_store;
 pub mod proto;
 mod task_queue;
 pub mod wasm_chunk_store;
 
-// TODO(DSM-11): remove testing cofiguration when log memory store is used in production.
-#[cfg(test)]
-pub mod log_memory_store;
-
 pub use self::task_queue::{TaskQueue, is_low_wasm_memory_hook_condition_satisfied};
 
-use self::wasm_chunk_store::{WasmChunkStore, WasmChunkStoreMetadata};
+use self::{
+    log_memory_store::LogMemoryStore,
+    wasm_chunk_store::{WasmChunkStore, WasmChunkStoreMetadata},
+};
 use super::queues::refunds::RefundPool;
 use super::queues::{CanisterInput, can_push};
 pub use super::queues::{CanisterOutputQueuesIterator, memory_usage_of_request};
@@ -340,6 +340,10 @@ pub struct SystemState {
     #[validate_eq(CompareWithValidateEq)]
     pub canister_log: CanisterLog,
 
+    /// The memory used for storing log entries.
+    #[validate_eq(CompareWithValidateEq)]
+    pub log_memory_store: LogMemoryStore,
+
     /// The Wasm memory limit. This is a field in developer-visible canister
     /// settings that allows the developer to limit the usage of the Wasm memory
     /// by the canister to leave some room in 4GiB for upgrade calls.
@@ -484,7 +488,8 @@ impl SystemState {
             initial_cycles,
             freeze_threshold,
             CanisterStatus::new_running(),
-            WasmChunkStore::new(fd_factory),
+            WasmChunkStore::new(fd_factory.clone()),
+            LogMemoryStore::new(fd_factory),
         )
     }
 
@@ -495,6 +500,7 @@ impl SystemState {
         freeze_threshold: NumSeconds,
         status: CanisterStatus,
         wasm_chunk_store: WasmChunkStore,
+        log_memory_store: LogMemoryStore,
     ) -> Self {
         Self {
             canister_id,
@@ -521,6 +527,7 @@ impl SystemState {
             // therefore it should not scale to memory limit from above.
             // Remove this field after migration is done.
             canister_log: CanisterLog::default_aggregate(),
+            log_memory_store,
             wasm_memory_limit: None,
             next_snapshot_id: 0,
             snapshots_memory_usage: NumBytes::new(0),
@@ -550,6 +557,7 @@ impl SystemState {
         wasm_chunk_store_metadata: WasmChunkStoreMetadata,
         log_visibility: LogVisibilityV2,
         canister_log: CanisterLog,
+        log_memory_store_data: PageMap,
         wasm_memory_limit: Option<NumBytes>,
         next_snapshot_id: u64,
         snapshots_memory_usage: NumBytes,
@@ -580,6 +588,7 @@ impl SystemState {
             ),
             log_visibility,
             canister_log,
+            log_memory_store: LogMemoryStore::from_checkpoint(log_memory_store_data),
             wasm_memory_limit,
             next_snapshot_id,
             snapshots_memory_usage,
@@ -653,6 +662,7 @@ impl SystemState {
             freeze_threshold,
             status,
             WasmChunkStore::new_for_testing(),
+            LogMemoryStore::new_for_testing(),
         )
     }
 
@@ -2219,6 +2229,7 @@ pub mod testing {
             // therefore it should not scale to memory limit from above.
             // Remove this field after migration is done.
             canister_log: CanisterLog::default_aggregate(),
+            log_memory_store: LogMemoryStore::new_for_testing(),
             wasm_memory_limit: Default::default(),
             next_snapshot_id: Default::default(),
             snapshots_memory_usage: Default::default(),

rs/state_layout/src/state_layout.rs

@@ -76,6 +76,7 @@ pub const OVERLAY: &str = "overlay";
 pub const VMEMORY_0: &str = "vmemory_0";
 pub const STABLE_MEMORY: &str = "stable_memory";
 pub const WASM_CHUNK_STORE: &str = "wasm_chunk_store";
+pub const LOG_MEMORY_STORE: &str = "log_memory_store";
 pub const BIN_FILE: &str = "bin";
 
 /// `ReadOnly` is the access policy used for reading checkpoints. We
@@ -308,7 +309,8 @@ struct CheckpointRefData {
 /// │      │       ├── software.wasm
 /// │      │       ├── stable_memory.bin
 /// │      │       ├── vmemory_0.bin
-/// │      │       └── wasm_chunk_store.bin
+/// │      │       ├── wasm_chunk_store.bin
+/// │      │       └── log_memory_store.bin
 /// │      ├── snapshots
 /// │      │   └── <hex(canister_id)>
 /// │      │       └──  <hex(snapshot_id)>
@@ -2196,6 +2198,7 @@ impl<Permissions: AccessPolicy> CanisterLayout<Permissions> {
             self.vmemory_0(),
             self.stable_memory(),
             self.wasm_chunk_store(),
+            self.log_memory_store(),
         ]
         .into_iter()
         {
@@ -2232,6 +2235,15 @@ impl<Permissions: AccessPolicy> CanisterLayout<Permissions> {
             _checkpoint: self.checkpoint.clone(),
         }
     }
+
+    pub fn log_memory_store(&self) -> PageMapLayout<Permissions> {
+        PageMapLayout {
+            root: self.canister_root.clone(),
+            name_stem: LOG_MEMORY_STORE.into(),
+            permissions_tag: PhantomData,
+            _checkpoint: self.checkpoint.clone(),
+        }
+    }
 }
 
 pub struct SnapshotLayout<Permissions: AccessPolicy> {
@@ -2286,6 +2298,7 @@ impl<Permissions: AccessPolicy> SnapshotLayout<Permissions> {
             self.vmemory_0(),
             self.stable_memory(),
             self.wasm_chunk_store(),
+            // log_memory_store is not included in canister snapshots.
         ]
         .into_iter()
         {
@@ -2322,6 +2335,8 @@ impl<Permissions: AccessPolicy> SnapshotLayout<Permissions> {
             _checkpoint: self.checkpoint.clone(),
         }
     }
+
+    // log_memory_store is not included in canister snapshots.
 }
 
 impl<P> SnapshotLayout<P>

rs/state_manager/src/checkpoint.rs

@@ -179,6 +179,7 @@ pub(crate) enum PageMapType {
     WasmMemory(CanisterId),
     StableMemory(CanisterId),
     WasmChunkStore(CanisterId),
+    LogMemoryStore(CanisterId),
     SnapshotWasmMemory(SnapshotId),
     SnapshotStableMemory(SnapshotId),
     SnapshotWasmChunkStore(SnapshotId),
@@ -190,6 +191,7 @@ impl PageMapType {
         let mut result = vec![];
         for (id, canister) in &state.canister_states {
             result.push(Self::WasmChunkStore(id.to_owned()));
+            result.push(Self::LogMemoryStore(id.to_owned()));
             if canister.execution_state.is_some() {
                 result.push(Self::WasmMemory(id.to_owned()));
                 result.push(Self::StableMemory(id.to_owned()));
@@ -223,6 +225,7 @@ impl PageMapType {
             PageMapType::WasmMemory(id) => Ok(layout.canister(id)?.vmemory_0()),
             PageMapType::StableMemory(id) => Ok(layout.canister(id)?.stable_memory()),
             PageMapType::WasmChunkStore(id) => Ok(layout.canister(id)?.wasm_chunk_store()),
+            PageMapType::LogMemoryStore(id) => Ok(layout.canister(id)?.log_memory_store()),
             PageMapType::SnapshotWasmMemory(id) => Ok(layout.snapshot(id)?.vmemory_0()),
             PageMapType::SnapshotStableMemory(id) => Ok(layout.snapshot(id)?.stable_memory()),
             PageMapType::SnapshotWasmChunkStore(id) => Ok(layout.snapshot(id)?.wasm_chunk_store()),
@@ -245,6 +248,9 @@ impl PageMapType {
             PageMapType::WasmChunkStore(id) => state
                 .canister_state(id)
                 .map(|can| can.system_state.wasm_chunk_store.page_map()),
+            PageMapType::LogMemoryStore(id) => state
+                .canister_state(id)
+                .map(|can| can.system_state.log_memory_store.page_map()),
             PageMapType::SnapshotWasmMemory(id) => state
                 .canister_snapshots
                 .get(*id)
@@ -275,6 +281,11 @@ fn strip_page_map_deltas(
             .wasm_chunk_store
             .page_map_mut()
             .strip_all_deltas(Arc::clone(&fd_factory));
+        canister
+            .system_state
+            .log_memory_store
+            .page_map_mut()
+            .strip_all_deltas(Arc::clone(&fd_factory));
         if let Some(execution_state) = canister.execution_state.as_mut() {
             execution_state
                 .wasm_memory
@@ -356,6 +367,10 @@ pub(crate) fn flush_canister_snapshots_and_page_maps(
             PageMapType::WasmChunkStore(id.to_owned()),
             canister.system_state.wasm_chunk_store.page_map_mut(),
         );
+        add_to_pagemaps_and_strip(
+            PageMapType::LogMemoryStore(id.to_owned()),
+            canister.system_state.log_memory_store.page_map_mut(),
+        );
         if let Some(execution_state) = canister.execution_state.as_mut() {
             add_to_pagemaps_and_strip(
                 PageMapType::WasmMemory(id.to_owned()),
@@ -881,6 +896,15 @@ pub fn load_canister_state(
     )?;
     durations.insert("wasm_chunk_store", starting_time.elapsed());
 
+    let starting_time = Instant::now();
+    let log_memory_store_layout = canister_layout.log_memory_store();
+    let log_memory_store_data = PageMap::open(
+        Box::new(log_memory_store_layout),
+        height,
+        Arc::clone(&fd_factory),
+    )?;
+    durations.insert("log_memory_store", starting_time.elapsed());
+
     let system_state = SystemState::new_from_checkpoint(
         canister_state_bits.controllers,
         *canister_id,
@@ -903,6 +927,7 @@ pub fn load_canister_state(
         canister_state_bits.wasm_chunk_store_metadata,
         canister_state_bits.log_visibility,
         canister_state_bits.canister_log,
+        log_memory_store_data,
         canister_state_bits.wasm_memory_limit,
         canister_state_bits.next_snapshot_id,
         canister_state_bits.snapshots_memory_usage,

rs/state_manager/tests/state_manager.rs

@@ -318,7 +318,13 @@ fn lsmt_merge_overhead() {
             assert_ne!(tip_size(&env), 0.0);
             assert_ne!(state_in_memory(&env), 0.0);
             assert!(tip_size(&env) / state_in_memory(&env) > 2.0);
-            assert!(tip_size(&env) / state_in_memory(&env) <= 2.5);
+            assert!(
+                tip_size(&env) / state_in_memory(&env) <= 2.5,
+                "tip_size: {} state_in_memory: {}, tip_size / state_in_memory: {}",
+                tip_size(&env),
+                state_in_memory(&env),
+                tip_size(&env) / state_in_memory(&env)
+            );
         }
     }
     // Create a checkpoint from the tip without writing any more data. As we merge in tip, the
@@ -366,7 +372,7 @@ fn lazy_pagemaps() {
     let canister_id = env.install_canister_wat(TEST_CANISTER, vec![], None);
 
     env.tick();
-    assert_eq!(page_maps_by_status("loaded", &env), 0);
+    assert_eq!(page_maps_by_status("loaded", &env), 1);
     assert!(page_maps_by_status("not_loaded", &env) > 0);
 
     env.execute_ingress(canister_id, "write_heap_64k", vec![])

rs/types/types/src/canister_log.rs

@@ -222,6 +222,11 @@ impl CanisterLog {
         self.records.bytes_used
     }
 
+    /// Returns true if the canister log is empty.
+    pub fn is_empty(&self) -> bool {
+        self.records.bytes_used == 0
+    }
+
     /// Returns the remaining space in the canister log buffer.
     pub fn remaining_bytes(&self) -> usize {
         let records = &self.records;