diff --git a/Cargo.lock b/Cargo.lock index 2476ced..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" @@ -2032,9 +2036,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" @@ -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]] @@ -2376,8 +2380,9 @@ dependencies = [ "dashmap", "flume", "futures", + "libc", "libnvme-sys", - "lru 0.16.0", + "lru-mem", "metrics", "metrics-exporter-prometheus", "ndarray", diff --git a/Cargo.toml b/Cargo.toml index ccfb831..3a9eaf7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,5 +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.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..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 = 1 # 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/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/Cargo.toml b/oxcache/Cargo.toml index 6df91d1..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" @@ -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/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/cache/mod.rs b/oxcache/src/cache/mod.rs index 8356443..f5ed74e 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,42 @@ 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 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 +304,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 @@ -316,6 +318,7 @@ impl Cache { // Clear old reverse slots 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; } } @@ -328,8 +331,11 @@ impl Cache { .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 +350,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 +399,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 +437,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/cli.rs b/oxcache/src/cli.rs index f32638c..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, @@ -57,7 +54,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, @@ -104,9 +101,8 @@ 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: Option, + pub eviction_interval_ms: Option, } #[derive(Debug, Deserialize)] @@ -216,19 +212,15 @@ 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; + 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<()>; + fn evict(self: Arc, cache: Arc, writer_pool: Arc, eviction_policy: Arc>) -> io::Result<()>; fn read(&self, location: ChunkLocation) -> std::io::Result; @@ -109,15 +115,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 +162,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); @@ -170,47 +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 { + 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 => { @@ -218,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 => { @@ -246,6 +237,7 @@ impl Zoned { Ok(()) } + } impl Zoned { @@ -271,7 +263,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 +276,7 @@ impl Zoned { } else { num_zones }; - + let zone_list = ZoneList::new( restricted_num_zones, config.chunks_per_zone, @@ -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; @@ -448,19 +421,27 @@ 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 { - match self.get_free_zone() { - Ok(res) => break res, + match self.get_free_zone(is_eviction) { + Ok(res) => { + break res; + }, Err(err) => { - tracing::trace!("[append] Failed to get free zone: {}", 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())?; }; @@ -492,31 +473,35 @@ 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(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(()); } - 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)] @@ -525,14 +510,42 @@ 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() + + 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 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?); + } + + Ok(all_results) } } }, @@ -545,51 +558,55 @@ 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 _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 { + // Used to verify no RACE, TODO: Remove! + // tokio::time::sleep(Duration::from_secs(5)).await; - writer_pool.send(WriteRequest{ - data: data.clone(), - responder: tx, - }).await?; + let (batch_tx, batch_rx) = flume::bounded(1); - 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)?; + let batch_request = BatchWriteRequest { + data: data_vec, + responder: batch_tx, + }; - Ok::<(CacheKey, ChunkLocation, bytes::Bytes), io::Error>((key.clone(), location, data.clone())) - }); - } + writer_pool.send_priority_batch(batch_request).await?; + + 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) => { @@ -599,8 +616,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(()) } } @@ -711,7 +726,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 { @@ -724,7 +739,7 @@ impl BlockInterface { } else { num_zones }; - + // Chunks per zone let chunks_per_zone = block_zone_capacity / chunk_size; @@ -797,7 +812,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(); @@ -844,10 +860,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(false) + }; + + match targets { EvictTarget::Chunk(chunk_locations, _) => { if chunk_locations.is_empty() { diff --git a/oxcache/src/eviction.rs b/oxcache/src/eviction.rs index 0f40fbe..15022ec 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, @@ -22,7 +22,7 @@ pub enum EvictionPolicyWrapper { #[derive(Debug)] pub enum EvictTarget { - Chunk(Vec, Vec), + Chunk(Vec, Option>), Zone(Vec), } @@ -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, @@ -73,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) }, } @@ -114,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, @@ -136,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(); } } @@ -145,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(); } } @@ -162,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 @@ -185,18 +185,22 @@ 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 larger than nr_chunks_per_zone={} to leave room for eviction", + high_water, nr_chunks_per_zone + ); Self { high_water, low_water, nr_zones, nr_chunks_per_zone, - lru: LruCache::unbounded(), - pq: ZonePriorityQueue::new(nr_zones, clean_high_water, clean_low_water) + lru: LruCache::new(usize::MAX), // Effectively unbounded + pq: ZonePriorityQueue::new(nr_zones, clean_low_water) } } } @@ -205,26 +209,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, ()); + self.lru.insert(chunk, ()).ok(); } 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, ()); + self.lru.insert(chunk, ()).ok(); } } 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 +231,22 @@ 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); + 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); + + // 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 @@ -249,21 +254,20 @@ impl EvictionPolicy for ChunkEvictionPolicy { fn get_clean_targets(&mut self) -> Self::CleanTarget { let mut clean_targets = self.pq.remove_if_thresh_met(); - clean_targets.sort_unstable(); + if clean_targets.is_empty() { + return clean_targets; + } - // 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::>(); + clean_targets.sort_unstable(); - self.lru.clear(); + let zones_to_clean: std::collections::HashSet = + clean_targets.iter().copied().collect(); - for k in new_lru_list { - self.lru.put(k, ()); - } + // 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 } @@ -313,19 +317,18 @@ 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 eviction_start = std::time::Instant::now(); + let result = match device_clone.evict(cache_clone.clone(), writer_pool.clone(), eviction_policy_clone.clone()) { Err(e) => { - tracing::error!("Error evicting: {}", e); Err(e.to_string()) } - Ok(_) => Ok(()), + Ok(_) => { + 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); @@ -387,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 {:?}", @@ -396,9 +402,11 @@ mod tests { } } + // TODO: Fix params + #[test] fn test_chunk_update_ordering() { - let mut policy = ChunkEvictionPolicy::new(1, 3, 1, 0, 2, 2); + let mut policy = ChunkEvictionPolicy::new(9, 12, 0, 6, 2); // zone=[_,_,_,_], lru=() let c = ChunkLocation::new(1, 0); @@ -522,9 +530,9 @@ mod tests { #[test] fn check_chunk_priority_queue() { - // 8 zones, 1 chunks per zone. Should evict at 3 inserted + // 4 zones, 2 chunks per zone. Should evict at 3 inserted let mut policy = ChunkEvictionPolicy::new( - 2, 6, 4, 1, 4, 2); + 3, 6, 1, 4, 2); for z in 0..3 { for i in 0..2 { @@ -533,9 +541,148 @@ mod tests { } let got = policy.get_evict_targets().len(); - assert_eq!(5, got, "Expected 5, but got {}", got); + assert_eq!(4, got, "Expected 4, but got {}", got); let got = policy.get_clean_targets().len(); - assert_eq!(3, got, "Expected 3, but got {}", got); + 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 + // 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); + + 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!"); } } diff --git a/oxcache/src/server.rs b/oxcache/src/server.rs index 4e35833..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, )?)); @@ -182,7 +180,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 df67da0..9d624e8 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,51 +52,122 @@ struct Writer { device: Arc, id: usize, receiver: Receiver, + priority_receiver: Receiver, eviction: Arc>, + priority_only: bool } impl Writer { fn new( id: usize, receiver: Receiver, + priority_receiver: Receiver, device: Arc, eviction: &Arc>, + priority_only: bool ) -> Self { Self { id, receiver, + priority_receiver, device, eviction: eviction.clone(), + priority_only + } + } + + 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)) { + 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; + } + } + } + } + } + fn receive_priority(&self) { + while let Ok(batch_msg) = self.priority_receiver.recv() { + self.process_batch_request(batch_msg); } } fn run(self) { - tracing::debug!("Writer {} started", self.id); - while let Ok(msg) = self.receiver.recv() { - // println!("Writer {} processing: {:?}", self.id, msg); + tracing::info!("Writer {} started", self.id); + if self.priority_only { + self.receive_priority(); + } else { + self.receive_all(); + } + tracing::info!("Writer {} exiting", self.id); + } - let start = std::time::Instant::now(); + 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); + 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() - ); + 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() { + + // 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); + let mut policy = mtx.lock().unwrap(); + policy.write_update(loc.clone()); + drop(policy); + } else { + tracing::error!("Failed to append: {:?}", result); + } + + 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"); } - tracing::info!("Writer {} exiting", self.id); } } @@ -92,6 +175,7 @@ impl Writer { #[derive(Debug)] pub struct WriterPool { sender: Sender, + priority_sender: Sender, handles: Vec>, } @@ -103,20 +187,32 @@ impl WriterPool { eviction_policy: &Arc>, ) -> 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 writer = Writer::new(id, rx_clone, device.clone(), eviction_policy); + let priority_rx_clone = priority_receiver.clone(); + + // 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); } - Self { sender, handles } + Self { + sender, + priority_sender, + handles, + } } /// 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 +222,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 +233,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 diff --git a/oxcache/src/zone_state/zone_list.rs b/oxcache/src/zone_state/zone_list.rs index 4931ea1..5dd6a12 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 { @@ -596,20 +617,20 @@ 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 {} 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 }) } @@ -624,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 d7b0cec..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 { @@ -114,43 +115,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, @@ -164,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!() } 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" 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