DOSE-940 zettacache initiates merge every checkpoint (openzfs#116)

cmd/zfs_object_agent/zettacache/src/atime_histogram.rs

@@ -55,26 +55,45 @@ impl AtimeHistogramPhys {
         self.first_live = new_first;
     }
 
-    pub fn atime_for_target_size(&self, target_size: u64) -> Atime {
-        info!(
-            "histogram starts at {:?} (live at {:?}) and has {} entries",
-            self.first_ghost,
-            self.first_live,
-            self.histogram.len()
-        );
-        let mut remaining = target_size;
-        for (index, &bytes) in self.histogram.iter().enumerate().rev() {
+    /// Given an atime "starting point", calculate the "end" atime such that
+    /// self.histogram[start..end].sum() is >= the provided target value or
+    /// return the next atime past the end of the vector if sum() < "target"
+    /// data in the vector. Return "start" if the target value is 0.
+    fn atime_for_target(&self, start: Atime, target: u64) -> Atime {
+        if target == 0 {
+            return start;
+        }
+        let mut end = start;
+        let mut remaining = target;
+        for &bytes in self.histogram[start - self.first_ghost..].iter() {
+            end = end.next();
             if remaining <= bytes {
-                debug!("found target size {} at bucket {}", target_size, index);
-                return self.first_ghost + index;
+                break;
             }
             remaining -= bytes;
         }
+        end
+    }
+
+    pub fn atime_for_eviction_target(&self, eviction_size: u64) -> Atime {
         debug!(
-            "cache smaller than target size {} by {} bytes",
-            target_size, remaining
+            "histogram live start at {:?} with {} entries, evicting {}MB",
+            self.first_live,
+            self.histogram.len(),
+            eviction_size / 1024 / 1024,
         );
-        self.first_ghost
+        self.atime_for_target(self.first_live, eviction_size)
+    }
+
+    pub fn atime_for_ghost_target(&self, ghost_reduction: u64) -> Atime {
+        let atime = self.atime_for_target(self.first_ghost, ghost_reduction);
+        debug!(
+            "ghost size reduction of {}MB moves start from {:?} to {:?}",
+            ghost_reduction / 1024 / 1024,
+            self.first_ghost,
+            atime
+        );
+        std::cmp::min(atime, self.first_live)
     }
 
     pub fn insert(&mut self, value: IndexValue) {
@@ -96,8 +115,8 @@ impl AtimeHistogramPhys {
         self.histogram.clear();
     }
 
-    pub fn sum(&self) -> u64 {
-        self.histogram.iter().sum()
+    pub fn sum_live(&self) -> u64 {
+        self.size_at(self.first_live)
     }
 
     /// Add up all the atime histogram buckets from key atime

cmd/zfs_object_agent/zettacache/src/zettacache.rs

@@ -2426,7 +2426,7 @@ impl ZettaCacheState {
         old_index: Arc<tokio::sync::RwLock<ZettaCacheIndex>>,
     ) -> Option<tokio::sync::mpsc::Receiver<MergeMessage>> {
         if self.pending_changes.len() < *MAX_PENDING_CHANGES
-            && self.atime_histogram.sum()
+            && self.block_allocator.size() - self.block_allocator.available()
                 < (self.block_allocator.size() / 100) * *HIGH_WATER_CACHE_SIZE_PCT
             && !self.block_allocator.rebalance_needed()
         {
@@ -2437,15 +2437,17 @@ impl ZettaCacheState {
             return None;
         }
 
+        let used = self.block_allocator.size() - self.block_allocator.available();
         let target_size = (self.block_allocator.size() / 100) * *TARGET_CACHE_SIZE_PCT;
+        let target_reduction = used.checked_sub(target_size).unwrap_or_default();
         info!(
             "target cache size for storage size {}GB is {}GB; {}MB used; {}MB high-water; {}MB freeing; histogram covers {}MB",
             self.block_allocator.size() / 1024 / 1024 / 1024,
             target_size / 1024 / 1024 / 1024,
-            (self.block_allocator.size() - self.block_allocator.available()) / 1024 / 1024,
+            used / 1024 / 1024,
             (self.block_allocator.size() / 100) * *HIGH_WATER_CACHE_SIZE_PCT / 1024 / 1024,
             self.block_allocator.freeing() / 1024 / 1024,
-            self.atime_histogram.sum() / 1024 / 1024,
+            self.atime_histogram.sum_live() / 1024 / 1024,
         );
         self.stats
             .track_instantaneous(BlockAllocatorSize, self.block_allocator.size());
@@ -2456,10 +2458,12 @@ impl ZettaCacheState {
             self.block_allocator.free_slabs_size(),
         );
 
-        let eviction_atime = self.atime_histogram.atime_for_target_size(target_size);
+        let eviction_atime = self
+            .atime_histogram
+            .atime_for_eviction_target(target_reduction);
         let ghost_atime = self
             .atime_histogram
-            .atime_for_target_size(target_size + target_size / 100 * *GHOST_CACHE_SIZE_PCT);
+            .atime_for_ghost_target(target_reduction / 100 * *GHOST_CACHE_SIZE_PCT);
 
         let old_operation_log_phys = self.operation_log.flush().await;