Implement chunk eviction

cortes.server.block.chunk.toml

@@ -0,0 +1,27 @@
+log_level = "info"
+
+[server]
+socket = "/tmp/oxcache.sock"
+disk = "/dev/nvme1n1"
+writer_threads = 14
+reader_threads = 14
+chunk_size = 65536
+block_zone_capacity = 1129316352
+max_write_size = 262144
+#max_zones = 100
+
+[remote]
+remote_type = "emulated" # emulated | S3
+bucket = "S3_BUCKET"
+remote_artificial_delay_microsec = 40632
+
+[eviction]
+eviction_policy = "chunk"
+high_water_evict = 1 # Number remaining from end, evicts if reaches here
+low_water_evict = 3  # Evict until below mark
+eviction_interval = 1  # Evict every 1s
+
+[metrics]
+ip_addr = "127.0.0.1"
+port = 9000
+file_metrics_directory = "./logs"

cortes.server.block.promo.toml

@@ -0,0 +1,27 @@
+log_level = "info"
+
+[server]
+socket = "/tmp/oxcache.sock"
+disk = "/dev/nvme1n1"
+writer_threads = 14
+reader_threads = 14
+chunk_size = 65536
+block_zone_capacity = 1129316352
+max_write_size = 262144
+#max_zones = 100
+
+[remote]
+remote_type = "emulated" # emulated | S3
+bucket = "S3_BUCKET"
+remote_artificial_delay_microsec = 40632
+
+[eviction]
+eviction_policy = "promotional"
+high_water_evict = 1 # Number remaining from end, evicts if reaches here
+low_water_evict = 3  # Evict until below mark
+eviction_interval = 1  # Evict every 1s
+
+[metrics]
+ip_addr = "127.0.0.1"
+port = 9000
+file_metrics_directory = "./logs"

cortes.server.zns.chunk.toml

@@ -0,0 +1,29 @@
+log_level = "debug"
+
+[server]
+socket = "/tmp/oxcache.sock"
+disk = "/dev/nvme0n2"
+writer_threads = 14
+reader_threads = 14
+chunk_size = 536870912
+block_zone_capacity = 1129316352
+max_write_size = 262144
+max_zones = 20
+
+[remote]
+remote_type = "emulated" # emulated | S3
+bucket = "S3_BUCKET"
+remote_artificial_delay_microsec = 40632
+
+[eviction]
+eviction_policy = "chunk"
+high_water_evict = 16 # Number remaining from end, evicts if reaches here
+low_water_evict = 32  # Evict until below mark
+eviction_interval = 1  # Evict every 1s
+high_water_clean = 16 # Number remaining from end, cleans if reaches here (only used with chunk)
+low_water_clean = 8  # Clean until below mark (only used with chunk)
+
+[metrics]
+ip_addr = "127.0.0.1"
+port = 9000
+file_metrics_directory = "./logs"

cortes.server.zns.promo.toml

@@ -0,0 +1,27 @@
+log_level = "info"
+
+[server]
+socket = "/tmp/oxcache.sock"
+disk = "/dev/nvme0n2"
+writer_threads = 14
+reader_threads = 14
+chunk_size = 1048576
+block_zone_capacity = 1129316352
+max_write_size = 262144
+#max_zones = 100
+
+[remote]
+remote_type = "emulated" # emulated | S3
+bucket = "S3_BUCKET"
+remote_artificial_delay_microsec = 40632
+
+[eviction]
+eviction_policy = "promotional"
+high_water_evict = 1 # Number remaining from end, evicts if reaches here
+low_water_evict = 3  # Evict until below mark
+eviction_interval = 1  # Evict every 1s
+
+[metrics]
+ip_addr = "127.0.0.1"
+port = 9000
+file_metrics_directory = "./logs"

nvme/src/types.rs

@@ -83,7 +83,7 @@ pub struct ZNSConfig {
 
     pub chunks_per_zone: Chunk, // Number of chunks that can be allocated in a zone
     pub chunk_size_in_lbas: LogicalBlock, // This is in logical blocks
-    pub chunk_size_in_bytes: Byte, // This is in logical blocks
+    pub chunk_size_in_bytes: Byte, // This is in bytes
 
     pub num_zones: Zone,
 }

oxcache/src/bin/simpleevaluationclient.rs

@@ -35,7 +35,7 @@ const MAX_FRAME_LENGTH: usize = 2 * 1024 * 1024 * 1024; // 2 GB
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let nr_queries = 100000;
+    let nr_queries = 10000;
     let nr_uuids = 1000;
     let queries: Arc<Mutex<Vec<GetRequest>>> = Arc::new(Mutex::new(Vec::new()));
     let args = Cli::parse();

oxcache/src/cache/bucket.rs

@@ -1,6 +1,6 @@
 use crate::request::GetRequest;
 use nvme::types::{Byte, Zone};
-use std::sync::Arc;
+use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
 use tokio::sync::Notify;
 
 #[derive(Debug, Clone, Eq, PartialEq, Hash)]
@@ -26,6 +26,88 @@ impl ChunkLocation {
     }
 }
 
+/// A ChunkLocation with pin counting for coordinating with eviction
+#[derive(Debug)]
+pub struct PinnedChunkLocation {
+    pub location: ChunkLocation,
+    pub pin_count: AtomicUsize,
+    pub unpin_notify: Notify,
+}
+
+impl PinnedChunkLocation {
+    pub fn new(location: ChunkLocation) -> Self {
+        Self {
+            location,
+            pin_count: AtomicUsize::new(0),
+            unpin_notify: Notify::new(),
+        }
+    }
+
+    /// Pin this location (increment ref count). Returns a guard that will unpin on drop.
+    pub fn pin(self: &Arc<Self>) -> PinGuard {
+        let new_count = self.pin_count.fetch_add(1, Ordering::SeqCst) + 1;
+        tracing::debug!("Pinning location {:?}, pin_count now: {}", self.location, new_count);
+        PinGuard::new(self)
+    }
+
+    pub fn pin_count(&self) -> usize {
+        self.pin_count.load(Ordering::SeqCst)
+    }
+
+    /// Check if this location can be evicted (pin count is 0)
+    pub fn can_evict(&self) -> bool {
+        self.pin_count() == 0
+    }
+
+    /// Wait until this location can be evicted (pin count reaches 0)
+    pub async fn wait_for_unpin(&self) {
+        while !self.can_evict() {
+            let notified = self.unpin_notify.notified();
+            if self.can_evict() {
+                break;
+            }
+            notified.await;
+        }
+    }
+
+    fn unpin(&self) {
+        let old_count = self.pin_count.fetch_sub(1, Ordering::SeqCst);
+        let new_count = old_count - 1;
+        tracing::debug!("Unpinning location {:?}, pin_count now: {}", self.location, new_count);
+
+        // If pin count reached 0, notify any waiting eviction threads
+        if new_count == 0 {
+            self.unpin_notify.notify_waiters();
+        }
+    }
+}
+
+/// Guard that unpins a ChunkLocation when dropped
+#[derive(Debug)]
+pub struct PinGuard {
+    location: ChunkLocation,
+    pinned_location: Arc<PinnedChunkLocation>,
+}
+
+impl PinGuard {
+    fn new(pinned_location: &Arc<PinnedChunkLocation>) -> Self {
+        Self { 
+            location: pinned_location.location.clone(),
+            pinned_location: Arc::clone(pinned_location),
+        }
+    }
+
+    pub fn location(&self) -> &ChunkLocation {
+        &self.location
+    }
+}
+
+impl Drop for PinGuard {
+    fn drop(&mut self) {
+        self.pinned_location.unpin();
+    }
+}
+
 impl Chunk {
     pub fn new(uuid: String, offset: Byte, size: Byte) -> Self {
         Self { uuid, offset, size }
@@ -44,6 +126,6 @@ impl From<GetRequest> for Chunk {
 
 #[derive(Debug)]
 pub enum ChunkState {
-    Ready(Arc<ChunkLocation>),
+    Ready(Arc<PinnedChunkLocation>),
     Waiting(Arc<Notify>),
 }