From cab2ad858859dc1f14a3dc1cfc984d80c11b0301 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Thu, 11 Sep 2025 11:27:04 -0700 Subject: [PATCH 01/13] Parallel reads --- oxcache/src/device.rs | 43 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 37 insertions(+), 6 deletions(-) diff --git a/oxcache/src/device.rs b/oxcache/src/device.rs index 1c7d5cb..00cd426 100644 --- a/oxcache/src/device.rs +++ b/oxcache/src/device.rs @@ -525,14 +525,45 @@ impl Device for Zoned { let self_clone = self_clone.clone(); |items: Vec<(CacheKey, ChunkLocation)>| { async move { - // Increasing chunk index, might not be neccesary let mut items = items; items.sort_by_key(|(_, loc)| loc.index); - tracing::debug!("Reading zones"); - // Reads from location and returns the bytes - items.iter().map(|(key, loc)| { - Ok((key.clone(), self_clone.read(loc.clone())?)) - }).collect() + tracing::debug!("Reading {} chunks in parallel", items.len()); + + if items.is_empty() { + return Ok(Vec::new()); + } + + // Batch reads to avoid overwhelming the device + const BATCH_SIZE: usize = 16; + let mut all_results = Vec::with_capacity(items.len()); + + for chunk in items.chunks(BATCH_SIZE) { + let futures: Vec<_> = chunk.iter().map(|(key, loc)| { + let self_clone = self_clone.clone(); + let key = key.clone(); + let loc = loc.clone(); + + tokio::task::spawn_blocking(move || { + tracing::trace!("Reading chunk at {:?}", loc); + self_clone.read(loc.clone()).map(|bytes| (key, bytes)) + }) + }).collect(); + + let futures_len = futures.len(); + let batch_results: Result, _> = futures::future::join_all(futures) + .await + .into_iter() + .map(|join_result| { + join_result.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("Task join error: {}", e)))? + }) + .collect(); + + all_results.extend(batch_results?); + tracing::trace!("Completed batch of {} reads", futures_len); + } + + tracing::debug!("Completed parallel reading of {} chunks", all_results.len()); + Ok(all_results) } } }, From ce33f7516b63b5a1d555a5a6bb481bb5f36ba2b9 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Thu, 11 Sep 2025 20:49:28 -0700 Subject: [PATCH 02/13] Optimize LRU rebuild --- oxcache/src/eviction.rs | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 0f40fbe..521c876 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -249,20 +249,27 @@ impl EvictionPolicy for ChunkEvictionPolicy { fn get_clean_targets(&mut self) -> Self::CleanTarget { let mut clean_targets = self.pq.remove_if_thresh_met(); + if clean_targets.is_empty() { + return clean_targets; + } + clean_targets.sort_unstable(); - // Search in the LRU for items which exist in the clean - // targets. These are valid chunks, but must be removed - // because their location is invalidated when the zone is - // reset. - let new_lru_list = self.lru.iter().rev().filter(|lru_item| { - clean_targets.binary_search(&lru_item.0.zone).is_err() - }).map(|(k, _)| k.clone()).collect::>(); - - self.lru.clear(); - - for k in new_lru_list { - self.lru.put(k, ()); + let zones_to_clean: std::collections::HashSet = + clean_targets.iter().copied().collect(); + let mut items_to_reinsert = Vec::new(); + + // Keep those not in cleaned zones + while let Some((chunk_loc, _)) = self.lru.pop_lru() { + if !zones_to_clean.contains(&chunk_loc.zone) { + items_to_reinsert.push(chunk_loc); + } + } + + // Re-insert in reverse order to maintain LRU ordering + // (most recently used items go back in last) + for chunk_loc in items_to_reinsert.into_iter().rev() { + self.lru.put(chunk_loc, ()); } clean_targets From cb36b25be867e5485e88a273898c8482c556e126 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Fri, 12 Sep 2025 22:56:38 -0700 Subject: [PATCH 03/13] Use more granular eviction locking --- oxcache/src/device.rs | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/oxcache/src/device.rs b/oxcache/src/device.rs index 00cd426..d8f047e 100644 --- a/oxcache/src/device.rs +++ b/oxcache/src/device.rs @@ -271,7 +271,7 @@ impl Zoned { config.chunk_size_in_lbas = chunk_size_in_logical_blocks; config.chunk_size_in_bytes = chunk_size; let num_zones: Zone = config.num_zones; - + // Apply max_zones restriction if specified let restricted_num_zones = if let Some(max_zones) = max_zones { if max_zones > num_zones { @@ -284,7 +284,7 @@ impl Zoned { } else { num_zones }; - + let zone_list = ZoneList::new( restricted_num_zones, config.chunks_per_zone, @@ -528,27 +528,27 @@ impl Device for Zoned { let mut items = items; items.sort_by_key(|(_, loc)| loc.index); tracing::debug!("Reading {} chunks in parallel", items.len()); - + if items.is_empty() { return Ok(Vec::new()); } - + // Batch reads to avoid overwhelming the device const BATCH_SIZE: usize = 16; let mut all_results = Vec::with_capacity(items.len()); - + for chunk in items.chunks(BATCH_SIZE) { let futures: Vec<_> = chunk.iter().map(|(key, loc)| { let self_clone = self_clone.clone(); let key = key.clone(); let loc = loc.clone(); - + tokio::task::spawn_blocking(move || { tracing::trace!("Reading chunk at {:?}", loc); self_clone.read(loc.clone()).map(|bytes| (key, bytes)) }) }).collect(); - + let futures_len = futures.len(); let batch_results: Result, _> = futures::future::join_all(futures) .await @@ -557,11 +557,11 @@ impl Device for Zoned { join_result.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("Task join error: {}", e)))? }) .collect(); - + all_results.extend(batch_results?); tracing::trace!("Completed batch of {} reads", futures_len); } - + tracing::debug!("Completed parallel reading of {} chunks", all_results.len()); Ok(all_results) } @@ -576,7 +576,6 @@ impl Device for Zoned { // Writer callback |payloads: Vec<(CacheKey, bytes::Bytes)>| { async move { - { // Return zones back to the zone list and reset the zone let _guard = self_clone.zone_append_lock[zone as usize].write().unwrap(); let (zone_mtx, cv) = &*self_clone.zones; @@ -742,7 +741,7 @@ impl BlockInterface { // Num_zones let num_zones = nvme_config.total_size_in_bytes / block_zone_capacity; - + // Apply max_zones restriction if specified let restricted_num_zones = if let Some(max_zones) = max_zones { if max_zones > num_zones { @@ -755,7 +754,7 @@ impl BlockInterface { } else { num_zones }; - + // Chunks per zone let chunks_per_zone = block_zone_capacity / chunk_size; From 74400864e1a8babe777f71c55b0a57bef576c8da Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Sun, 14 Sep 2025 22:25:21 -0700 Subject: [PATCH 04/13] Temp --- Cargo.lock | 5 +- Cargo.toml | 3 + oxcache/Cargo.toml | 1 + oxcache/src/cache/mod.rs | 89 +++++++++++++----------- oxcache/src/device.rs | 95 +++++++++++++------------- oxcache/src/eviction.rs | 22 +++--- oxcache/src/server.rs | 1 + oxcache/src/writerpool.rs | 140 +++++++++++++++++++++++++++++++------- 8 files changed, 230 insertions(+), 126 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2476ced..470422b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2032,9 +2032,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" [[package]] name = "libc" -version = "0.2.172" +version = "0.2.175" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa" +checksum = "6a82ae493e598baaea5209805c49bbf2ea7de956d50d7da0da1164f9c6d28543" [[package]] name = "libloading" @@ -2376,6 +2376,7 @@ dependencies = [ "dashmap", "flume", "futures", + "libc", "libnvme-sys", "lru 0.16.0", "metrics", diff --git a/Cargo.toml b/Cargo.toml index ccfb831..bb76723 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,3 +4,6 @@ members = ["libnvme-sys", "nvme", "oxcache"] [profile.dev.package."*"] opt-level = 3 +strip = "none" + + diff --git a/oxcache/Cargo.toml b/oxcache/Cargo.toml index 6df91d1..5a5e0df 100644 --- a/oxcache/Cargo.toml +++ b/oxcache/Cargo.toml @@ -37,6 +37,7 @@ priority-queue = "2.5.0" tracing-appender = "0.2.3" tracing = "0.1.41" chrono = "0.4.41" +libc = "0.2.175" [[test]] name = "integration_tests" diff --git a/oxcache/src/cache/mod.rs b/oxcache/src/cache/mod.rs index 8356443..ed054d4 100644 --- a/oxcache/src/cache/mod.rs +++ b/oxcache/src/cache/mod.rs @@ -8,6 +8,8 @@ use std::{collections::HashMap, io}; use bytes::Bytes; use tokio::sync::{Notify, RwLock}; use crate::cache::bucket::Chunk as CacheKey; +use crate::writerpool::{WriterPool}; +use std::future::Future; use crate::server::validate_read_response; pub mod bucket; @@ -175,6 +177,7 @@ impl Cache { let mut reverse_mapping_guard = self.bm.write().await; reverse_mapping_guard.zone_to_entry[location.as_index()] = Some(key); + tracing::debug!("LRU_SYNC: Added chunk {:?} to bucket map reverse mapping", location); tracing::debug!("[map-dbg] location {:?} map updated", location); // tracing::debug!("[map-dbg] state is {:#?}", *reverse_mapping_guard); @@ -245,6 +248,7 @@ impl Cache { zone: Zone, reader: R, writer: W, + writer_pool: Arc, ) -> io::Result<()> where R: FnOnce(Vec<(CacheKey, ChunkLocation)>) -> RFut + Send, @@ -252,43 +256,43 @@ impl Cache { W: FnOnce(Vec<(CacheKey, Bytes)>) -> WFut + Send, WFut: Future>> + Send, { - use ndarray::s; // Reset the existing entries // Collect items and corresponding notifiers let (items, notifies) = { - // TODO: Can we deadlock here? let mut bm = self.bm.write().await; + let zone_slice = s![zone as usize, ..]; let mut out = Vec::new(); let mut notifies = Vec::new(); + let mut chunks_processed = 0; + let mut chunks_found = 0; + // Iterate through the entire list of chunks in the zone for opt_key in bm.zone_to_entry.slice(zone_slice).iter() { + chunks_processed += 1; // Collect only if Some if let Some(key) = opt_key.clone() { + chunks_found += 1; + let chunk_process_start = std::time::Instant::now(); let entry = bm .buckets .get(&key) .ok_or_else(|| io::Error::new(ErrorKind::NotFound, "Missing entry"))? .clone(); - + // Wait for pins to be released before proceeding loop { - tracing::debug!("EVICTION: Obtaining lock for entry {:?}", key); let mut st = entry.write().await; - tracing::debug!("EVICTION: Obtained lock for entry {:?}", key); let old_loc = match &*st { ChunkState::Ready(pinned_loc) => { // Wait for this location to be unpinned if !pinned_loc.can_evict() { let pinned_loc_clone = Arc::clone(pinned_loc); drop(st); // Release lock before waiting - tracing::warn!("EVICTION: Waiting for chunk at {:?} to be unpinned (pin_count={})", - pinned_loc_clone.location, pinned_loc_clone.pin_count()); + pinned_loc_clone.wait_for_unpin().await; - tracing::warn!("EVICTION: Chunk at {:?} unpinned (pin_count={}), retrying", - pinned_loc_clone.location, pinned_loc_clone.pin_count()); continue; // Retry after being notified } pinned_loc.location.clone() @@ -301,11 +305,10 @@ impl Cache { }; let notify = Arc::new(Notify::new()); - tracing::debug!("EVICTION: Setting entry {:?} to Waiting state", key); // Update state to waiting *st = ChunkState::Waiting(Arc::clone(¬ify)); drop(st); - tracing::debug!("EVICTION: Added entry {:?} to eviction list", key); + out.push((key, old_loc, entry)); notifies.push(notify); break; // Successfully processed this entry @@ -314,8 +317,10 @@ impl Cache { } // Clear old reverse slots + let reverse_clear_start = std::time::Instant::now(); for (key, old_loc, _) in &out { if bm.zone_to_entry[old_loc.as_index()].as_ref() == Some(key) { + // tracing::info!("LRU_SYNC: Removing chunk {:?} from bucket map reverse mapping (clean_zone_and_update_map)", old_loc); bm.zone_to_entry[old_loc.as_index()] = None; } } @@ -324,12 +329,16 @@ impl Cache { // Read the valid chunks from the zone // Buffer all chunks + let read_start = std::time::Instant::now(); let read_input: Vec<_> = items .iter() .map(|(k, l, _)| (k.clone(), l.clone())) .collect(); + let payloads = match reader(read_input).await { - Ok(p) => p, + Ok(p) => { + p + }, Err(e) => { // TODO: Should we bother? Probably still fatal // rollback @@ -344,48 +353,48 @@ impl Cache { } }; - // Write data out + // Write data out using reserved space let new_locs = writer(payloads).await?; // Update states & reverse map - { - let mut bm = self.bm.write().await; - - // Set Ready and reverse map - for (key, new_loc, b) in new_locs { - // set entry - if let Some((_, _, entry)) = items.iter().find(|(k, _, _)| *k == key) { - let mut st = entry.write().await; - *st = ChunkState::Ready(Arc::new(PinnedChunkLocation::new(new_loc.clone()))); - } else { - return Err(io::Error::new( - ErrorKind::NotFound, - format!("Missing entry for {:?}", key), - )); - } + for (key, new_loc, b) in &new_locs { + if let Some((_, _, entry)) = items.iter().find(|(k, _, _)| *k == *key) { + let mut st = entry.write().await; + *st = ChunkState::Ready(Arc::new(PinnedChunkLocation::new(new_loc.clone()))); + } else { + return Err(io::Error::new( + ErrorKind::NotFound, + format!("Missing entry for {:?}", key), + )); + } - let ch = key.clone(); + #[cfg(debug_assertions)] + validate_read_response(&b, &key.uuid, key.offset, key.size); + } + // Batch update reverse map (single bm lock) + { + let mut bm = self.bm.write().await; + for (key, new_loc, _) in new_locs { bm.zone_to_entry[new_loc.as_index()] = Some(key); - - validate_read_response(&b, &ch.uuid, ch.offset, ch.size); } } - // for loc in new_locs { - // validate_read_response() - // } - for n in notifies { n.notify_waiters(); } + Ok(()) } pub async fn remove_entries(&self, chunks: &[ChunkLocation]) -> tokio::io::Result<()> { + let thread_id = std::thread::current().id(); + tracing::info!("DEADLOCK_DEBUG: [Thread {:?}] remove_entries called for {} chunks", thread_id, chunks.len()); // to_relocate is a list of ChunkLocations that the caller wants to update // We pass in each chunk location and the writer function should return back with the list of updated chunk locations + tracing::info!("DEADLOCK_DEBUG: [Thread {:?}] Attempting to acquire BM write lock for remove_entries", thread_id); let mut bucket_guard = self.bm.write().await; + tracing::info!("DEADLOCK_DEBUG: [Thread {:?}] Acquired BM write lock for remove_entries", thread_id); // tracing::debug!("State is {:#?} before remove_entries", *bucket_guard); for chunk in chunks { @@ -393,9 +402,9 @@ impl Cache { let chunk_id = match &bucket_guard.zone_to_entry[chunk.as_index()] { Some(id) => id.clone(), None => { - // Oh, so this fails - tracing::error!("Couldn't find chunk {:?} in reverse map, state is {:?}", chunk, *bucket_guard); - return Err(io::Error::new(ErrorKind::NotFound, format!("Couldn't find chunk {:?} while removing entries", chunk))) + // This should not happen - indicates LRU/bucket map sync issue + tracing::error!("SYNC_BUG: Chunk {:?} was in LRU but not found in reverse map", chunk); + return Err(io::Error::new(ErrorKind::NotFound, format!("LRU/bucket map sync bug: chunk {:?} missing from reverse map", chunk))) }, }; @@ -431,10 +440,14 @@ impl Cache { // Now safe to remove from maps while holding entry write lock bucket_guard.zone_to_entry[chunk.as_index()].take(); let _removed_entry = bucket_guard.buckets.remove(&chunk_id); + tracing::info!("LRU_SYNC: Removed chunk {:?} from bucket map via remove_entries", chunk); tracing::debug!("Found chunk {:?} when removing entries", _removed_entry.is_some()); // entry_guard is dropped here, releasing the entry write lock } + tracing::info!("DEADLOCK_DEBUG: [Thread {:?}] Releasing BM write lock for remove_entries", thread_id); + // bucket_guard is dropped here + tracing::info!("DEADLOCK_DEBUG: [Thread {:?}] Completed remove_entries for {} chunks", thread_id, chunks.len()); Ok(()) } } diff --git a/oxcache/src/device.rs b/oxcache/src/device.rs index d8f047e..65a9fce 100644 --- a/oxcache/src/device.rs +++ b/oxcache/src/device.rs @@ -3,12 +3,11 @@ use crate::cache::Cache; use crate::cache::bucket::ChunkLocation; use crate::eviction::{EvictTarget, EvictorMessage}; use crate::server::RUNTIME; -use crate::writerpool::{WriteRequest, WriterPool}; +use crate::writerpool::{WriterPool, BatchWriteRequest}; use crate::zone_state::zone_list::{ZoneList, ZoneObtainFailure}; use aligned_vec::{AVec, Alignment, RuntimeAlign}; use bytes::Bytes; use flume::Sender; -use futures::future::join_all; use nvme::info::{get_active_zones, get_lba_at, is_zoned_device, nvme_get_info, report_zones_all}; use nvme::ops::{close_zone, finish_zone, reset_zone, zns_append}; use nvme::types::{Byte, Chunk, LogicalBlock, NVMeConfig, PerformOn, ZNSConfig, Zone, ZoneState}; @@ -19,6 +18,8 @@ use crate::metrics::{MetricType, METRICS}; use crate::zone_state::zone_priority_queue::ZonePriorityQueue; use crate::cache::bucket::Chunk as CacheKey; + +#[derive(Debug)] pub struct Zoned { nvme_config: NVMeConfig, config: ZNSConfig, @@ -109,15 +110,10 @@ pub trait Device: Send + Sync { "Unaligned read size" ); - // println!("Reading {} lbas, lba loc = {}, into ({}..{})", lbas_read, lba_loc, byte_ind, end); - - if let Err(err) = nvme::ops::read(nvme_config, lba_loc, &mut read_buffer[byte_ind..end]) - { + if let Err(err) = nvme::ops::read(nvme_config, lba_loc, &mut read_buffer[byte_ind..end]) { return Err(err.try_into().unwrap()); } - // println!("Read {} lbas, lba loc = {}, into ({}..{})", lbas_read, lba_loc, byte_ind, end); - byte_ind += chunk_size; lba_loc += lbas_read; } @@ -161,6 +157,7 @@ pub fn get_device( } fn trigger_eviction(eviction_channel: Sender) -> io::Result<()> { + tracing::info!("DEVICE: [Thread {:?}] Sending eviction trigger", std::thread::current().id()); let (resp_tx, resp_rx) = flume::bounded(1); if let Err(e) = eviction_channel.send(EvictorMessage { sender: resp_tx }) { tracing::error!("[append] Failed to send eviction message: {}", e); @@ -205,6 +202,7 @@ impl Zoned { /// Wrapper for ZoneList, handles mutex and notification fn get_free_zone(&self) -> io::Result { + tracing::info!("DEVICE: [Thread {:?}] get_free_zone called, checking zone availability", std::thread::current().id()); let (mtx, wait_notify) = &*self.zones; let mut zone_list = mtx.lock().unwrap(); @@ -246,6 +244,7 @@ impl Zoned { Ok(()) } + } impl Zoned { @@ -448,6 +447,7 @@ impl Zoned { } } + impl Device for Zoned { /// Hold internal state to keep track of zone state fn append(&self, data: Bytes) -> std::io::Result { @@ -455,10 +455,14 @@ impl Device for Zoned { let sz = data.len() as u64; let zone_index: Zone = loop { + tracing::info!("DEVICE: [Thread {:?}] Attempting to get free zone", std::thread::current().id()); match self.get_free_zone() { - Ok(res) => break res, + Ok(res) => { + tracing::info!("DEVICE: [Thread {:?}] Successfully got free zone: {:?}", std::thread::current().id(), res); + break res; + }, Err(err) => { - tracing::trace!("[append] Failed to get free zone: {}", err); + tracing::info!("DEVICE: [Thread {:?}] Failed to get free zone: {}, triggering eviction", std::thread::current().id(), err); } }; trigger_eviction(self.eviction_channel.clone())?; @@ -502,21 +506,19 @@ impl Device for Zoned { tracing::debug!("[evict:Chunk] No chunks evicted"); return Ok(()); } - tracing::debug!("[evict:Chunk] Evicting chunks {:?}", chunk_locations); // Remove from map (invalidation) RUNTIME.block_on(cache.remove_entries(&chunk_locations))?; // Cleaning let self_clone = self.clone(); - for zone in clean_locations { - // Spawn a task that asynchronously runs this function. + for zone in clean_locations.iter() { let cache_clone = cache.clone(); let self_clone = self_clone.clone(); let writer_pool = writer_pool.clone(); - RUNTIME.spawn( - async move { + + RUNTIME.block_on( cache_clone.clean_zone_and_update_map( zone.clone(), // Reads all valid chunks in zone and returns buffer [(Chunk, Bytes)] @@ -527,7 +529,6 @@ impl Device for Zoned { async move { let mut items = items; items.sort_by_key(|(_, loc)| loc.index); - tracing::debug!("Reading {} chunks in parallel", items.len()); if items.is_empty() { return Ok(Vec::new()); @@ -538,6 +539,7 @@ impl Device for Zoned { let mut all_results = Vec::with_capacity(items.len()); for chunk in items.chunks(BATCH_SIZE) { + let futures: Vec<_> = chunk.iter().map(|(key, loc)| { let self_clone = self_clone.clone(); let key = key.clone(); @@ -549,7 +551,6 @@ impl Device for Zoned { }) }).collect(); - let futures_len = futures.len(); let batch_results: Result, _> = futures::future::join_all(futures) .await .into_iter() @@ -559,10 +560,8 @@ impl Device for Zoned { .collect(); all_results.extend(batch_results?); - tracing::trace!("Completed batch of {} reads", futures_len); } - tracing::debug!("Completed parallel reading of {} chunks", all_results.len()); Ok(all_results) } } @@ -577,49 +576,51 @@ impl Device for Zoned { |payloads: Vec<(CacheKey, bytes::Bytes)>| { async move { { // Return zones back to the zone list and reset the zone - let _guard = self_clone.zone_append_lock[zone as usize].write().unwrap(); + let _guard = self_clone.zone_append_lock[*zone as usize].write().unwrap(); let (zone_mtx, cv) = &*self_clone.zones; let mut zones = zone_mtx.lock().unwrap(); - zones.reset_zone(zone, &*self_clone)?; + zones.reset_zone(*zone, &*self_clone)?; cv.notify_all(); } // Drop the mutex, so we don't have to put it in an await - // Queue valid chunks to be written back to, in the writer pool - let mut futures = Vec::with_capacity(payloads.len()); - for (key, data) in payloads { - let writer_pool = writer_pool.clone(); + // Use prioritized batch write for eviction + let keys: Vec<_> = payloads.iter().map(|(key, _)| key.clone()).collect(); + let data_vec: Vec<_> = payloads.iter().map(|(_, data)| data.clone()).collect(); - // Generate a vector of futures that send the data and then get their location in the cache - let (tx, rx) = flume::bounded(1); - futures.push(async move { + let (batch_tx, batch_rx) = flume::bounded(1); - writer_pool.send(WriteRequest{ - data: data.clone(), - responder: tx, - }).await?; + let batch_request = BatchWriteRequest { + data: data_vec, + responder: batch_tx, + }; - let location = rx.recv_async().await.map_err(|e| { - io::Error::new(io::ErrorKind::Other, - format!("failed to send write request: {}", e)) - }).and_then(|response| response.location)?; + writer_pool.send_priority_batch(batch_request).await?; - Ok::<(CacheKey, ChunkLocation, bytes::Bytes), io::Error>((key.clone(), location, data.clone())) - }); - } + let batch_response = batch_rx.recv_async().await.map_err(|e| { + io::Error::new(io::ErrorKind::Other, + format!("failed to receive batch write response: {}", e)) + })?; + + // Convert batch response back to individual results + let write_results: Result, io::Error> = + keys.into_iter() + .zip(batch_response.locations.into_iter()) + .zip(payloads.into_iter()) + .map(|((key, location_result), (_, data))| { + location_result.map(|loc| (key, loc, data)) + }) + .collect(); - // Await for the results all at the same time, so they can race - let write_results = join_all(futures).await.into_iter().collect::, io::Error>>()?; + let write_results = write_results?; Ok(write_results) // Vec<(Chunk, ChunkLocation)> } } }, - ).await - }); - - tracing::debug!("[evict:Chunk] Cleaned zone {}", zone); + writer_pool.clone(), + ) + )?; } - Ok(()) } EvictTarget::Zone(zones_to_evict) => { @@ -629,8 +630,6 @@ impl Device for Zoned { let mut zones = zone_mtx.lock().unwrap(); zones.reset_zones(&zones_to_evict, &*self)?; - tracing::debug!("Zones evicted: {:?}", zones_to_evict); - Ok(()) } } diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 521c876..fb60276 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -205,26 +205,21 @@ impl EvictionPolicy for ChunkEvictionPolicy { type Target = Vec; type CleanTarget = Vec; fn write_update(&mut self, chunk: ChunkLocation) { - tracing::debug!("Write LRU update at chunk {:?}", chunk); self.lru.put(chunk, ()); } fn read_update(&mut self, chunk: ChunkLocation) { - tracing::debug!("Read LRU update at chunk {:?}", chunk); - // assert!(self.lru.contains(&chunk)); // TODO: Race cond with chunk evict? if self.lru.contains(&chunk) { self.lru.put(chunk, ()); } } fn get_evict_targets(&mut self) -> Self::Target { - let span = tracing::debug_span!("get_evict_targets"); - let _enter = span.enter(); let lru_len = self.lru.len() as Chunk; let nr_chunks = self.nr_zones * self.nr_chunks_per_zone; let high_water_mark = nr_chunks - self.high_water; + if lru_len < high_water_mark { - tracing::debug!("Nothing to evict: lru_len is {} which is less than {}", lru_len, high_water_mark); return vec![]; } @@ -232,16 +227,13 @@ impl EvictionPolicy for ChunkEvictionPolicy { let cap = lru_len - low_water_mark; let mut targets = Vec::with_capacity(cap as usize); - tracing::debug!("targets:"); while self.lru.len() as Chunk >= low_water_mark { let targ = self.lru.pop_lru().unwrap().0; - tracing::debug!("{:?}", targ); let target_zone = targ.zone; targets.push(targ); // Adjust pq self.pq.modify_priority(target_zone, 1); - tracing::trace!("Increased priority for zone {}", target_zone); } targets @@ -252,20 +244,20 @@ impl EvictionPolicy for ChunkEvictionPolicy { if clean_targets.is_empty() { return clean_targets; } - + clean_targets.sort_unstable(); let zones_to_clean: std::collections::HashSet = clean_targets.iter().copied().collect(); let mut items_to_reinsert = Vec::new(); - + // Keep those not in cleaned zones while let Some((chunk_loc, _)) = self.lru.pop_lru() { if !zones_to_clean.contains(&chunk_loc.zone) { items_to_reinsert.push(chunk_loc); } } - + // Re-insert in reverse order to maintain LRU ordering // (most recently used items go back in last) for chunk_loc in items_to_reinsert.into_iter().rev() { @@ -321,6 +313,7 @@ impl Evictor { }; let mut policy = eviction_policy_clone.lock().unwrap(); + let targets = policy.get_evict_targets(); drop(policy); @@ -328,10 +321,11 @@ impl Evictor { let device_clone = device_clone.clone(); let result = match device_clone.evict(targets, cache_clone.clone(), writer_pool.clone()) { Err(e) => { - tracing::error!("Error evicting: {}", e); Err(e.to_string()) } - Ok(_) => Ok(()), + Ok(_) => { + Ok(()) + }, }; if let Some(sender) = sender { diff --git a/oxcache/src/server.rs b/oxcache/src/server.rs index 4e35833..76801d0 100644 --- a/oxcache/src/server.rs +++ b/oxcache/src/server.rs @@ -171,6 +171,7 @@ impl Server { self.config.writer_threads, Arc::clone(&self.device), &eviction_policy, + (self.device.get_num_zones() * self.device.get_chunks_per_zone()) as usize, )); let readerpool = Arc::new(ReaderPool::start( self.config.reader_threads, diff --git a/oxcache/src/writerpool.rs b/oxcache/src/writerpool.rs index df67da0..fb51d09 100644 --- a/oxcache/src/writerpool.rs +++ b/oxcache/src/writerpool.rs @@ -8,6 +8,7 @@ use flume::{Receiver, Sender, unbounded}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; use crate::metrics::{MetricType, METRICS}; +use std::sync::atomic::{AtomicUsize, Ordering}; #[derive(Debug)] pub struct WriteResponse { @@ -27,6 +28,17 @@ pub struct WriteRequestInternal { pub update_lru: bool, } +#[derive(Debug)] +pub struct BatchWriteRequest { + pub data: Vec, + pub responder: Sender, +} + +#[derive(Debug)] +pub struct BatchWriteResponse { + pub locations: Vec>, +} + fn request_update_lru(req: WriteRequest) -> WriteRequestInternal { WriteRequestInternal { data: req.data, responder: req.responder, update_lru: true } } @@ -40,6 +52,7 @@ struct Writer { device: Arc, id: usize, receiver: Receiver, + priority_receiver: Receiver, eviction: Arc>, } @@ -47,52 +60,106 @@ impl Writer { fn new( id: usize, receiver: Receiver, + priority_receiver: Receiver, device: Arc, eviction: &Arc>, ) -> Self { Self { id, receiver, + priority_receiver, device, eviction: eviction.clone(), } } fn run(self) { - tracing::debug!("Writer {} started", self.id); - while let Ok(msg) = self.receiver.recv() { - // println!("Writer {} processing: {:?}", self.id, msg); - - let start = std::time::Instant::now(); - - let result = self.device.append(msg.data).inspect(|loc| { - let mtx = Arc::clone(&self.eviction); + tracing::info!("Writer {} started", self.id); + loop { + // Prioritize batch requests (eviction) over regular requests + let batch_msg = self.priority_receiver.try_recv(); + if let Ok(batch_req) = batch_msg { + self.process_batch_request(batch_req); + continue; + } - if msg.update_lru { - let mut policy = mtx.lock().unwrap(); - policy.write_update(loc.clone()); + // If no priority request, handle regular requests with timeout + match self.receiver.recv_timeout(std::time::Duration::from_millis(10)) { + Ok(msg) => { + self.process_regular_request(msg); + } + Err(flume::RecvTimeoutError::Timeout) => { + // Check for batch requests again after timeout + continue; + } + Err(flume::RecvTimeoutError::Disconnected) => { + // Check if priority channel is also disconnected + if self.priority_receiver.is_disconnected() { + break; + } } - }); - METRICS.update_metric_histogram_latency("device_write_latency_ms", start.elapsed(), MetricType::MsLatency); - - let resp = WriteResponse { location: result }; - let snd = msg.responder.send(resp); - if snd.is_err() { - tracing::error!( - "Failed to send response from writer: {}", - snd.err().unwrap() - ); } } tracing::info!("Writer {} exiting", self.id); } + + fn process_regular_request(&self, msg: WriteRequestInternal) { + let start = std::time::Instant::now(); + + let result = self.device.append(msg.data).inspect(|loc| { + let mtx = Arc::clone(&self.eviction); + + if msg.update_lru { + let mut policy = mtx.lock().unwrap(); + policy.write_update(loc.clone()); + } + }); + METRICS.update_metric_histogram_latency("device_write_latency_ms", start.elapsed(), MetricType::MsLatency); + + let resp = WriteResponse { location: result }; + let snd = msg.responder.send(resp); + if snd.is_err() { + tracing::error!( + "Failed to send response from writer: {}", + snd.err().unwrap() + ); + } + } + + fn process_batch_request(&self, batch_req: BatchWriteRequest) { + let data_len = batch_req.data.len(); // Store length before moving + let mut locations = Vec::with_capacity(data_len); + + for data in batch_req.data.into_iter() { + + let result = self.device.append(data); + + // CRITICAL: Update LRU for batch writes (eviction writes) + if let Ok(ref loc) = result { + let mtx = Arc::clone(&self.eviction); + let mut policy = mtx.lock().unwrap(); + policy.write_update(loc.clone()); + drop(policy); + } + + locations.push(result); + } + + let resp = BatchWriteResponse { locations }; + let snd = batch_req.responder.send(resp); + if snd.is_err() { + tracing::error!("Failed to send batch response from writer"); + } + } } /// Pool of writer threads sharing a single receiver #[derive(Debug)] pub struct WriterPool { sender: Sender, + priority_sender: Sender, handles: Vec>, + max_capacity: usize, } impl WriterPool { @@ -101,22 +168,33 @@ impl WriterPool { num_writers: usize, device: Arc, eviction_policy: &Arc>, + max_capacity: usize, ) -> Self { let (sender, receiver): (Sender, Receiver) = unbounded(); + let (priority_sender, priority_receiver): (Sender, Receiver) = unbounded(); let mut handles = Vec::with_capacity(num_writers); for id in 0..num_writers { let rx_clone = receiver.clone(); - let writer = Writer::new(id, rx_clone, device.clone(), eviction_policy); + let priority_rx_clone = priority_receiver.clone(); + let writer = Writer::new(id, rx_clone, priority_rx_clone, device.clone(), eviction_policy); let handle = thread::spawn(move || writer.run()); handles.push(handle); } - Self { sender, handles } + Self { + sender, + priority_sender, + handles, + max_capacity, + } } /// Send a message to the writer pool pub async fn send(&self, message: WriteRequest) -> std::io::Result<()> { + // For now, bypass capacity checking for regular writes since the semaphore + // approach was causing issues. The reservation system is primarily for + // preventing eviction deadlocks, not for general capacity management. self.sender.send_async(request_update_lru(message)).await.map_err(|e| { std::io::Error::new( std::io::ErrorKind::Other, @@ -126,6 +204,9 @@ impl WriterPool { } pub async fn send_no_update_lru(&self, message: WriteRequest) -> std::io::Result<()> { + // For now, bypass capacity checking for regular writes since the semaphore + // approach was causing issues. The reservation system is primarily for + // preventing eviction deadlocks, not for general capacity management. self.sender.send_async(request_no_update_lru(message)).await.map_err(|e| { std::io::Error::new( std::io::ErrorKind::Other, @@ -134,9 +215,20 @@ impl WriterPool { }) } + /// Send a prioritized batch request for eviction writes + pub async fn send_priority_batch(&self, message: BatchWriteRequest) -> std::io::Result<()> { + self.priority_sender.send_async(message).await.map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::Other, + format!("WriterPool::send_priority_batch failed: {}", e), + ) + }) + } + /// Stop the pool and wait for all writer threads to finish. pub fn stop(self) { - drop(self.sender); // Close the channel + drop(self.sender); // Close the regular channel + drop(self.priority_sender); // Close the priority channel for handle in self.handles { if let Err(e) = handle.join() { // A panic occurred — e is a Box From 25b96ba9798148347b6f89f3391ea8e10c63686f Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Tue, 16 Sep 2025 21:21:45 -0700 Subject: [PATCH 05/13] Add dedicated eviction thread --- oxcache/src/device.rs | 4 ++++ oxcache/src/server.rs | 1 - oxcache/src/writerpool.rs | 35 ++++++++++++++++++++++++++--------- 3 files changed, 30 insertions(+), 10 deletions(-) diff --git a/oxcache/src/device.rs b/oxcache/src/device.rs index 65a9fce..292f13a 100644 --- a/oxcache/src/device.rs +++ b/oxcache/src/device.rs @@ -14,6 +14,7 @@ use nvme::types::{Byte, Chunk, LogicalBlock, NVMeConfig, PerformOn, ZNSConfig, Z use std::io::ErrorKind; use std::os::fd::RawFd; use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock}; +use std::time::Duration; use crate::metrics::{MetricType, METRICS}; use crate::zone_state::zone_priority_queue::ZonePriorityQueue; use crate::cache::bucket::Chunk as CacheKey; @@ -587,6 +588,9 @@ impl Device for Zoned { let keys: Vec<_> = payloads.iter().map(|(key, _)| key.clone()).collect(); let data_vec: Vec<_> = payloads.iter().map(|(_, data)| data.clone()).collect(); + // Used to verify no RACE, TODO: Remove! + tokio::time::sleep(Duration::from_secs(5)).await; + let (batch_tx, batch_rx) = flume::bounded(1); let batch_request = BatchWriteRequest { diff --git a/oxcache/src/server.rs b/oxcache/src/server.rs index 76801d0..4e35833 100644 --- a/oxcache/src/server.rs +++ b/oxcache/src/server.rs @@ -171,7 +171,6 @@ impl Server { self.config.writer_threads, Arc::clone(&self.device), &eviction_policy, - (self.device.get_num_zones() * self.device.get_chunks_per_zone()) as usize, )); let readerpool = Arc::new(ReaderPool::start( self.config.reader_threads, diff --git a/oxcache/src/writerpool.rs b/oxcache/src/writerpool.rs index fb51d09..fa5c18a 100644 --- a/oxcache/src/writerpool.rs +++ b/oxcache/src/writerpool.rs @@ -54,6 +54,7 @@ struct Writer { receiver: Receiver, priority_receiver: Receiver, eviction: Arc>, + priority_only: bool } impl Writer { @@ -63,6 +64,7 @@ impl Writer { priority_receiver: Receiver, device: Arc, eviction: &Arc>, + priority_only: bool ) -> Self { Self { id, @@ -70,18 +72,18 @@ impl Writer { priority_receiver, device, eviction: eviction.clone(), + priority_only } } - fn run(self) { - tracing::info!("Writer {} started", self.id); + fn receive_all(&self) { loop { // Prioritize batch requests (eviction) over regular requests let batch_msg = self.priority_receiver.try_recv(); if let Ok(batch_req) = batch_msg { self.process_batch_request(batch_req); continue; - } + }; // If no priority request, handle regular requests with timeout match self.receiver.recv_timeout(std::time::Duration::from_millis(10)) { @@ -100,6 +102,20 @@ impl Writer { } } } + } + fn receive_priority(&self) { + while let Ok(batch_msg) = self.priority_receiver.recv() { + self.process_batch_request(batch_msg); + } + } + + fn run(self) { + tracing::info!("Writer {} started", self.id); + if self.priority_only { + self.receive_priority(); + } else { + self.receive_all(); + } tracing::info!("Writer {} exiting", self.id); } @@ -134,12 +150,13 @@ impl Writer { let result = self.device.append(data); - // CRITICAL: Update LRU for batch writes (eviction writes) if let Ok(ref loc) = result { let mtx = Arc::clone(&self.eviction); let mut policy = mtx.lock().unwrap(); policy.write_update(loc.clone()); drop(policy); + } else { + tracing::error!("Failed to append: {:?}", result); } locations.push(result); @@ -159,7 +176,6 @@ pub struct WriterPool { sender: Sender, priority_sender: Sender, handles: Vec>, - max_capacity: usize, } impl WriterPool { @@ -168,16 +184,18 @@ impl WriterPool { num_writers: usize, device: Arc, eviction_policy: &Arc>, - max_capacity: usize, ) -> Self { let (sender, receiver): (Sender, Receiver) = unbounded(); let (priority_sender, priority_receiver): (Sender, Receiver) = unbounded(); let mut handles = Vec::with_capacity(num_writers); - for id in 0..num_writers { + // Regular writers + for id in 0..=num_writers { let rx_clone = receiver.clone(); let priority_rx_clone = priority_receiver.clone(); - let writer = Writer::new(id, rx_clone, priority_rx_clone, device.clone(), eviction_policy); + + // Will create ONE priority writer (last) via id == num_writers + let writer = Writer::new(id, rx_clone, priority_rx_clone, device.clone(), eviction_policy, id == num_writers); let handle = thread::spawn(move || writer.run()); handles.push(handle); } @@ -186,7 +204,6 @@ impl WriterPool { sender, priority_sender, handles, - max_capacity, } } From f7dbb2f811f1df387816e1f812de90a86e7d0e13 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Fri, 19 Sep 2025 20:29:14 -0700 Subject: [PATCH 06/13] Cleanup and add eviction bypass --- cortes.server.block.chunk.toml | 2 +- cortes.server.block.promo.toml | 2 +- cortes.server.block.toml | 2 +- cortes.server.zns.chunk.toml | 2 +- cortes.server.zns.promo.toml | 2 +- cortes.server.zns.toml | 2 +- example.server.toml | 2 +- github_runner.server.toml | 2 +- local.server.toml | 2 +- oxcache/src/cache/mod.rs | 3 - oxcache/src/cli.rs | 8 +-- oxcache/src/device.rs | 97 +++++++++++------------------ oxcache/src/eviction.rs | 5 ++ oxcache/src/server.rs | 2 +- oxcache/src/writerpool.rs | 3 +- oxcache/src/zone_state/zone_list.rs | 31 +++++++-- oxcache/tests/mock_device.rs | 38 ++--------- qemu.block.promo.toml | 2 +- qemu.block.toml | 2 +- qemu.zns.promo.toml | 2 +- qemu.zns.toml | 2 +- 21 files changed, 92 insertions(+), 121 deletions(-) diff --git a/cortes.server.block.chunk.toml b/cortes.server.block.chunk.toml index d104f37..8544f3a 100644 --- a/cortes.server.block.chunk.toml +++ b/cortes.server.block.chunk.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/cortes.server.block.promo.toml b/cortes.server.block.promo.toml index 5f360a0..296202c 100644 --- a/cortes.server.block.promo.toml +++ b/cortes.server.block.promo.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/cortes.server.block.toml b/cortes.server.block.toml index 5f360a0..296202c 100644 --- a/cortes.server.block.toml +++ b/cortes.server.block.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/cortes.server.zns.chunk.toml b/cortes.server.zns.chunk.toml index 0e1a926..41d25a2 100644 --- a/cortes.server.zns.chunk.toml +++ b/cortes.server.zns.chunk.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # 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) diff --git a/cortes.server.zns.promo.toml b/cortes.server.zns.promo.toml index f97d55f..8d2da01 100644 --- a/cortes.server.zns.promo.toml +++ b/cortes.server.zns.promo.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/cortes.server.zns.toml b/cortes.server.zns.toml index f97d55f..8d2da01 100644 --- a/cortes.server.zns.toml +++ b/cortes.server.zns.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/example.server.toml b/example.server.toml index 7cb4015..9272fd4 100644 --- a/example.server.toml +++ b/example.server.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 eviction_policy = "promotional" high_water_evict = 5 # Number remaining from end, evicts if reaches here low_water_evict = 7 # Evict until below mark -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/github_runner.server.toml b/github_runner.server.toml index ef4f4e2..b30eeda 100644 --- a/github_runner.server.toml +++ b/github_runner.server.toml @@ -16,6 +16,6 @@ remote_artificial_delay_microsec = 40632 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 +eviction_interval_ms = 1000 # Evict every 1s [metrics] \ No newline at end of file diff --git a/local.server.toml b/local.server.toml index 1598a23..7a19d21 100644 --- a/local.server.toml +++ b/local.server.toml @@ -18,7 +18,7 @@ remote_artificial_delay_microsec = 40632 eviction_policy = "promotional" high_water_evict = 5 # Number remaining from end, evicts if reaches here low_water_evict = 7 # Evict until below mark -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] ip_addr = "127.0.0.1" diff --git a/oxcache/src/cache/mod.rs b/oxcache/src/cache/mod.rs index ed054d4..f5ed74e 100644 --- a/oxcache/src/cache/mod.rs +++ b/oxcache/src/cache/mod.rs @@ -275,7 +275,6 @@ impl Cache { // Collect only if Some if let Some(key) = opt_key.clone() { chunks_found += 1; - let chunk_process_start = std::time::Instant::now(); let entry = bm .buckets .get(&key) @@ -317,7 +316,6 @@ impl Cache { } // Clear old reverse slots - let reverse_clear_start = std::time::Instant::now(); for (key, old_loc, _) in &out { if bm.zone_to_entry[old_loc.as_index()].as_ref() == Some(key) { // tracing::info!("LRU_SYNC: Removing chunk {:?} from bucket map reverse mapping (clean_zone_and_update_map)", old_loc); @@ -329,7 +327,6 @@ impl Cache { // Read the valid chunks from the zone // Buffer all chunks - let read_start = std::time::Instant::now(); let read_input: Vec<_> = items .iter() .map(|(k, l, _)| (k.clone(), l.clone())) diff --git a/oxcache/src/cli.rs b/oxcache/src/cli.rs index f32638c..6879d98 100644 --- a/oxcache/src/cli.rs +++ b/oxcache/src/cli.rs @@ -57,7 +57,7 @@ pub struct CliArgs { pub block_zone_capacity: Option, #[arg(long)] - pub eviction_interval: Option, + pub eviction_interval_ms: Option, #[arg(long)] pub remote_artificial_delay_microsec: Option, @@ -106,7 +106,7 @@ pub struct ParsedEvictionConfig { pub low_water_evict: Option, pub high_water_clean: Option, pub low_water_clean: Option, - pub eviction_interval: Option, + pub eviction_interval_ms: Option, } #[derive(Debug, Deserialize)] @@ -226,9 +226,9 @@ pub fn load_config(cli: &CliArgs) -> Result std::io::Result; + fn append(&self, data: Bytes) -> std::io::Result { + self.append_with_eviction_bypass(data, false) + } + + fn append_with_eviction_bypass(&self, data: Bytes, is_eviction: bool) -> std::io::Result; /// This is expected to remove elements from the cache as well fn evict(self: Arc, locations: EvictTarget, cache: Arc, writer_pool: Arc) -> io::Result<()>; @@ -168,48 +172,36 @@ fn trigger_eviction(eviction_channel: Sender) -> io::Result<()> )); }; - if let Err(e) = resp_rx.recv() { - tracing::error!("[append] Failed to receive eviction message: {}", e); + match resp_rx.recv() { + Ok(result) => { + match result { + Ok(_) => { + tracing::debug!("DEVICE: [Thread {:?}] Eviction completed successfully", std::thread::current().id()); + } + Err(e) => { + tracing::error!("DEVICE: [Thread {:?}] Eviction failed: {}", std::thread::current().id(), e); + return Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Eviction failed: {}", e))); + } + } + } + Err(e) => { + tracing::error!("DEVICE: [Thread {:?}] Failed to receive eviction response: {}", std::thread::current().id(), e); + return Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Failed to receive eviction response: {}", e))); + } } Ok(()) } impl Zoned { - fn compact_zone( - &self, - zone_to_compact: Zone, - chunks_to_keep: &[ChunkLocation], - buffer: &mut [u8], - ) -> io::Result> { - let mut new_locations = Vec::with_capacity(chunks_to_keep.len()); - for chunk in chunks_to_keep { - let starting_byte_loc: Byte = - self.config.chunks_to_bytes(&self.nvme_config, chunk.index); - let ending_byte_loc: Byte = self - .config - .chunks_to_bytes(&self.nvme_config, chunk.index + 1); - let new_idx = zns_append( - &self.nvme_config, - &self.config, - zone_to_compact, - &mut buffer[starting_byte_loc as usize..ending_byte_loc as usize], - ) - .map_err(|err| std::io::Error::new(ErrorKind::Other, err.to_string()))?; - new_locations.push(ChunkLocation::new(zone_to_compact, new_idx)); - } - Ok(new_locations) - } - /// Wrapper for ZoneList, handles mutex and notification - fn get_free_zone(&self) -> io::Result { - tracing::info!("DEVICE: [Thread {:?}] get_free_zone called, checking zone availability", std::thread::current().id()); + fn get_free_zone(&self, is_eviction: bool) -> io::Result { let (mtx, wait_notify) = &*self.zones; let mut zone_list = mtx.lock().unwrap(); debug_assert!(get_active_zones(self.nvme_config.fd, self.nvme_config.nsid).unwrap() <= self.config.max_active_resources as usize); - match zone_list.remove() { + match zone_list.remove_with_eviction_bypass(is_eviction) { Ok(zone_idx) => Ok(zone_idx), Err(error) => match error { ZoneObtainFailure::EvictNow => { @@ -217,7 +209,7 @@ impl Zoned { } ZoneObtainFailure::Wait => loop { zone_list = wait_notify.wait(zone_list).unwrap(); - match zone_list.remove() { + match zone_list.remove_with_eviction_bypass(is_eviction) { Ok(idx) => return Ok(idx), Err(err) => match err { ZoneObtainFailure::EvictNow => { @@ -405,25 +397,6 @@ impl Zoned { )) .try_into() .unwrap()); - - // return match self.complete_write(zone_index, false) { - // Ok(()) => Err(err - // .add_context(format!("Write failed at zone {}\n", zone_index)) - // .try_into() - // .unwrap()), - // Err(err2) => Err(err - // .add_context(format!("Write failed at zone {}", zone_index)) - // .add_context(format!("Zone state: {:#?}", { - // let (_nz, state) = report_zones_all(self.nvme_config.fd, self.nvme_config.nsid).unwrap(); - // state.iter().map(|state|{ - // state.zone_state.clone() - // }).collect::>() - // })) - // .add_context(format!("Zone list state:\n{:#?}", self.zones.0.lock().unwrap())) - // .add_context(format!("Additional failure while trying to handle error: {}\n", err2.to_string())) - // .try_into() - // .unwrap()), - // }; } } byte_ind += write_sz; @@ -451,21 +424,24 @@ impl Zoned { impl Device for Zoned { /// Hold internal state to keep track of zone state - fn append(&self, data: Bytes) -> std::io::Result { - - let sz = data.len() as u64; + fn append_with_eviction_bypass(&self, data: Bytes, is_eviction: bool) -> std::io::Result { let zone_index: Zone = loop { - tracing::info!("DEVICE: [Thread {:?}] Attempting to get free zone", std::thread::current().id()); - match self.get_free_zone() { + match self.get_free_zone(is_eviction) { Ok(res) => { - tracing::info!("DEVICE: [Thread {:?}] Successfully got free zone: {:?}", std::thread::current().id(), res); break res; }, Err(err) => { - tracing::info!("DEVICE: [Thread {:?}] Failed to get free zone: {}, triggering eviction", std::thread::current().id(), err); + if is_eviction { + // If eviction itself can't get a zone, we're truly stuck + tracing::error!("DEVICE: [Thread {:?}] Eviction failed to get free zone: {}", std::thread::current().id(), err); + return Err(err); + } + tracing::debug!("DEVICE: [Thread {:?}] Failed to get free zone: {}, triggering eviction", std::thread::current().id(), err); } }; + // Add a small delay to prevent eviction spam + std::thread::sleep(std::time::Duration::from_millis(10)); trigger_eviction(self.eviction_channel.clone())?; }; @@ -589,7 +565,7 @@ impl Device for Zoned { let data_vec: Vec<_> = payloads.iter().map(|(_, data)| data.clone()).collect(); // Used to verify no RACE, TODO: Remove! - tokio::time::sleep(Duration::from_secs(5)).await; + // tokio::time::sleep(Duration::from_secs(5)).await; let (batch_tx, batch_rx) = flume::bounded(1); @@ -830,7 +806,8 @@ impl BlockInterface { impl Device for BlockInterface { /// Hold internal state to keep track of "ssd" zone state - fn append(&self, data: Bytes) -> std::io::Result { + fn append_with_eviction_bypass(&self, data: Bytes, _is_eviction: bool) -> std::io::Result { + // Block devices don't need eviction bypass logic let sz = data.len() as u64; let mtx = self.state.clone(); diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index fb60276..f9185e5 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -190,6 +190,11 @@ impl ChunkEvictionPolicy { nr_zones: Zone, nr_chunks_per_zone: Chunk, ) -> Self { + assert!( + high_water > nr_chunks_per_zone, + "high_water={} must be larget than nr_chunks_per_zone={} to leave room for eviction", + high_water, nr_chunks_per_zone + ); Self { high_water, low_water, diff --git a/oxcache/src/server.rs b/oxcache/src/server.rs index 4e35833..2145926 100644 --- a/oxcache/src/server.rs +++ b/oxcache/src/server.rs @@ -182,7 +182,7 @@ impl Server { Arc::clone(&self.device), Arc::clone(&eviction_policy), Arc::clone(&self.cache), - Duration::from_secs(self.config.eviction.eviction_interval), + Duration::from_millis(self.config.eviction.eviction_interval), self.evict_rx.clone(), writerpool.clone() )?; diff --git a/oxcache/src/writerpool.rs b/oxcache/src/writerpool.rs index fa5c18a..9d624e8 100644 --- a/oxcache/src/writerpool.rs +++ b/oxcache/src/writerpool.rs @@ -148,7 +148,8 @@ impl Writer { for data in batch_req.data.into_iter() { - let result = self.device.append(data); + // Use eviction bypass for priority batch requests (eviction writes) + let result = self.device.append_with_eviction_bypass(data, true); if let Ok(ref loc) = result { let mtx = Arc::clone(&self.eviction); diff --git a/oxcache/src/zone_state/zone_list.rs b/oxcache/src/zone_state/zone_list.rs index 4931ea1..e6913a2 100644 --- a/oxcache/src/zone_state/zone_list.rs +++ b/oxcache/src/zone_state/zone_list.rs @@ -207,7 +207,7 @@ impl ZoneList { open_zones: VecDeque::with_capacity(max_active_resources), writing_zones: HashMap::with_capacity(max_active_resources), chunks_per_zone, - max_active_resources: max_active_resources-1, // Keep one reserved for eviction + max_active_resources: max_active_resources, zones, #[cfg(debug_assertions)] state_tracker: ZoneStateTracker::new( @@ -220,12 +220,26 @@ impl ZoneList { // Get a zone to write to pub fn remove(&mut self) -> Result { + self.remove_with_eviction_bypass(false) + } + + // Get a zone to write to, with option to bypass eviction check for eviction operations + pub fn remove_with_eviction_bypass(&mut self, is_eviction: bool) -> Result { #[cfg(debug_assertions)] self.check_invariants(); - if self.is_full() { - // Need to evict - tracing::debug!("Full, need to evict now"); + let remaining_zones = self.free_zones.len() + self.open_zones.len(); + tracing::debug!("remove_with_eviction_bypass: is_eviction={}, remaining_zones={}", is_eviction, remaining_zones); + + if !is_eviction && self.should_evict() { + // Need to evict before we run out completely (but not during eviction itself) + tracing::debug!("Low on zones, need to evict now"); + return Err(EvictNow); + } + + if is_eviction && self.is_full() { + // Even eviction can't proceed if completely full + tracing::debug!("Completely full, even eviction cannot proceed"); return Err(EvictNow); } @@ -443,6 +457,13 @@ impl ZoneList { self.free_zones.is_empty() && self.open_zones.is_empty() } + // Check if we should trigger eviction (before completely full) + pub fn should_evict(&self) -> bool { + // Trigger eviction when we have 1 or fewer zones left + let remaining_zones = self.free_zones.len() + self.open_zones.len(); + remaining_zones <= 1 + } + // Gets the number of open zones by counting the unique // zones listed in open_zones and writing_zones pub fn get_open_zones(&self) -> usize { @@ -609,7 +630,7 @@ mod zone_list_tests { struct MockDevice {} impl Device for MockDevice { - fn append(&self, _data: Bytes) -> std::io::Result { + fn append_with_eviction_bypass(&self, _data: Bytes, _is_eviction: bool) -> std::io::Result { Ok(ChunkLocation { zone: 0, index: 0 }) } diff --git a/oxcache/tests/mock_device.rs b/oxcache/tests/mock_device.rs index d7b0cec..8e28827 100644 --- a/oxcache/tests/mock_device.rs +++ b/oxcache/tests/mock_device.rs @@ -114,43 +114,13 @@ impl MockZonedDevice { /// Does not actually write data. Only validates that the state of things are correct impl Device for MockZonedDevice { fn append(&self, data: Bytes) -> std::io::Result { - // let zone_index = loop { - // match self.get_free_zone() { - // Ok(res) => break res, - // Err(err) => { - // eprintln!("[append] Failed to get free zone: {}", err); - // } - // }; - // trigger_eviction(self.eviction_channel.clone())?; - // }; - // // Note: this performs a copy every time because we need to - // // pass in a mutable vector to libnvme - // assert_eq!( - // data.as_ptr() as usize % self.nvme_config.logical_block_size as usize, - // 0 - // ); - - // match zns_append( - // &self.nvme_config, - // &self.config, - // zone_index as u64, - // data.as_ref(), - // ) { - // Ok(lba) => { - // self.complete_write(zone_index)?; - // let chunk = lba / self.config.chunk_size as u64; - // Ok(ChunkLocation::new(zone_index, chunk)) - // } - // Err(mut err) => { - // self.complete_write(zone_index)?; - // err.add_context(format!("Write failed at zone {}\n", zone_index)); - // Err(err.try_into().unwrap()) - // } - // } - Ok(ChunkLocation { zone: 1, index: 2 }) } + fn append_with_eviction_bypass(&self, data: bytes::Bytes, _: bool) -> Result { + self.append(data) + } + fn read_into_buffer( &self, _max_write_size: Byte, diff --git a/qemu.block.promo.toml b/qemu.block.promo.toml index 6a6186d..e58034a 100644 --- a/qemu.block.promo.toml +++ b/qemu.block.promo.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 eviction_policy = "promotional" high_water_evict = 5 # Number remaining from end, evicts if reaches here low_water_evict = 25 # Evict until below mark -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] file_metrics_directory = "./logs" \ No newline at end of file diff --git a/qemu.block.toml b/qemu.block.toml index 6eb596b..0e42434 100644 --- a/qemu.block.toml +++ b/qemu.block.toml @@ -21,7 +21,7 @@ high_water_evict = 50 # Number remaining from end, evicts if reaches here low_water_evict = 100 # Evict until below mark high_water_clean = 40 # Number remaining from end, cleans if reaches here (only used with chunk) low_water_clean = 20 # Clean until below mark (only used with chunk) -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] file_metrics_directory = "./logs" \ No newline at end of file diff --git a/qemu.zns.promo.toml b/qemu.zns.promo.toml index 7b3394d..eff574f 100644 --- a/qemu.zns.promo.toml +++ b/qemu.zns.promo.toml @@ -19,7 +19,7 @@ remote_artificial_delay_microsec = 40632 eviction_policy = "promotional" high_water_evict = 5 # Number remaining from end, evicts if reaches here low_water_evict = 25 # Evict until below mark -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] file_metrics_directory = "./logs" diff --git a/qemu.zns.toml b/qemu.zns.toml index 74b7159..09efdda 100644 --- a/qemu.zns.toml +++ b/qemu.zns.toml @@ -21,7 +21,7 @@ high_water_evict = 100 # Number remaining from end, evicts if reaches here low_water_evict = 150 # Evict until below mark high_water_clean = 40 # Number of invalids when we begin cleaning low_water_clean = 20 # Clean until below mark (only used with chunk) -eviction_interval = 1 # Evict every 1s +eviction_interval_ms = 1000 # Evict every 1s [metrics] file_metrics_directory = "./logs" From 4a78b955fc0b4490b234e991b0c9b7c263e352e5 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Sat, 20 Sep 2025 22:45:11 -0700 Subject: [PATCH 07/13] Add eviction test params --- vendor/workloadgen | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/workloadgen b/vendor/workloadgen index 55905e8..dd3956c 160000 --- a/vendor/workloadgen +++ b/vendor/workloadgen @@ -1 +1 @@ -Subproject commit 55905e86e6c2364f4184056c66b15d1c4ff710c1 +Subproject commit dd3956c577cd13916357c12e66324e0ebcce0695 From 3f0c893393f808133813fc42b7ef215078309216 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Sun, 21 Sep 2025 13:15:31 -0700 Subject: [PATCH 08/13] Remove high water clean Adds unneccesary complexity --- Cargo.toml | 6 +- cortes.server.zns.chunk.toml | 4 +- oxcache/src/cli.rs | 21 +- oxcache/src/device.rs | 23 +- oxcache/src/eviction.rs | 303 +++++++++--------- oxcache/src/server.rs | 2 - oxcache/src/zone_state/zone_list.rs | 6 +- oxcache/src/zone_state/zone_priority_queue.rs | 9 +- oxcache/tests/mock_device.rs | 5 +- 9 files changed, 180 insertions(+), 199 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index bb76723..3a9eaf7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,8 +2,10 @@ resolver = "3" members = ["libnvme-sys", "nvme", "oxcache"] +[profile.dev] +debug-assertions = false + [profile.dev.package."*"] opt-level = 3 strip = "none" - - +debug-assertions = false diff --git a/cortes.server.zns.chunk.toml b/cortes.server.zns.chunk.toml index 41d25a2..8a473f1 100644 --- a/cortes.server.zns.chunk.toml +++ b/cortes.server.zns.chunk.toml @@ -19,11 +19,11 @@ remote_artificial_delay_microsec = 40632 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_ms = 1000 # Evict every 1s +eviction_interval_ms = 250 # 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" \ No newline at end of file +file_metrics_directory = "./logs" diff --git a/oxcache/src/cli.rs b/oxcache/src/cli.rs index 6879d98..6cf9f7f 100644 --- a/oxcache/src/cli.rs +++ b/oxcache/src/cli.rs @@ -47,9 +47,6 @@ pub struct CliArgs { #[arg(long)] pub low_water_evict: Option, - #[arg(long)] - pub high_water_clean: Option, - #[arg(long)] pub low_water_clean: Option, @@ -104,7 +101,6 @@ pub struct ParsedEvictionConfig { pub eviction_policy: Option, pub high_water_evict: Option, pub low_water_evict: Option, - pub high_water_clean: Option, pub low_water_clean: Option, pub eviction_interval_ms: Option, } @@ -216,10 +212,6 @@ pub fn load_config(cli: &CliArgs) -> Result Result high_water_clean { - return Err("low_water_clean must be less than high_water_clean".into()); - } - - if high_water_clean > low_water_evict { - return Err("high_water_clean must be less than or equal to low_water_evict".into()); + if low_water_clean >= (low_water_evict - high_water_evict) { + return Err("low_water_clean must be less than (low_water_evict - high_water_evict)".into()); } } @@ -315,7 +299,6 @@ pub fn load_config(cli: &CliArgs) -> Result std::io::Result; /// This is expected to remove elements from the cache as well - fn evict(self: Arc, locations: EvictTarget, cache: Arc, writer_pool: Arc) -> io::Result<()>; + fn evict(self: Arc, cache: Arc, writer_pool: Arc, eviction_policy: Arc>) -> io::Result<()>; fn read(&self, location: ChunkLocation) -> std::io::Result; @@ -473,11 +473,16 @@ impl Device for Zoned { Ok(Bytes::from_owner(buffer)) } - fn evict(self: Arc, locations: EvictTarget, cache: Arc, writer_pool: Arc) -> io::Result<()> { + fn evict(self: Arc, cache: Arc, writer_pool: Arc, eviction_policy: Arc>) -> io::Result<()> { let usage = self.get_use_percentage(); METRICS.update_metric_gauge("usage_percentage", usage as f64); - match locations { + let targets = { + let mut policy = eviction_policy.lock().unwrap(); + policy.get_evict_targets() + }; + + match targets { EvictTarget::Chunk(chunk_locations, clean_locations) => { if chunk_locations.is_empty() { tracing::debug!("[evict:Chunk] No chunks evicted"); @@ -854,10 +859,16 @@ impl Device for BlockInterface { Ok(Bytes::from(data)) } - fn evict(self: Arc, locations: EvictTarget, cache: Arc, _writer_pool: Arc) -> io::Result<()> { + fn evict(self: Arc, cache: Arc, _writer_pool: Arc, eviction_policy: Arc>) -> io::Result<()> { let usage = self.get_use_percentage(); METRICS.update_metric_gauge("usage_percentage", usage as f64); - match locations { + + let targets = { + let mut policy = eviction_policy.lock().unwrap(); + policy.get_evict_targets() + }; + + match targets { EvictTarget::Chunk(chunk_locations, _) => { if chunk_locations.is_empty() { diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index f9185e5..02f863f 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -33,18 +33,16 @@ impl EvictionPolicyWrapper { low_water: Zone, nr_zones: Zone, nr_chunks_per_zone: Chunk, - clean_high_water: Option, clean_low_water: Option, ) -> tokio::io::Result { match identifier.to_lowercase().as_str() { "chunk" => { - if clean_high_water.is_none() || clean_low_water.is_none() { + if clean_low_water.is_none() { return Err(std::io::Error::new(ErrorKind::InvalidInput, "Chunk eviction must have clean_high_water and clean_low_water")); } Ok(EvictionPolicyWrapper::Chunk(ChunkEvictionPolicy::new( high_water, low_water, - clean_high_water.unwrap(), clean_low_water.unwrap(), nr_zones, nr_chunks_per_zone, @@ -185,14 +183,13 @@ impl ChunkEvictionPolicy { pub fn new( high_water: Chunk, low_water: Chunk, - clean_high_water: Chunk, clean_low_water: Chunk, nr_zones: Zone, nr_chunks_per_zone: Chunk, ) -> Self { assert!( high_water > nr_chunks_per_zone, - "high_water={} must be larget than nr_chunks_per_zone={} to leave room for eviction", + "high_water={} must be larger than nr_chunks_per_zone={} to leave room for eviction", high_water, nr_chunks_per_zone ); Self { @@ -201,7 +198,7 @@ impl ChunkEvictionPolicy { nr_zones, nr_chunks_per_zone, lru: LruCache::unbounded(), - pq: ZonePriorityQueue::new(nr_zones, clean_high_water, clean_low_water) + pq: ZonePriorityQueue::new(nr_zones, clean_low_water) } } } @@ -317,14 +314,8 @@ impl Evictor { } }; - let mut policy = eviction_policy_clone.lock().unwrap(); - - let targets = policy.get_evict_targets(); - - drop(policy); - let device_clone = device_clone.clone(); - let result = match device_clone.evict(targets, cache_clone.clone(), writer_pool.clone()) { + let result = match device_clone.evict(cache_clone.clone(), writer_pool.clone(), eviction_policy_clone.clone()) { Err(e) => { Err(e.to_string()) } @@ -402,146 +393,148 @@ mod tests { } } - #[test] - fn test_chunk_update_ordering() { - let mut policy = ChunkEvictionPolicy::new(1, 3, 1, 0, 2, 2); - - // zone=[_,_,_,_], lru=() - let c = ChunkLocation::new(1, 0); - let mut order: VecDeque = VecDeque::new(); - order.push_front(c.clone()); - policy.write_update(c); - // zone=[_,_,(1,0),_], lru=((1,0)) - compare_order(&mut policy.lru, &order); - - let et = policy.get_evict_targets(); - let expect_none: VecDeque = VecDeque::new(); - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - let c = ChunkLocation::new(1, 1); - policy.write_update(c.clone()); - // zone=[_,_,(1,0),(1,1)], lru=((1,0),(1,1)) - order.push_front(c); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - let c = ChunkLocation::new(1, 0); - // Expect order to update - policy.read_update(c.clone()); - // zone=[_,_,(1,0),(1,1)], lru=((1,1),(1,0)) - let c = order.pop_back().unwrap(); - order.push_front(c); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - let c = ChunkLocation::new(0, 0); - policy.write_update(c.clone()); - // zone=[(0,0),_,(1,0),(1,1)], lru=((1,0),(1,1),(0,0)) - order.push_front(c); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - let order = order - .clone() - .into_iter() - .rev() - .collect::>(); - assert_eq!(order, et, "Expected = {:?}, but got {:?}", order, et); - } - - #[test] - fn test_promotional_update_ordering() { - let mut policy = PromotionalEvictionPolicy::new(1, 3, 4, 2); - - // zone=[_,_,_,_], lru=() - let mut order: VecDeque = VecDeque::new(); - policy.write_update(ChunkLocation::new(3, 0)); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - let expect_none: Vec = vec![]; - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - // zone=[_,_,_,_], lru=() - policy.write_update(ChunkLocation::new(3, 1)); - // zone=[_,_,_,3], lru=(3) - order.push_back(3); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - policy.write_update(ChunkLocation::new(1, 0)); - // There should be no change - // zone=[_,_,_,3], lru=(3) - compare_order(&mut policy.lru, &order); - - policy.write_update(ChunkLocation::new(1, 1)); - // zone=[_,1,_,3], lru=(3, 1) - order.push_front(1); - compare_order(&mut policy.lru, &order); - let et = policy.get_evict_targets(); - assert_eq!( - expect_none, et, - "Expected = {:?}, but got {:?}", - expect_none, et - ); - - policy.write_update(ChunkLocation::new(2, 0)); - policy.write_update(ChunkLocation::new(2, 1)); - order.push_front(2); - // zone=[_,1,2,3], lru=(3, 1, 2) - compare_order(&mut policy.lru, &order); - - // Should update in place, and adjust order - policy.read_update(ChunkLocation::new(3, 1)); - let c = order.pop_back().unwrap(); - order.push_front(c); - // zone=[_,1,2,3], lru=(1, 2, 3) - compare_order(&mut policy.lru, &order); - - let et = policy.get_evict_targets(); - let expect = VecDeque::from(vec![1, 2, 3]); - assert_eq!(expect, et, "Expected = {:?}, but got {:?}", expect, et); - - compare_order(&mut policy.lru, &VecDeque::from(vec![])); - } - - #[test] - fn check_chunk_priority_queue() { - // 8 zones, 1 chunks per zone. Should evict at 3 inserted - let mut policy = ChunkEvictionPolicy::new( - 2, 6, 4, 1, 4, 2); - - for z in 0..3 { - for i in 0..2 { - policy.write_update(ChunkLocation::new(z, i)); - } - } - - let got = policy.get_evict_targets().len(); - assert_eq!(5, got, "Expected 5, but got {}", got); - - let got = policy.get_clean_targets().len(); - assert_eq!(3, got, "Expected 3, but got {}", got); - } + // TODO: Fix params + + // #[test] + // fn test_chunk_update_ordering() { + // let mut policy = ChunkEvictionPolicy::new(3, 5, 0, 6, 2); + // + // // zone=[_,_,_,_], lru=() + // let c = ChunkLocation::new(1, 0); + // let mut order: VecDeque = VecDeque::new(); + // order.push_front(c.clone()); + // policy.write_update(c); + // // zone=[_,_,(1,0),_], lru=((1,0)) + // compare_order(&mut policy.lru, &order); + // + // let et = policy.get_evict_targets(); + // let expect_none: VecDeque = VecDeque::new(); + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // let c = ChunkLocation::new(1, 1); + // policy.write_update(c.clone()); + // // zone=[_,_,(1,0),(1,1)], lru=((1,0),(1,1)) + // order.push_front(c); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // let c = ChunkLocation::new(1, 0); + // // Expect order to update + // policy.read_update(c.clone()); + // // zone=[_,_,(1,0),(1,1)], lru=((1,1),(1,0)) + // let c = order.pop_back().unwrap(); + // order.push_front(c); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // let c = ChunkLocation::new(0, 0); + // policy.write_update(c.clone()); + // // zone=[(0,0),_,(1,0),(1,1)], lru=((1,0),(1,1),(0,0)) + // order.push_front(c); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // let order = order + // .clone() + // .into_iter() + // .rev() + // .collect::>(); + // assert_eq!(order, et, "Expected = {:?}, but got {:?}", order, et); + // } + // + // #[test] + // fn test_promotional_update_ordering() { + // let mut policy = PromotionalEvictionPolicy::new(1, 3, 4, 2); + // + // // zone=[_,_,_,_], lru=() + // let mut order: VecDeque = VecDeque::new(); + // policy.write_update(ChunkLocation::new(3, 0)); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // let expect_none: Vec = vec![]; + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // // zone=[_,_,_,_], lru=() + // policy.write_update(ChunkLocation::new(3, 1)); + // // zone=[_,_,_,3], lru=(3) + // order.push_back(3); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // policy.write_update(ChunkLocation::new(1, 0)); + // // There should be no change + // // zone=[_,_,_,3], lru=(3) + // compare_order(&mut policy.lru, &order); + // + // policy.write_update(ChunkLocation::new(1, 1)); + // // zone=[_,1,_,3], lru=(3, 1) + // order.push_front(1); + // compare_order(&mut policy.lru, &order); + // let et = policy.get_evict_targets(); + // assert_eq!( + // expect_none, et, + // "Expected = {:?}, but got {:?}", + // expect_none, et + // ); + // + // policy.write_update(ChunkLocation::new(2, 0)); + // policy.write_update(ChunkLocation::new(2, 1)); + // order.push_front(2); + // // zone=[_,1,2,3], lru=(3, 1, 2) + // compare_order(&mut policy.lru, &order); + // + // // Should update in place, and adjust order + // policy.read_update(ChunkLocation::new(3, 1)); + // let c = order.pop_back().unwrap(); + // order.push_front(c); + // // zone=[_,1,2,3], lru=(1, 2, 3) + // compare_order(&mut policy.lru, &order); + // + // let et = policy.get_evict_targets(); + // let expect = VecDeque::from(vec![1, 2, 3]); + // assert_eq!(expect, et, "Expected = {:?}, but got {:?}", expect, et); + // + // compare_order(&mut policy.lru, &VecDeque::from(vec![])); + // } + // + // #[test] + // fn check_chunk_priority_queue() { + // // 8 zones, 1 chunks per zone. Should evict at 3 inserted + // let mut policy = ChunkEvictionPolicy::new( + // 2, 6, 1, 4, 2); + // + // for z in 0..3 { + // for i in 0..2 { + // policy.write_update(ChunkLocation::new(z, i)); + // } + // } + // + // let got = policy.get_evict_targets().len(); + // assert_eq!(5, got, "Expected 5, but got {}", got); + // + // let got = policy.get_clean_targets().len(); + // assert_eq!(3, got, "Expected 3, but got {}", got); + // } } diff --git a/oxcache/src/server.rs b/oxcache/src/server.rs index 2145926..b1607a9 100644 --- a/oxcache/src/server.rs +++ b/oxcache/src/server.rs @@ -53,7 +53,6 @@ pub struct ServerEvictionConfig { pub eviction_type: String, pub high_water_evict: u64, pub low_water_evict: u64, - pub high_water_clean: Option, pub low_water_clean: Option, pub eviction_interval: u64, } @@ -163,7 +162,6 @@ impl Server { self.config.eviction.low_water_evict, self.device.get_num_zones(), self.device.get_chunks_per_zone(), - self.config.eviction.high_water_clean, self.config.eviction.low_water_clean, )?)); diff --git a/oxcache/src/zone_state/zone_list.rs b/oxcache/src/zone_state/zone_list.rs index e6913a2..5dd6a12 100644 --- a/oxcache/src/zone_state/zone_list.rs +++ b/oxcache/src/zone_state/zone_list.rs @@ -617,14 +617,14 @@ impl ZoneList { #[cfg(test)] mod zone_list_tests { - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use crate::{ cache::{bucket::ChunkLocation, Cache}, device::Device, eviction::EvictTarget, writerpool::WriterPool, zone_state::zone_list::ZoneObtainFailure::{EvictNow, Wait} }; use bytes::Bytes; use nvme::types::{Byte, LogicalBlock, NVMeConfig, Zone}; - + use crate::eviction::EvictionPolicyWrapper; use super::ZoneList; struct MockDevice {} @@ -645,7 +645,7 @@ mod zone_list_tests { } /// This is expected to remove elements from the cache as well - fn evict(self: Arc, _locations: EvictTarget, _cache: Arc, _writer_pool: Arc) -> std::io::Result<()> { + fn evict(self: Arc, _cache: Arc, _writer_pool: Arc, _eviction_policy: Arc>) -> std::io::Result<()> { Ok(()) } diff --git a/oxcache/src/zone_state/zone_priority_queue.rs b/oxcache/src/zone_state/zone_priority_queue.rs index d9609b9..5d235ec 100644 --- a/oxcache/src/zone_state/zone_priority_queue.rs +++ b/oxcache/src/zone_state/zone_priority_queue.rs @@ -8,14 +8,11 @@ type ZonePriority = Chunk; pub struct ZonePriorityQueue { invalid_count: ZonePriority, invalid_queue: PriorityQueue, // max-heap by priority - high_water_thresh: Chunk, // trigger cleaning when >= this low_water_thresh: Chunk, // clean down to < this } impl ZonePriorityQueue { - pub fn new(num_zones: ZoneIndex, high_water_thresh: Chunk, low_water_thresh: Chunk) -> Self { - assert!(high_water_thresh > low_water_thresh); - + pub fn new(num_zones: ZoneIndex, low_water_thresh: Chunk) -> Self { let mut invalid_queue = PriorityQueue::new(); for z in 0..num_zones { invalid_queue.push(z, 0); @@ -24,7 +21,6 @@ impl ZonePriorityQueue { Self { invalid_queue, invalid_count: 0, - high_water_thresh: high_water_thresh, low_water_thresh: low_water_thresh, } } @@ -43,9 +39,6 @@ impl ZonePriorityQueue { pub fn remove_if_thresh_met(&mut self) -> Vec { let mut zones = Vec::new(); tracing::trace!("[evict:Chunk] Cleaning zones, invalid={}", self.invalid_count); - if self.invalid_count < self.high_water_thresh { - return zones; - } while self.invalid_count >= self.low_water_thresh { zones.push(self.pop_reset()); } diff --git a/oxcache/tests/mock_device.rs b/oxcache/tests/mock_device.rs index 8e28827..d527582 100644 --- a/oxcache/tests/mock_device.rs +++ b/oxcache/tests/mock_device.rs @@ -7,6 +7,7 @@ use std::sync::Mutex; use oxcache::{ cache::bucket::ChunkLocation, device::Device, eviction::EvictorMessage, zone_state::zone_list::{self, ZoneList, ZoneObtainFailure} }; +use oxcache::eviction::EvictionPolicyWrapper; use oxcache::writerpool::WriterPool; struct Chunk { @@ -134,9 +135,9 @@ impl Device for MockZonedDevice { fn evict( self: Arc, - locations: oxcache::eviction::EvictTarget, cache: std::sync::Arc, - writer_pool: Arc + writer_pool: Arc, + eviction_policy: Arc> ) -> std::io::Result<()> { todo!() } From ab87d1cc3cce6b329b0207f1dc1a578e8097905c Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Tue, 23 Sep 2025 14:46:21 -0700 Subject: [PATCH 09/13] Fix evict tests --- oxcache/src/eviction.rs | 300 ++++++++++++++++++++-------------------- 1 file changed, 152 insertions(+), 148 deletions(-) diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 02f863f..88e64c5 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -229,13 +229,14 @@ impl EvictionPolicy for ChunkEvictionPolicy { let cap = lru_len - low_water_mark; let mut targets = Vec::with_capacity(cap as usize); - while self.lru.len() as Chunk >= low_water_mark { - let targ = self.lru.pop_lru().unwrap().0; - let target_zone = targ.zone; - targets.push(targ); + for _ in 0..cap { + if let Some((targ, _)) = self.lru.pop_lru() { + let target_zone = targ.zone; + targets.push(targ); - // Adjust pq - self.pq.modify_priority(target_zone, 1); + // Adjust pq + self.pq.modify_priority(target_zone, 1); + } } targets @@ -315,6 +316,7 @@ impl Evictor { }; let device_clone = device_clone.clone(); + let eviction_start = std::time::Instant::now(); let result = match device_clone.evict(cache_clone.clone(), writer_pool.clone(), eviction_policy_clone.clone()) { Err(e) => { Err(e.to_string()) @@ -323,6 +325,8 @@ impl Evictor { Ok(()) }, }; + let eviction_duration = eviction_start.elapsed(); + tracing::info!("[Eviction] Total eviction took {:?}", eviction_duration); if let Some(sender) = sender { tracing::debug!("Sending eviction response to sender: {:?}", result); @@ -395,146 +399,146 @@ mod tests { // TODO: Fix params - // #[test] - // fn test_chunk_update_ordering() { - // let mut policy = ChunkEvictionPolicy::new(3, 5, 0, 6, 2); - // - // // zone=[_,_,_,_], lru=() - // let c = ChunkLocation::new(1, 0); - // let mut order: VecDeque = VecDeque::new(); - // order.push_front(c.clone()); - // policy.write_update(c); - // // zone=[_,_,(1,0),_], lru=((1,0)) - // compare_order(&mut policy.lru, &order); - // - // let et = policy.get_evict_targets(); - // let expect_none: VecDeque = VecDeque::new(); - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // let c = ChunkLocation::new(1, 1); - // policy.write_update(c.clone()); - // // zone=[_,_,(1,0),(1,1)], lru=((1,0),(1,1)) - // order.push_front(c); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // let c = ChunkLocation::new(1, 0); - // // Expect order to update - // policy.read_update(c.clone()); - // // zone=[_,_,(1,0),(1,1)], lru=((1,1),(1,0)) - // let c = order.pop_back().unwrap(); - // order.push_front(c); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // let c = ChunkLocation::new(0, 0); - // policy.write_update(c.clone()); - // // zone=[(0,0),_,(1,0),(1,1)], lru=((1,0),(1,1),(0,0)) - // order.push_front(c); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // let order = order - // .clone() - // .into_iter() - // .rev() - // .collect::>(); - // assert_eq!(order, et, "Expected = {:?}, but got {:?}", order, et); - // } - // - // #[test] - // fn test_promotional_update_ordering() { - // let mut policy = PromotionalEvictionPolicy::new(1, 3, 4, 2); - // - // // zone=[_,_,_,_], lru=() - // let mut order: VecDeque = VecDeque::new(); - // policy.write_update(ChunkLocation::new(3, 0)); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // let expect_none: Vec = vec![]; - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // // zone=[_,_,_,_], lru=() - // policy.write_update(ChunkLocation::new(3, 1)); - // // zone=[_,_,_,3], lru=(3) - // order.push_back(3); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // policy.write_update(ChunkLocation::new(1, 0)); - // // There should be no change - // // zone=[_,_,_,3], lru=(3) - // compare_order(&mut policy.lru, &order); - // - // policy.write_update(ChunkLocation::new(1, 1)); - // // zone=[_,1,_,3], lru=(3, 1) - // order.push_front(1); - // compare_order(&mut policy.lru, &order); - // let et = policy.get_evict_targets(); - // assert_eq!( - // expect_none, et, - // "Expected = {:?}, but got {:?}", - // expect_none, et - // ); - // - // policy.write_update(ChunkLocation::new(2, 0)); - // policy.write_update(ChunkLocation::new(2, 1)); - // order.push_front(2); - // // zone=[_,1,2,3], lru=(3, 1, 2) - // compare_order(&mut policy.lru, &order); - // - // // Should update in place, and adjust order - // policy.read_update(ChunkLocation::new(3, 1)); - // let c = order.pop_back().unwrap(); - // order.push_front(c); - // // zone=[_,1,2,3], lru=(1, 2, 3) - // compare_order(&mut policy.lru, &order); - // - // let et = policy.get_evict_targets(); - // let expect = VecDeque::from(vec![1, 2, 3]); - // assert_eq!(expect, et, "Expected = {:?}, but got {:?}", expect, et); - // - // compare_order(&mut policy.lru, &VecDeque::from(vec![])); - // } - // - // #[test] - // fn check_chunk_priority_queue() { - // // 8 zones, 1 chunks per zone. Should evict at 3 inserted - // let mut policy = ChunkEvictionPolicy::new( - // 2, 6, 1, 4, 2); - // - // for z in 0..3 { - // for i in 0..2 { - // policy.write_update(ChunkLocation::new(z, i)); - // } - // } - // - // let got = policy.get_evict_targets().len(); - // assert_eq!(5, got, "Expected 5, but got {}", got); - // - // let got = policy.get_clean_targets().len(); - // assert_eq!(3, got, "Expected 3, but got {}", got); - // } + #[test] + fn test_chunk_update_ordering() { + let mut policy = ChunkEvictionPolicy::new(9, 12, 0, 6, 2); + + // zone=[_,_,_,_], lru=() + let c = ChunkLocation::new(1, 0); + let mut order: VecDeque = VecDeque::new(); + order.push_front(c.clone()); + policy.write_update(c); + // zone=[_,_,(1,0),_], lru=((1,0)) + compare_order(&mut policy.lru, &order); + + let et = policy.get_evict_targets(); + let expect_none: VecDeque = VecDeque::new(); + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + let c = ChunkLocation::new(1, 1); + policy.write_update(c.clone()); + // zone=[_,_,(1,0),(1,1)], lru=((1,0),(1,1)) + order.push_front(c); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + let c = ChunkLocation::new(1, 0); + // Expect order to update + policy.read_update(c.clone()); + // zone=[_,_,(1,0),(1,1)], lru=((1,1),(1,0)) + let c = order.pop_back().unwrap(); + order.push_front(c); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + let c = ChunkLocation::new(0, 0); + policy.write_update(c.clone()); + // zone=[(0,0),_,(1,0),(1,1)], lru=((1,0),(1,1),(0,0)) + order.push_front(c); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + let order = order + .clone() + .into_iter() + .rev() + .collect::>(); + assert_eq!(order, et, "Expected = {:?}, but got {:?}", order, et); + } + + #[test] + fn test_promotional_update_ordering() { + let mut policy = PromotionalEvictionPolicy::new(1, 3, 4, 2); + + // zone=[_,_,_,_], lru=() + let mut order: VecDeque = VecDeque::new(); + policy.write_update(ChunkLocation::new(3, 0)); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + let expect_none: Vec = vec![]; + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + // zone=[_,_,_,_], lru=() + policy.write_update(ChunkLocation::new(3, 1)); + // zone=[_,_,_,3], lru=(3) + order.push_back(3); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + policy.write_update(ChunkLocation::new(1, 0)); + // There should be no change + // zone=[_,_,_,3], lru=(3) + compare_order(&mut policy.lru, &order); + + policy.write_update(ChunkLocation::new(1, 1)); + // zone=[_,1,_,3], lru=(3, 1) + order.push_front(1); + compare_order(&mut policy.lru, &order); + let et = policy.get_evict_targets(); + assert_eq!( + expect_none, et, + "Expected = {:?}, but got {:?}", + expect_none, et + ); + + policy.write_update(ChunkLocation::new(2, 0)); + policy.write_update(ChunkLocation::new(2, 1)); + order.push_front(2); + // zone=[_,1,2,3], lru=(3, 1, 2) + compare_order(&mut policy.lru, &order); + + // Should update in place, and adjust order + policy.read_update(ChunkLocation::new(3, 1)); + let c = order.pop_back().unwrap(); + order.push_front(c); + // zone=[_,1,2,3], lru=(1, 2, 3) + compare_order(&mut policy.lru, &order); + + let et = policy.get_evict_targets(); + let expect = VecDeque::from(vec![1, 2, 3]); + assert_eq!(expect, et, "Expected = {:?}, but got {:?}", expect, et); + + compare_order(&mut policy.lru, &VecDeque::from(vec![])); + } + + #[test] + fn check_chunk_priority_queue() { + // 4 zones, 2 chunks per zone. Should evict at 3 inserted + let mut policy = ChunkEvictionPolicy::new( + 3, 6, 1, 4, 2); + + for z in 0..3 { + for i in 0..2 { + policy.write_update(ChunkLocation::new(z, i)); + } + } + + let got = policy.get_evict_targets().len(); + assert_eq!(4, got, "Expected 4, but got {}", got); + + let got = policy.get_clean_targets().len(); + assert_eq!(2, got, "Expected 2, but got {}", got); + } } From 31343a7af9bd309051875e0747f791dac55409b9 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Tue, 23 Sep 2025 16:54:50 -0700 Subject: [PATCH 10/13] Optimize LRU by using retain and lru-mem vs full rebuild --- oxcache/Cargo.toml | 2 +- oxcache/src/eviction.rs | 176 +++++++++++++++++++++++++++++++++++----- 2 files changed, 155 insertions(+), 23 deletions(-) diff --git a/oxcache/Cargo.toml b/oxcache/Cargo.toml index 5a5e0df..fc6534a 100644 --- a/oxcache/Cargo.toml +++ b/oxcache/Cargo.toml @@ -22,7 +22,7 @@ async-trait = "0.1.88" rand = "0.9.1" rand_pcg = "0.9.0" flume = "0.11.1" -lru = "0.16.0" +lru-mem = "0.3.0" ndarray = "0.16.1" uuid = { version = "1.17.0", features = ["v4"] } byteorder = "1.5.0" diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 88e64c5..7245379 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -4,7 +4,7 @@ use crate::device::Device; use crate::writerpool::WriterPool; use crate::zone_state::zone_priority_queue::{ZoneIndex, ZonePriorityQueue}; use flume::{Receiver, Sender}; -use lru::LruCache; +use lru_mem::{LruCache, MemSize}; use nvme::types::{Chunk, Zone}; use std::sync::{ Arc, Mutex, @@ -14,6 +14,8 @@ use std::thread::{self, JoinHandle}; use std::time::Duration; use crate::zone_state::zone_priority_queue; +// Note: Unit type () should already implement MemSize via blanket implementations + #[derive(Debug)] pub enum EvictionPolicyWrapper { Promotional(PromotionalEvictionPolicy), @@ -112,7 +114,7 @@ impl PromotionalEvictionPolicy { nr_zones: Zone, nr_chunks_per_zone: Chunk, ) -> Self { - let lru = LruCache::unbounded(); + let lru = LruCache::new(usize::MAX); // Effectively unbounded Self { high_water, low_water, @@ -134,7 +136,7 @@ impl EvictionPolicy for PromotionalEvictionPolicy { // We only want to put it in the LRU once the zone is full if chunk.index == self.nr_chunks_per_zone - 1 { - self.lru.put(chunk.zone, ()); + self.lru.insert(chunk.zone, ()).ok(); } } @@ -143,7 +145,7 @@ impl EvictionPolicy for PromotionalEvictionPolicy { // If it has filled before we want to update every time "promoting" it // Following this, only zones that have filled prior are updated if self.lru.contains(&chunk.zone) { - self.lru.put(chunk.zone, ()); + self.lru.insert(chunk.zone, ()).ok(); } } @@ -160,7 +162,7 @@ impl EvictionPolicy for PromotionalEvictionPolicy { let mut targets = Vec::with_capacity(cap as usize); while self.lru.len() as Zone >= low_water_mark { - targets.push(self.lru.pop_lru().unwrap().0) + targets.push(self.lru.remove_lru().unwrap().0) } targets @@ -197,7 +199,7 @@ impl ChunkEvictionPolicy { low_water, nr_zones, nr_chunks_per_zone, - lru: LruCache::unbounded(), + lru: LruCache::new(usize::MAX), // Effectively unbounded pq: ZonePriorityQueue::new(nr_zones, clean_low_water) } } @@ -207,12 +209,12 @@ impl EvictionPolicy for ChunkEvictionPolicy { type Target = Vec; type CleanTarget = Vec; fn write_update(&mut self, chunk: ChunkLocation) { - self.lru.put(chunk, ()); + self.lru.insert(chunk, ()).ok(); } fn read_update(&mut self, chunk: ChunkLocation) { if self.lru.contains(&chunk) { - self.lru.put(chunk, ()); + self.lru.insert(chunk, ()).ok(); } } @@ -230,7 +232,7 @@ impl EvictionPolicy for ChunkEvictionPolicy { let mut targets = Vec::with_capacity(cap as usize); for _ in 0..cap { - if let Some((targ, _)) = self.lru.pop_lru() { + if let Some((targ, _)) = self.lru.remove_lru() { let target_zone = targ.zone; targets.push(targ); @@ -252,20 +254,12 @@ impl EvictionPolicy for ChunkEvictionPolicy { let zones_to_clean: std::collections::HashSet = clean_targets.iter().copied().collect(); - let mut items_to_reinsert = Vec::new(); - - // Keep those not in cleaned zones - while let Some((chunk_loc, _)) = self.lru.pop_lru() { - if !zones_to_clean.contains(&chunk_loc.zone) { - items_to_reinsert.push(chunk_loc); - } - } - // Re-insert in reverse order to maintain LRU ordering - // (most recently used items go back in last) - for chunk_loc in items_to_reinsert.into_iter().rev() { - self.lru.put(chunk_loc, ()); - } + // Efficient selective removal - O(k) where k = items removed + // instead of O(n) where n = total LRU size + self.lru.retain(|chunk_loc, _| { + !zones_to_clean.contains(&chunk_loc.zone) + }); clean_targets } @@ -541,4 +535,142 @@ mod tests { let got = policy.get_clean_targets().len(); assert_eq!(2, got, "Expected 2, but got {}", got); } + + #[test] + fn performance_test_large_lru_get_clean_targets() { + // Performance test with ~15.6M chunks in LRU + // 904 zones, 17232 chunks per zone = 15,581,728 total chunks + let nr_zones = 904; + let nr_chunks_per_zone = 17232; + let total_chunks = nr_zones * nr_chunks_per_zone; + + // Set clean_low_water to trigger cleaning when zones have 1+ evicted chunks + let clean_low_water = 1; + + // High/low water marks - trigger eviction when LRU approaches capacity + let high_water = total_chunks - (total_chunks / 20); // 95% capacity + let low_water = total_chunks - (total_chunks / 10); // 90% capacity + + let mut policy = ChunkEvictionPolicy::new( + high_water, low_water, clean_low_water, nr_zones, nr_chunks_per_zone); + + println!("Filling LRU with {} chunks across {} zones...", total_chunks, nr_zones); + let start_fill = std::time::Instant::now(); + + // Fill the LRU with chunks from all zones + for zone in 0..nr_zones { + for chunk_idx in 0..nr_chunks_per_zone { + policy.write_update(ChunkLocation::new(zone, chunk_idx)); + } + } + + let fill_duration = start_fill.elapsed(); + println!("LRU fill took: {:?}", fill_duration); + println!("LRU size: {}", policy.lru.len()); + + // Trigger some evictions to populate the priority queue + // This will evict 500 chunks and mark zones for potential cleaning + println!("Triggering evictions to populate priority queue..."); + let evict_start = std::time::Instant::now(); + let evicted = policy.get_evict_targets(); + let evict_duration = evict_start.elapsed(); + println!("Evicted {} chunks in {:?}", evicted.len(), evict_duration); + + // Now test get_clean_targets performance with different scenarios + + // Scenario 1: Small cleanup (few zones) + println!("\n=== Scenario 1: Small cleanup ==="); + let start_small = std::time::Instant::now(); + let clean_targets_small = policy.get_clean_targets(); + let small_duration = start_small.elapsed(); + println!("Small cleanup: {} zones cleaned in {:?}", + clean_targets_small.len(), small_duration); + println!("LRU size after small cleanup: {}", policy.lru.len()); + + // Refill LRU with new chunks to simulate continued cache activity + println!("Refilling LRU with new chunks for scenario 2..."); + let refill_start = std::time::Instant::now(); + let chunks_to_add = (total_chunks as usize) / 3; // Add 33% more chunks + for i in 0..chunks_to_add { + let zone = i % (nr_zones as usize); + let chunk_idx = (i / (nr_zones as usize)) % (nr_chunks_per_zone as usize); + // Use high zone/chunk indices to avoid conflicts with existing chunks + policy.write_update(ChunkLocation::new( + (zone + nr_zones as usize) as u64, + chunk_idx as u64 + )); + } + let refill_duration = refill_start.elapsed(); + println!("Refilled LRU with {} chunks in {:?}", chunks_to_add, refill_duration); + println!("LRU size after refill: {}", policy.lru.len()); + + // Trigger evictions to populate priority queue for scenario 2 + let evict2_start = std::time::Instant::now(); + let evicted2 = policy.get_evict_targets(); + let evict2_duration = evict2_start.elapsed(); + println!("Second eviction: {} chunks in {:?}", evicted2.len(), evict2_duration); + + // Scenario 2: Medium cleanup + println!("\n=== Scenario 2: Medium cleanup ==="); + let start_medium = std::time::Instant::now(); + let clean_targets_medium = policy.get_clean_targets(); + let medium_duration = start_medium.elapsed(); + println!("Medium cleanup: {} zones cleaned in {:?}", + clean_targets_medium.len(), medium_duration); + println!("LRU size after medium cleanup: {}", policy.lru.len()); + + // Refill LRU again for scenario 3 + println!("Refilling LRU with new chunks for scenario 3..."); + let refill2_start = std::time::Instant::now(); + let chunks_to_add2 = (total_chunks as usize) / 2; // Add 50% more chunks + for i in 0..chunks_to_add2 { + let zone = i % (nr_zones as usize); + let chunk_idx = (i / (nr_zones as usize)) % (nr_chunks_per_zone as usize); + // Use even higher indices to avoid conflicts + policy.write_update(ChunkLocation::new( + (zone + 2 * nr_zones as usize) as u64, + chunk_idx as u64 + )); + } + let refill2_duration = refill2_start.elapsed(); + println!("Refilled LRU with {} chunks in {:?}", chunks_to_add2, refill2_duration); + println!("LRU size after second refill: {}", policy.lru.len()); + + // Trigger evictions for scenario 3 + let evict3_start = std::time::Instant::now(); + let evicted3 = policy.get_evict_targets(); + let evict3_duration = evict3_start.elapsed(); + println!("Third eviction: {} chunks in {:?}", evicted3.len(), evict3_duration); + + // Scenario 3: Large cleanup + println!("\n=== Scenario 3: Large cleanup ==="); + let start_large = std::time::Instant::now(); + let clean_targets_large = policy.get_clean_targets(); + let large_duration = start_large.elapsed(); + println!("Large cleanup: {} zones cleaned in {:?}", + clean_targets_large.len(), large_duration); + println!("LRU size after large cleanup: {}", policy.lru.len()); + + // Performance analysis + println!("\n=== Performance Analysis ==="); + println!("Initial LRU size: {}", total_chunks); + println!("Small cleanup time: {:?} ({} zones)", small_duration, clean_targets_small.len()); + println!("Medium cleanup time: {:?} ({} zones)", medium_duration, clean_targets_medium.len()); + println!("Large cleanup time: {:?} ({} zones)", large_duration, clean_targets_large.len()); + + // Calculate time per LRU item processed + if policy.lru.len() > 0 { + let time_per_item_ns = large_duration.as_nanos() as f64 / total_chunks as f64; + println!("Approximate time per LRU item processed: {:.2} ns", time_per_item_ns); + } + + // Verify correctness - LRU should still function properly + assert!(policy.lru.len() <= total_chunks as usize); + + // Test that we can still perform normal operations + policy.write_update(ChunkLocation::new(999, 14)); + policy.read_update(ChunkLocation::new(0, 0)); + + println!("Test completed successfully!"); + } } From e224b53aa2cd90351d5176fa0cc928ac64bcab21 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Tue, 23 Sep 2025 17:11:43 -0700 Subject: [PATCH 11/13] Batch pq updates --- Cargo.lock | 16 ++++++++++------ oxcache/src/cache/bucket.rs | 10 ++++++++++ oxcache/src/eviction.rs | 19 +++++++++++++------ 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 470422b..9649a68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -269,7 +269,7 @@ dependencies = [ "http 0.2.12", "http 1.3.1", "http-body 0.4.6", - "lru 0.12.5", + "lru", "percent-encoding", "regex-lite", "sha2", @@ -1542,6 +1542,10 @@ name = "hashbrown" version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", + "allocator-api2", +] [[package]] name = "hashbrown" @@ -2092,12 +2096,12 @@ dependencies = [ ] [[package]] -name = "lru" -version = "0.16.0" +name = "lru-mem" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86ea4e65087ff52f3862caff188d489f1fab49a0cb09e01b2e3f1a617b10aaed" +checksum = "cf5c8c26d903a41c80d4cc171940a57a4d1bc51139ebd6aad87e2f9ae3774780" dependencies = [ - "hashbrown 0.15.3", + "hashbrown 0.14.5", ] [[package]] @@ -2378,7 +2382,7 @@ dependencies = [ "futures", "libc", "libnvme-sys", - "lru 0.16.0", + "lru-mem", "metrics", "metrics-exporter-prometheus", "ndarray", diff --git a/oxcache/src/cache/bucket.rs b/oxcache/src/cache/bucket.rs index eed47e2..385dcf8 100644 --- a/oxcache/src/cache/bucket.rs +++ b/oxcache/src/cache/bucket.rs @@ -2,6 +2,7 @@ use crate::request::GetRequest; use nvme::types::{Byte, Zone}; use std::sync::{Arc, atomic::{AtomicUsize, Ordering}}; use tokio::sync::Notify; +use lru_mem::HeapSize; #[derive(Debug, Clone, Eq, PartialEq, Hash)] pub struct Chunk { @@ -26,6 +27,15 @@ impl ChunkLocation { } } +impl HeapSize for ChunkLocation { + fn heap_size(&self) -> usize { + // ChunkLocation contains no heap-allocated data + 0 + } +} + +// ValueSize is automatically implemented via blanket implementation + /// A ChunkLocation with pin counting for coordinating with eviction #[derive(Debug)] pub struct PinnedChunkLocation { diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 7245379..5a34b97 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -14,8 +14,6 @@ use std::thread::{self, JoinHandle}; use std::time::Duration; use crate::zone_state::zone_priority_queue; -// Note: Unit type () should already implement MemSize via blanket implementations - #[derive(Debug)] pub enum EvictionPolicyWrapper { Promotional(PromotionalEvictionPolicy), @@ -231,16 +229,24 @@ impl EvictionPolicy for ChunkEvictionPolicy { let cap = lru_len - low_water_mark; let mut targets = Vec::with_capacity(cap as usize); + let mut zone_counts = std::collections::HashMap::new(); + + // Collect evicted items and count by zone (batch the counting) for _ in 0..cap { if let Some((targ, _)) = self.lru.remove_lru() { let target_zone = targ.zone; targets.push(targ); - // Adjust pq - self.pq.modify_priority(target_zone, 1); + // Batch count instead of individual priority queue updates + *zone_counts.entry(target_zone).or_insert(0) += 1; } } + // Batch update priority queue (far fewer operations) + for (zone, count) in zone_counts { + self.pq.modify_priority(zone, count); + } + targets } @@ -548,8 +554,9 @@ mod tests { let clean_low_water = 1; // High/low water marks - trigger eviction when LRU approaches capacity - let high_water = total_chunks - (total_chunks / 20); // 95% capacity - let low_water = total_chunks - (total_chunks / 10); // 90% capacity + // Use more reasonable eviction ratios to avoid massive bulk evictions + let high_water = total_chunks - (total_chunks / 100); // 99% capacity + let low_water = total_chunks - (total_chunks / 50); // 98% capacity let mut policy = ChunkEvictionPolicy::new( high_water, low_water, clean_low_water, nr_zones, nr_chunks_per_zone); From a10cc6121a769174ce9dce6c35d9690b1819a80b Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Thu, 25 Sep 2025 21:50:06 -0700 Subject: [PATCH 12/13] Fix chunk evict on block Clean shouldn't run on block --- oxcache/src/device.rs | 5 +++-- oxcache/src/eviction.rs | 8 +++++--- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/oxcache/src/device.rs b/oxcache/src/device.rs index b9f933b..c5356c3 100644 --- a/oxcache/src/device.rs +++ b/oxcache/src/device.rs @@ -479,11 +479,12 @@ impl Device for Zoned { let targets = { let mut policy = eviction_policy.lock().unwrap(); - policy.get_evict_targets() + policy.get_evict_targets(true) }; match targets { EvictTarget::Chunk(chunk_locations, clean_locations) => { + let clean_locations = clean_locations.unwrap(); if chunk_locations.is_empty() { tracing::debug!("[evict:Chunk] No chunks evicted"); return Ok(()); @@ -865,7 +866,7 @@ impl Device for BlockInterface { let targets = { let mut policy = eviction_policy.lock().unwrap(); - policy.get_evict_targets() + policy.get_evict_targets(false) }; match targets { diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 5a34b97..55c7691 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -22,7 +22,7 @@ pub enum EvictionPolicyWrapper { #[derive(Debug)] pub enum EvictTarget { - Chunk(Vec, Vec), + Chunk(Vec, Option>), Zone(Vec), } @@ -71,14 +71,16 @@ impl EvictionPolicyWrapper { } } - pub fn get_evict_targets(&mut self) -> EvictTarget { + pub fn get_evict_targets(&mut self, get_clean_targets: bool) -> EvictTarget { match self { EvictionPolicyWrapper::Promotional(promotional) => { EvictTarget::Zone(promotional.get_evict_targets()) } EvictionPolicyWrapper::Chunk(c) => { let et = c.get_evict_targets(); - let ct = c.get_clean_targets(); + let ct = if get_clean_targets { + Some(c.get_clean_targets()) + } else { None }; EvictTarget::Chunk(et, ct) }, } From a5b4f6a16b2bd89a407f47dd1f2667a164fcb7c2 Mon Sep 17 00:00:00 2001 From: John Ramsden Date: Thu, 25 Sep 2025 22:34:33 -0700 Subject: [PATCH 13/13] Fix lru iter ordering --- oxcache/src/eviction.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 55c7691..15022ec 100644 --- a/oxcache/src/eviction.rs +++ b/oxcache/src/eviction.rs @@ -390,7 +390,10 @@ mod tests { order.len(), lru.len() ); - for (_index, ((lru_key, _), order_item)) in lru.iter().zip(order.iter()).enumerate() { + // The lru_mem crate iterates from most recently used to least recently used + // but our order VecDeque is constructed with push_front for most recent + // So we need to reverse the iteration order to match + for (_index, ((lru_key, _), order_item)) in lru.iter().zip(order.iter().rev()).enumerate() { assert_eq!( order_item, lru_key, "Expected {:?}, but got {:?}",