roll is done externally

Author: fulmicoton

src/mem/arena.rs

@@ -1,10 +1,3 @@
-use std::time::Duration;
-#[cfg(not(test))]
-use std::time::Instant;
-
-#[cfg(test)]
-use mock_instant::Instant;
-
 /// 256 KiB.
 #[cfg(not(test))]
 pub const PAGE_SIZE: usize = 1 << 18;
@@ -56,44 +49,27 @@ pub struct Arena {
 struct ArenaStats {
     max_num_used_pages_former: usize,
     max_num_used_pages_current: usize,
-    call_counter: u8,
-    next_window_start: Instant,
 }
 
-const WINDOW: Duration = Duration::from_secs(60);
-
 impl Default for ArenaStats {
     fn default() -> ArenaStats {
         ArenaStats {
-            // We arbitrarily initialize num used pages former to 100.
             max_num_used_pages_former: 0,
             max_num_used_pages_current: 0,
-            call_counter: 0u8,
-            next_window_start: Instant::now(),
         }
     }
 }
 
 impl ArenaStats {
     /// This method happens when we are changing time window.
-    fn roll(&mut self, now: Instant) {
+    fn roll(&mut self) {
         self.max_num_used_pages_former = self.max_num_used_pages_current;
         self.max_num_used_pages_current = 0;
-        self.next_window_start = now + WINDOW;
     }
 
     /// Records the number of used pages, and returns an estimation of the maximum number of pages
     /// in the last 5 minutes.
     pub fn record_num_used_page(&mut self, num_used_pages: usize) -> usize {
-        // The only function of the call counter is to avoid calling `Instant::now()`
-        // at every single call.
-        self.call_counter = (self.call_counter + 1) % 64;
-        if self.call_counter == 0u8 {
-            let now = Instant::now();
-            if now > self.next_window_start {
-                self.roll(now);
-            }
-        }
         self.max_num_used_pages_current = self.max_num_used_pages_current.max(num_used_pages);
         self.max_num_used_pages_former
             .max(self.max_num_used_pages_current)
@@ -105,20 +81,17 @@ impl Arena {
     pub fn acquire_page(&mut self) -> PageId {
         if let Some(page_id) = self.free_page_ids.pop() {
             assert!(self.pages[page_id.0].is_some());
-            self.gc();
             return page_id;
         }
         let page: Page = vec![0u8; PAGE_SIZE].into_boxed_slice();
         if let Some(free_slot) = self.free_slots.pop() {
             let slot = &mut self.pages[free_slot.0];
             assert!(slot.is_none());
             *slot = Some(page);
-            self.gc();
             free_slot
         } else {
             let new_page_id = self.pages.len();
             self.pages.push(Some(page));
-            self.gc();
             PageId(new_page_id)
         }
     }
@@ -139,20 +112,25 @@ impl Arena {
         self.gc();
     }
 
-    /// `gc` releases memory by deallocating ALL of the free pages.
-    pub fn gc(&mut self) {
+    /// Clients are expected roll the stats regularly.
+    pub fn roll_and_gc(&mut self) {
+        self.stats.roll();
+        self.gc();
+    }
+
+    /// `gc` releases memory by some of the free pages.
+    fn gc(&mut self) {
         let num_used_pages = self.num_used_pages();
         let max_used_num_pages_in_last_5_min = self.stats.record_num_used_page(num_used_pages);
-        // We pick a target slightly higher than the maximum number of pages used in the last 5
-        // minutes to avoid needless allocations when we are experience a general increase
+        // We pick a target slightly higher than the maximum number of pages to avoid needless
+        // allocations when we are experience a general increase
         // in memory usage.
         let target_num_pages = max_used_num_pages_in_last_5_min + 10;
         let num_pages_to_free = self.num_allocated_pages().saturating_sub(target_num_pages);
-        assert!(num_pages_to_free <= self.free_page_ids.len());
-        for _ in 0..num_pages_to_free {
-            let page_id = self.free_page_ids.pop().unwrap();
-            self.pages[page_id.0] = None;
-            self.free_slots.push(page_id);
+        let num_free_pages_to_keep = self.free_page_ids.len() - num_pages_to_free;
+        for free_page_id in self.free_page_ids.drain(num_free_pages_to_keep..) {
+            self.pages[free_page_id.0] = None;
+            self.free_slots.push(free_page_id);
         }
     }
 

src/mem/queues.rs

@@ -160,6 +160,10 @@ impl MemQueues {
         Some(queue.truncate_up_to_included(position, &mut self.arena))
     }
 
+    pub fn roll_and_gc(&mut self) {
+        self.arena.roll_and_gc();
+    }
+
     /// Return a tuple of (size, capacity) of memory used by the memqueues
     pub fn size(&self) -> (usize, usize) {
         let size = self

src/mem/rolling_buffer.rs

@@ -176,7 +176,8 @@ impl<'a> PagesBuf<'a> {
         std::borrow::Cow::Owned(buf)
     }
 
-    // Contrary to Buf::chunk, this method returns a slice with a `'a` lifetime (so it can outlive 'self).
+    // Contrary to Buf::chunk, this method returns a slice with a `'a` lifetime (so it can outlive
+    // 'self).
     fn chunk_with_lifetime(&self) -> &'a [u8] {
         let Some(first_page_id) = self.page_ids.first().copied() else {
             return &[];

src/multi_record_log.rs

@@ -293,6 +293,10 @@ impl MultiRecordLog {
         Ok(removed_count)
     }
 
+    pub fn memory_gc(&mut self) {
+        self.in_mem_queues.roll_and_gc();
+    }
+
     fn run_gc_if_necessary(&mut self) -> io::Result<()> {
         debug!("run_gc_if_necessary");
         if self