DOSE-505 Eviction: Incremental freeing (once per checkpoint) (openzfs…

…#451)

* Checkpoint driven incremental async merge and evict
- mostly lockless lookup
  * use eviction policy check
  * use only old index
- create a new channel for each merge task
  * merge progress sent to checkpoint task
  * next_index returned as last message
- constructs free lists in merge
  * pass w/index phys once a second to checkpoint task
  * index is flushed in merge task rather than checkpoint task
  * frees done in checkpoint task (for consistent checkpoint data)
- can restart merge using checkpoint and index last_key

cmd/zfs_object_agent/zettacache/src/block_based_log.rs

@@ -354,7 +354,8 @@ impl<T: BlockBasedLogEntry> BlockBasedLogWithSummary<T> {
         }
     }
 
-    // Works only if there are no pending entries
+    /// Works only if there are no pending entries.
+    /// Use flush() to retrieve the phys when there are pending entries.
     pub fn get_phys(&self) -> BlockBasedLogWithSummaryPhys<T> {
         assert!(self.this.pending_entries.is_empty());
         assert!(self.chunk_summary.pending_entries.is_empty());

cmd/zfs_object_agent/zettacache/src/index.rs

@@ -3,6 +3,7 @@ use crate::block_access::*;
 use crate::block_based_log::*;
 use crate::extent_allocator::ExtentAllocator;
 use crate::zettacache::AtimeHistogramPhys;
+use more_asserts::*;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 
@@ -40,62 +41,87 @@ pub struct IndexEntry {
 impl OnDisk for IndexEntry {}
 impl BlockBasedLogEntry for IndexEntry {}
 
-#[derive(Serialize, Deserialize, Debug, Default)]
+#[derive(Serialize, Deserialize, Debug, Default, Clone)]
 pub struct ZettaCacheIndexPhys {
+    last_key: Option<IndexKey>,
     atime_histogram: AtimeHistogramPhys,
     log: BlockBasedLogWithSummaryPhys<IndexEntry>,
 }
 
 impl ZettaCacheIndexPhys {
     pub fn new(min_atime: Atime) -> Self {
         Self {
+            last_key: None,
             atime_histogram: AtimeHistogramPhys::new(min_atime),
             log: Default::default(),
         }
     }
 }
 
 pub struct ZettaCacheIndex {
+    pub last_key: Option<IndexKey>,
     pub atime_histogram: AtimeHistogramPhys,
     pub log: BlockBasedLogWithSummary<IndexEntry>,
 }
 
+impl std::fmt::Debug for ZettaCacheIndex {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ZettaCacheIndex")
+            .field("last_key", &self.last_key)
+            .finish()
+    }
+}
+
 impl ZettaCacheIndex {
     pub async fn open(
         block_access: Arc<BlockAccess>,
         extent_allocator: Arc<ExtentAllocator>,
         phys: ZettaCacheIndexPhys,
     ) -> Self {
         Self {
+            last_key: phys.last_key,
             atime_histogram: phys.atime_histogram,
             log: BlockBasedLogWithSummary::open(block_access, extent_allocator, phys.log).await,
         }
     }
 
     pub async fn flush(&mut self) -> ZettaCacheIndexPhys {
         ZettaCacheIndexPhys {
+            last_key: self.last_key,
             atime_histogram: self.atime_histogram.clone(),
             log: self.log.flush().await,
         }
     }
 
+    /// Retrieve the index phys. This only works if there are no pending log entries.
+    /// Use flush() to retrieve the phys when there are pending entries.
     pub fn get_phys(&self) -> ZettaCacheIndexPhys {
         ZettaCacheIndexPhys {
+            last_key: self.last_key,
             atime_histogram: self.atime_histogram.clone(),
             log: self.log.get_phys(),
         }
     }
 
-    pub fn get_histogram_start(&self) -> Atime {
-        self.atime_histogram.get_start()
+    pub fn first_atime(&self) -> Atime {
+        self.atime_histogram.first()
+    }
+
+    pub fn update_last_key(&mut self, key: IndexKey) {
+        if let Some(last_key) = self.last_key {
+            assert_gt!(key, last_key);
+        }
+        self.last_key = Some(key);
     }
 
     pub fn append(&mut self, entry: IndexEntry) {
+        self.update_last_key(entry.key);
         self.atime_histogram.insert(entry.value);
         self.log.append(entry);
     }
 
     pub fn clear(&mut self) {
+        self.last_key = None;
         self.atime_histogram.clear();
         self.log.clear();
     }