diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ccdffa1..2cb7ced 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -152,6 +152,8 @@ See [adaptive fill admission](CONFIGURATION.md#adaptive-fill-admission) for rate The managed-memory limit covers the index mapping, heat bits, L1, append buffers, reclaim buffers, metadata, cache-owned thread stacks, recovery scratch, and transient reads. Total deployment memory additionally includes allocator metadata, Tokio, process overhead, and the kernel page cache. `CacheConfig::new` rejects invalid or insufficient memory budgets before file access; actual allocation can still fail during open. +`MemoryReservation` holds a charge until its allocation is released. `BufferLease` owns an aligned allocation directly and drops it before returning an optional individual charge; append staging instead keeps one aggregate reservation for its fixed buffers and record arrays. A failed buffer allocation releases its reservation automatically. + ### Storage path C² owns one logical data path. Multi-device deployments stripe below the filesystem with RAID0 or an equivalent layer. Request routing, recovery identity, and descriptor count stay independent of device topology. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7643719..ef442ed 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -33,6 +33,8 @@ Follow the surrounding code and the design constraints in [ARCHITECTURE.md](ARCH Name types by their concrete domain role. Use `Options` for editable inputs awaiting validation and `Config` for validated or resolved configuration. Keep names such as `Layout` for geometry and use specific domain nouns such as `Desc`, `Candidate`, `Victims`, or `Inspection` for runtime data. Avoid `Plan` as a type-name suffix; computing data before using it does not by itself make that data a plan. Keep related fields, local variables, and functions consistent with the type's role. +Use ordinary names for fields; annotate required but unread ownership fields with a narrowly scoped `#[expect(dead_code)]` and explain their role with `reason`. Local guards and other resources retained until scope exit may use conventional underscore-prefixed bindings such as `_guard` without a lint attribute. Preserve their lifetimes and drop order; the wildcard pattern `_` does not retain a value. Remove unnecessary bindings and private parameters, and use a narrowly scoped `#[expect(unused_variables)]` for required but unused trait parameters. + Declare restricted visibility at module boundaries and use `pub` for items in those modules' APIs. Keep items private when only their defining module and its descendants need them. For items reachable through public modules or re-exported public types, reserve `pub` for intentional public API and use narrower visibility for internal callers. ## Documentation diff --git a/benchmarks/cache/main.rs b/benchmarks/cache/main.rs index 98bce7e..9cf057b 100644 --- a/benchmarks/cache/main.rs +++ b/benchmarks/cache/main.rs @@ -408,7 +408,6 @@ async fn run(config: BenchConfig) -> io::Result<()> { .transpose()?; report( "put_drain", - "put + drain", "write", config.write_clients, &write.measurement, @@ -437,7 +436,7 @@ async fn run(config: BenchConfig) -> io::Result<()> { cache.close_warm().await?; drop(cache); let warm_close = started.elapsed(); - report_latency("warm_close", "warm close", warm_close); + report_latency("warm_close", warm_close); let cache = Arc::new(Cache::open(files.data(), cache_config.clone()).await?); if cache.startup_mode() != StartupMode::Warm { @@ -459,7 +458,6 @@ async fn run(config: BenchConfig) -> io::Result<()> { if l1_entry_eligible { report( "l2_promote", - "L2 get + promote", "read", config.l2_clients(), &l2_read.measurement, @@ -470,7 +468,6 @@ async fn run(config: BenchConfig) -> io::Result<()> { } else { report( "l2_read", - "L2 get", "read", config.l2_clients(), &l2_read.measurement, @@ -485,7 +482,7 @@ async fn run(config: BenchConfig) -> io::Result<()> { l2_read.measurement.elapsed, &l2_read.primary, ); - report_read_latency("l2_latency", "L2 get latency", &l2_read.latency); + report_read_latency("l2_latency", &l2_read.latency); l2_read.measurement } else { let _ = concurrent_writes( @@ -522,7 +519,6 @@ async fn run(config: BenchConfig) -> io::Result<()> { .await?; report( "l2_hot_scan", - "L2 cold scan", "read", config.l2_clients(), &cold_scan.measurement, @@ -536,7 +532,7 @@ async fn run(config: BenchConfig) -> io::Result<()> { cold_scan.measurement.elapsed, &cold_scan.primary, ); - report_read_latency("l2_scan_latency", "L2 scan latency", &cold_scan.latency); + report_read_latency("l2_scan_latency", &cold_scan.latency); report_tiers( "hot_during_scan", "hot during scan", @@ -668,7 +664,6 @@ async fn run(config: BenchConfig) -> io::Result<()> { .await?; report( "resident_l1", - "resident L1 get", "read", config.clients, &resident.measurement, @@ -676,7 +671,7 @@ async fn run(config: BenchConfig) -> io::Result<()> { Some(&resident.latency), config.read_latency_sample_interval, ); - report_read_latency("resident_l1_latency", "L1 get latency", &resident.latency); + report_read_latency("resident_l1_latency", &resident.latency); if let Some(before) = before { let after = cache.snapshot()?; println!( @@ -961,10 +956,8 @@ fn verify_value(ordinal: usize, value: &[u8]) -> io::Result<()> { Ok(()) } -#[allow(clippy::too_many_arguments)] fn report( phase: &str, - _name: &str, operation: &str, workers: usize, measurement: &Measurement, @@ -1032,7 +1025,7 @@ fn report_tiers(phase: &str, name: &str, elapsed: Duration, tiers: &TierCounts) ); } -fn report_latency(phase: &str, _name: &str, elapsed: Duration) { +fn report_latency(phase: &str, elapsed: Duration) { JobReport::new("cache", None, phase, "control", elapsed, 1).emit(); println!( "result phase={phase} elapsed_ns={} operations=0 bytes=0 ops_per_sec=0.000 mib_per_sec=0.000 checksum=0000000000000000", @@ -1040,7 +1033,7 @@ fn report_latency(phase: &str, _name: &str, elapsed: Duration) { ); } -fn report_read_latency(phase: &str, _name: &str, latency: &LatencyHistogram) { +fn report_read_latency(phase: &str, latency: &LatencyHistogram) { let summary = latency.summary(); if summary.samples == 0 { return; diff --git a/cache2/src/cache/runtime/mod.rs b/cache2/src/cache/runtime/mod.rs index ad068a0..6c1d4cf 100644 --- a/cache2/src/cache/runtime/mod.rs +++ b/cache2/src/cache/runtime/mod.rs @@ -883,12 +883,12 @@ impl CacheRuntime { current_bytes, } => { if ADMIT_L1 { - let _published = state.memory.publish(hash, key, value, seqno); + state.memory.publish(hash, key, value, seqno); } else { // Prevent an older exact-key L1 value from indefinitely // shadowing the prefetched L2 record. Contention remains a // valid best-effort stale outcome. - let _removed = state.memory.delete(hash, key, seqno); + state.memory.delete(hash, key, seqno); } if should_wake_write( previous_bytes, @@ -940,7 +940,7 @@ impl CacheRuntime { } return Err(write_overload_error()); }; - let _removed = state.memory.delete(hash, key, seqno); + state.memory.delete(hash, key, seqno); if let Some(activity) = activity { RuntimeMetrics::increment(&activity.deletes); } diff --git a/cache2/src/io/engine/tests.rs b/cache2/src/io/engine/tests.rs index cfce4f8..5c6d912 100644 --- a/cache2/src/io/engine/tests.rs +++ b/cache2/src/io/engine/tests.rs @@ -118,13 +118,22 @@ impl BlockingIo { } impl PositionedIo for BlockingIo { - fn read_at(&self, buffer: &mut [u8], _offset: u64) -> io::Result { + fn read_at( + &self, + buffer: &mut [u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { self.enter_and_wait(); buffer.fill(0); Ok(buffer.len()) } - fn write_at(&self, _point: WritePoint, buffer: &[u8], _offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + buffer: &[u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { self.enter_and_wait(); Ok(buffer.len()) } @@ -149,7 +158,11 @@ impl PanicOnceIo { } impl PositionedIo for PanicOnceIo { - fn read_at(&self, buffer: &mut [u8], _offset: u64) -> io::Result { + fn read_at( + &self, + buffer: &mut [u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { if self.panic_next_read.swap(false, Ordering::AcqRel) { panic!("injected io panic"); } @@ -157,13 +170,22 @@ impl PositionedIo for PanicOnceIo { Ok(buffer.len()) } - fn write_at(&self, _point: WritePoint, buffer: &[u8], _offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + buffer: &[u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { Ok(buffer.len()) } } impl PositionedIo for ShortThenErrorIo { - fn read_at(&self, buffer: &mut [u8], _offset: u64) -> io::Result { + fn read_at( + &self, + buffer: &mut [u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { if self.read_calls.fetch_add(1, Ordering::Relaxed) == 0 { let transferred = 3.min(buffer.len()); buffer[..transferred].fill(0x5a); @@ -173,7 +195,12 @@ impl PositionedIo for ShortThenErrorIo { } } - fn write_at(&self, _point: WritePoint, buffer: &[u8], _offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + buffer: &[u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { if self.write_calls.fetch_add(1, Ordering::Relaxed) == 0 { Ok(3.min(buffer.len())) } else { @@ -199,7 +226,11 @@ fn read_buffer(managed_memory: &Arc, length: usize) -> IoBuffer { fn write_buffer(managed_memory: &Arc, bytes: &[u8]) -> IoBuffer { let mut lease = managed_memory.try_read_buffer(bytes.len()).unwrap(); - lease.prepare(bytes.len()).unwrap().copy_from_slice(bytes); + let target = lease.read_target(bytes.len()).unwrap(); + // SAFETY: the exclusively owned target fits the source, and the allocations + // do not overlap. Publish the initialized range only after copying it. + unsafe { target.copy_from_nonoverlapping(bytes.as_ptr(), bytes.len()) }; + lease.mark_initialized(bytes.len()).unwrap(); IoBuffer::for_write(lease, bytes.len()).unwrap() } diff --git a/cache2/src/io/file.rs b/cache2/src/io/file.rs index f2a5f7b..4cb2b58 100644 --- a/cache2/src/io/file.rs +++ b/cache2/src/io/file.rs @@ -523,7 +523,7 @@ impl StorageFile for CacheFile { } } - fn sync(&self, _point: SyncPoint, mode: SyncMode) -> io::Result<()> { + fn sync(&self, #[expect(unused_variables)] point: SyncPoint, mode: SyncMode) -> io::Result<()> { match mode { SyncMode::Data => self.file.sync_data(), SyncMode::All => self.file.sync_all(), @@ -572,7 +572,12 @@ impl PositionedIo for CacheFile { self.file.read_at(buffer, offset) } - fn write_at(&self, _point: WritePoint, buffer: &[u8], offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + buffer: &[u8], + offset: u64, + ) -> io::Result { self.file.write_at(buffer, offset) } } @@ -874,7 +879,11 @@ mod tests { } impl PositionedIo for InterruptedIo { - fn read_at(&self, _buffer: &mut [u8], _offset: u64) -> io::Result { + fn read_at( + &self, + #[expect(unused_variables)] buffer: &mut [u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { self.calls.fetch_add(1, Ordering::Relaxed); Err(io::Error::new( io::ErrorKind::Interrupted, @@ -882,7 +891,12 @@ mod tests { )) } - fn write_at(&self, _point: WritePoint, _buffer: &[u8], _offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + #[expect(unused_variables)] buffer: &[u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { self.calls.fetch_add(1, Ordering::Relaxed); Err(io::Error::new( io::ErrorKind::Interrupted, diff --git a/cache2/src/managed_memory.rs b/cache2/src/managed_memory.rs index 3b3dab9..da037f3 100644 --- a/cache2/src/managed_memory.rs +++ b/cache2/src/managed_memory.rs @@ -58,15 +58,14 @@ pub struct ManagedMemory { memory: Arc, } -/// A fixed runtime allocation charged to the same hard memory limit as the -/// request pools. The owner keeps this guard for exactly as long as the -/// associated bounded structure exists. -pub struct RuntimeMemoryReservation { +/// A charge against the cache-wide memory limit. Keep this guard until the +/// associated allocation has been released. +pub struct MemoryReservation { memory: Arc, bytes: usize, } -impl Drop for RuntimeMemoryReservation { +impl Drop for MemoryReservation { fn drop(&mut self) { self.memory.release(self.bytes); } @@ -87,14 +86,11 @@ impl ManagedMemory { Ok(Self { memory }) } - pub fn reserve_runtime_memory( - &self, - bytes: usize, - ) -> Result { + pub fn reserve(&self, bytes: usize) -> Result { if !self.memory.try_reserve(bytes) { return Err(ManagedMemoryError::Allocation); } - Ok(RuntimeMemoryReservation { + Ok(MemoryReservation { memory: Arc::clone(&self.memory), bytes, }) @@ -104,7 +100,16 @@ impl ManagedMemory { /// cache-wide hard memory limit. The caller maps failure to either a /// fail-open miss or an explicit bounded-wait overload. pub fn try_read_buffer(&self, length: usize) -> Option { - BufferLease::try_standalone(length, Arc::clone(&self.memory)) + let capacity = align_up(length, BUFFER_ALIGNMENT)?; + if capacity == 0 || capacity > isize::MAX as usize { + return None; + } + let reservation = self.reserve(capacity).ok()?; + let buffer = AlignedBuffer::try_new(capacity)?; + Some(BufferLease { + buffer, + reservation: Some(reservation), + }) } pub fn snapshot(&self) -> ManagedMemorySnapshot { @@ -125,13 +130,14 @@ pub struct ManagedMemorySnapshot { } pub struct BufferLease { - owner: BufferOwner, - buffer: Option, -} - -enum BufferOwner { - Fixed, - Standalone { memory: Arc }, + // Fields drop in declaration order: free the allocation before returning + // its budget. Fixed staging buffers use their owner's aggregate charge. + buffer: AlignedBuffer, + #[expect( + dead_code, + reason = "Returns the memory charge after the allocation is dropped." + )] + reservation: Option, } impl BufferLease { @@ -141,68 +147,16 @@ impl BufferLease { "fixed buffer size must be a non-zero 4096-byte multiple", )); } - let ptr = allocate_buffer(length).ok_or(ManagedMemoryError::Allocation)?; - let mut buffer = AlignedBuffer { - ptr, - capacity: length, - initialized: 0, - }; + let mut buffer = AlignedBuffer::try_new(length).ok_or(ManagedMemoryError::Allocation)?; buffer.prepare_zeroed(length); Ok(Self { - owner: BufferOwner::Fixed, - buffer: Some(buffer), + buffer, + reservation: None, }) } - fn try_standalone(length: usize, memory: Arc) -> Option { - let maximum = align_up(length, BUFFER_ALIGNMENT)?; - if maximum == 0 || maximum > isize::MAX as usize { - return None; - } - if !memory.try_reserve(maximum) { - return None; - } - let Some(ptr) = allocate_buffer(maximum) else { - memory.release(maximum); - return None; - }; - Some(Self { - owner: BufferOwner::Standalone { memory }, - buffer: Some(AlignedBuffer { - ptr, - capacity: maximum, - initialized: 0, - }), - }) - } - - #[cfg(test)] - pub fn prepare(&mut self, length: usize) -> Result<&mut [u8], ()> { - let buffer = self.buffer.as_mut().expect("buffer lease owns a buffer"); - if length > buffer.capacity { - return Err(()); - } - // Callers encode complete records. Clearing here also fixes padding and - // prevents bytes from a prior key/value escaping into a later write. - buffer.prepare_zeroed(length); - Ok(buffer.prefix_mut(length)) - } - - /// Grow the leased buffer without clearing bytes already in the buffer. - /// - /// Fresh capacity is zeroed before it is exposed as initialized bytes. - #[cfg(test)] - fn grow_preserving(&mut self, length: usize) -> Result<&mut [u8], ()> { - let buffer = self.buffer.as_mut().expect("buffer lease owns a buffer"); - if length > buffer.capacity { - return Err(()); - } - buffer.zero_uninitialized_through(length); - Ok(buffer.prefix_mut(length)) - } - pub fn prepared(&self, length: usize) -> Result<&[u8], ()> { - let buffer = self.buffer.as_ref().ok_or(())?; + let buffer = &self.buffer; if length > buffer.initialized { return Err(()); } @@ -212,7 +166,7 @@ impl BufferLease { } pub fn prepared_mut(&mut self, length: usize) -> Result<&mut [u8], ()> { - let buffer = self.buffer.as_mut().ok_or(())?; + let buffer = &mut self.buffer; if length > buffer.initialized { return Err(()); } @@ -220,13 +174,11 @@ impl BufferLease { } pub fn has_capacity(&self, length: usize) -> bool { - self.buffer - .as_ref() - .is_some_and(|buffer| length <= buffer.capacity) + length <= self.buffer.capacity } pub fn read_target(&self, length: usize) -> Result<*mut u8, ()> { - let buffer = self.buffer.as_ref().ok_or(())?; + let buffer = &self.buffer; if length > buffer.capacity { return Err(()); } @@ -234,7 +186,7 @@ impl BufferLease { } pub fn mark_initialized(&mut self, length: usize) -> Result<(), ()> { - let buffer = self.buffer.as_mut().ok_or(())?; + let buffer = &mut self.buffer; if length > buffer.capacity { return Err(()); } @@ -244,26 +196,7 @@ impl BufferLease { #[cfg(test)] fn address(&self) -> usize { - self.buffer - .as_ref() - .expect("buffer lease owns a buffer") - .ptr - .as_ptr() as usize - } -} - -impl Drop for BufferLease { - fn drop(&mut self) { - if let Some(buffer) = self.buffer.take() { - match &self.owner { - BufferOwner::Fixed => drop(buffer), - BufferOwner::Standalone { memory, .. } => { - let capacity = buffer.capacity; - drop(buffer); - memory.release(capacity); - } - } - } + self.buffer.ptr.as_ptr() as usize } } @@ -278,6 +211,18 @@ struct AlignedBuffer { unsafe impl Send for AlignedBuffer {} impl AlignedBuffer { + fn try_new(capacity: usize) -> Option { + let layout = Layout::from_size_align(capacity, BUFFER_ALIGNMENT).ok()?; + // SAFETY: both constructors validate that capacity is non-zero; the + // layout has valid power-of-two alignment and fits in isize. + let ptr = NonNull::new(unsafe { alloc(layout) })?; + Some(Self { + ptr, + capacity, + initialized: 0, + }) + } + fn prepare_zeroed(&mut self, length: usize) { debug_assert!(length <= self.capacity); // SAFETY: the allocation is valid for `capacity` bytes and this value @@ -287,21 +232,6 @@ impl AlignedBuffer { self.initialized = self.initialized.max(length); } - #[cfg(test)] - fn zero_uninitialized_through(&mut self, length: usize) { - debug_assert!(length <= self.capacity); - if length > self.initialized { - // SAFETY: the uninitialized tail is inside the owned allocation. - unsafe { - self.ptr - .as_ptr() - .add(self.initialized) - .write_bytes(0, length - self.initialized); - } - self.initialized = length; - } - } - fn prefix_mut(&mut self, length: usize) -> &mut [u8] { debug_assert!(length <= self.capacity); debug_assert!(length <= self.initialized); @@ -309,34 +239,15 @@ impl AlignedBuffer { // mutable borrow is exclusive, and `length <= initialized`. unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), length) } } - - fn deallocate(&mut self) { - if self.capacity == 0 { - return; - } - deallocate_buffer(self.ptr, self.capacity); - self.ptr = NonNull::dangling(); - self.capacity = 0; - self.initialized = 0; - } -} - -fn allocate_buffer(capacity: usize) -> Option> { - let layout = Layout::from_size_align(capacity, BUFFER_ALIGNMENT).ok()?; - // SAFETY: `layout` has non-zero size and valid power-of-two alignment. - NonNull::new(unsafe { alloc(layout) }) -} - -fn deallocate_buffer(pointer: NonNull, capacity: usize) { - let layout = Layout::from_size_align(capacity, BUFFER_ALIGNMENT) - .expect("stored aligned-buffer layout is valid"); - // SAFETY: `pointer` was allocated with this exact layout and is owned here. - unsafe { dealloc(pointer.as_ptr(), layout) }; } impl Drop for AlignedBuffer { fn drop(&mut self) { - self.deallocate(); + let layout = Layout::from_size_align(self.capacity, BUFFER_ALIGNMENT) + .expect("stored aligned-buffer layout is valid"); + // SAFETY: the pointer was allocated with this exact layout and this + // buffer owns it until drop. + unsafe { dealloc(self.ptr.as_ptr(), layout) }; } } @@ -436,25 +347,42 @@ mod tests { } #[test] - fn preserving_growth_keeps_prefix_and_zeroes_fresh_capacity() { - let mut buffer = BufferLease::try_fixed(3 * BUFFER_ALIGNMENT).unwrap(); - - let first = buffer.grow_preserving(BUFFER_ALIGNMENT).unwrap(); - assert!(first.iter().all(|byte| *byte == 0)); - first.fill(0x5a); - - let grown = buffer.grow_preserving(2 * BUFFER_ALIGNMENT).unwrap(); - assert!(grown[..BUFFER_ALIGNMENT].iter().all(|byte| *byte == 0x5a)); - assert!(grown[BUFFER_ALIGNMENT..].iter().all(|byte| *byte == 0)); - assert_eq!(buffer.address() % BUFFER_ALIGNMENT, 0); + fn read_buffers_expose_only_the_initialized_prefix() { + let managed_memory = ManagedMemory::try_new(limits()).unwrap(); + let mut buffer = managed_memory.try_read_buffer(5000).unwrap(); + let bytes = b"completed read prefix"; + let target = buffer.read_target(5000).unwrap(); + // SAFETY: simulate a completed read into the exclusively owned target; + // the source and destination do not overlap and the bytes fit. + unsafe { target.copy_from_nonoverlapping(bytes.as_ptr(), bytes.len()) }; + + assert!(buffer.prepared(bytes.len()).is_err()); + buffer.mark_initialized(bytes.len()).unwrap(); + assert_eq!(buffer.prepared(bytes.len()).unwrap(), bytes); + assert!(buffer.prepared(bytes.len() + 1).is_err()); + assert!(buffer.prepared_mut(bytes.len() + 1).is_err()); + + // A shorter read can reuse the buffer without invalidating its prefix. + buffer.mark_initialized(1).unwrap(); + assert_eq!(buffer.prepared(bytes.len()).unwrap(), bytes); + assert!(buffer.mark_initialized(2 * BUFFER_ALIGNMENT + 1).is_err()); + assert!(buffer.prepared(bytes.len() + 1).is_err()); } #[test] - fn failed_preserving_growth_keeps_the_existing_buffer() { - let mut buffer = BufferLease::try_fixed(2 * BUFFER_ALIGNMENT).unwrap(); - buffer.grow_preserving(BUFFER_ALIGNMENT).unwrap().fill(0xa5); + fn fixed_buffers_are_zeroed_and_reject_out_of_bounds_access() { + let mut buffer = BufferLease::try_fixed(BUFFER_ALIGNMENT).unwrap(); + assert!( + buffer + .prepared(BUFFER_ALIGNMENT) + .unwrap() + .iter() + .all(|byte| *byte == 0) + ); + buffer.prepared_mut(BUFFER_ALIGNMENT).unwrap().fill(0xa5); - assert!(buffer.grow_preserving(3 * BUFFER_ALIGNMENT).is_err()); + assert!(buffer.read_target(BUFFER_ALIGNMENT + 1).is_err()); + assert!(buffer.prepared_mut(BUFFER_ALIGNMENT + 1).is_err()); assert!( buffer .prepared(BUFFER_ALIGNMENT) diff --git a/cache2/src/memory/mod.rs b/cache2/src/memory/mod.rs index 037e868..1eaf1e1 100644 --- a/cache2/src/memory/mod.rs +++ b/cache2/src/memory/mod.rs @@ -167,7 +167,7 @@ struct MemoryCharge { struct MemoryValueInner { bytes: Box<[u8]>, key_length: usize, - _charge: MemoryCharge, + charge: MemoryCharge, } #[derive(Clone)] @@ -197,7 +197,7 @@ impl MemoryValue { Ok(Self(Arc::new(MemoryValueInner { bytes: bytes.into_boxed_slice(), key_length: key.len(), - _charge: charge, + charge, }))) } @@ -225,7 +225,7 @@ impl MemoryValue { fn disarm_exclusive_charge(&mut self) -> usize { let inner = Arc::get_mut(&mut self.0).expect("exclusive resident value gained an owner"); - inner._charge.disarm() + inner.charge.disarm() } } diff --git a/cache2/src/region/appender.rs b/cache2/src/region/appender.rs index 53f9091..d7d898e 100644 --- a/cache2/src/region/appender.rs +++ b/cache2/src/region/appender.rs @@ -262,7 +262,11 @@ mod tests { } impl PositionedIo for RecordingIo { - fn read_at(&self, _buffer: &mut [u8], _offset: u64) -> io::Result { + fn read_at( + &self, + #[expect(unused_variables)] buffer: &mut [u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { Err(io::Error::new(io::ErrorKind::Unsupported, "read unused")) } @@ -305,7 +309,7 @@ mod tests { }); let engine = IoEngine::for_test(io.clone(), 1).unwrap(); let mut lease = BufferLease::try_fixed(4096).unwrap(); - lease.prepare(4096).unwrap().fill(0x5a); + lease.prepared_mut(4096).unwrap().fill(0x5a); let absolute = DATA_REGION_AREA_OFFSET + geometry().region_size; let io_recovery = IoRecovery::new(Some(Duration::from_secs(5))); let mut attempt = BackgroundIoAttempt::new(&io_recovery, None); @@ -337,7 +341,7 @@ mod tests { let io = Arc::new(RecordingIo::default()); let engine = IoEngine::for_test(io.clone(), 1).unwrap(); let mut lease = BufferLease::try_fixed(4096).unwrap(); - lease.prepare(4096).unwrap().fill(0x5a); + lease.prepared_mut(4096).unwrap().fill(0x5a); let buffer = IoBuffer::for_write(lease, 4096).unwrap(); let absolute = DATA_REGION_AREA_OFFSET + geometry().region_size; diff --git a/cache2/src/region/index/storage/mod.rs b/cache2/src/region/index/storage/mod.rs index 6a4e059..6a51531 100644 --- a/cache2/src/region/index/storage/mod.rs +++ b/cache2/src/region/index/storage/mod.rs @@ -641,7 +641,7 @@ impl IndexStorage { let data_pointer = backing.as_mut_ptr(); let page_states = allocate_page_states(layout.page_count, PAGE_STATE_VALID)?; let core = Arc::new(IndexStorageCore { - _backing: UnsafeCell::new(backing), + backing: UnsafeCell::new(backing), data_pointer, data_offset: 0, slot_count, @@ -706,7 +706,7 @@ impl IndexStorage { let data_pointer = backing.as_mut_ptr(); let page_states = allocate_page_states(layout.page_count, PAGE_STATE_UNCHECKED)?; let core = Arc::new(IndexStorageCore { - _backing: UnsafeCell::new(backing), + backing: UnsafeCell::new(backing), data_pointer, data_offset, slot_count, @@ -920,7 +920,11 @@ impl IndexStorage { } struct IndexStorageCore { - _backing: UnsafeCell, + #[expect( + dead_code, + reason = "Owns the mapping accessed through the cached data pointer." + )] + backing: UnsafeCell, data_pointer: *mut u8, data_offset: usize, slot_count: usize, @@ -1146,7 +1150,7 @@ impl IndexStorageCore { fn data_ptr(&self) -> *const u8 { // SAFETY: `data_pointer` addresses the stable allocation owned by - // `_backing`, and `data_offset + image_len` was checked against that + // `backing`, and `data_offset + image_len` was checked against that // allocation during construction. unsafe { self.data_pointer.cast_const().add(self.data_offset) } } diff --git a/cache2/src/region/persistence/tests.rs b/cache2/src/region/persistence/tests.rs index 3836ad6..733e2f0 100644 --- a/cache2/src/region/persistence/tests.rs +++ b/cache2/src/region/persistence/tests.rs @@ -369,7 +369,7 @@ fn run_crash_child(case: &str, paths: RegionPaths) -> ! { let data = data_path_superblock(); match case { "open" => { - let _store = CacheSession::for_test(paths, data, 4096).unwrap(); + let _session = CacheSession::for_test(paths, data, 4096).unwrap(); kill_process(); } "write" | "drain" => { @@ -913,8 +913,7 @@ fn completed_owned_span_publishes_index_without_a_steady_state_sync() { outcome => panic!("unexpected staging outcome: {outcome:?}"), } } - let (first_key, first_hash, _first_seqno) = - first.expect("4 MiB span must contain target-size records"); + let (first_key, first_hash, _) = first.expect("4 MiB span must contain target-size records"); let (last_key, last_hash, last_seqno) = last.expect("4 MiB span must retain its final record"); assert!(staged_records > 240); assert_eq!(regions.lookup_snapshot(first_hash).unwrap(), None); diff --git a/cache2/src/region/reader.rs b/cache2/src/region/reader.rs index 4245027..6000b60 100644 --- a/cache2/src/region/reader.rs +++ b/cache2/src/region/reader.rs @@ -346,7 +346,12 @@ mod tests { Ok(buffer.len()) } - fn write_at(&self, _point: WritePoint, _buffer: &[u8], _offset: u64) -> io::Result { + fn write_at( + &self, + #[expect(unused_variables)] point: WritePoint, + #[expect(unused_variables)] buffer: &[u8], + #[expect(unused_variables)] offset: u64, + ) -> io::Result { Err(io::Error::new(io::ErrorKind::Unsupported, "write unused")) } } diff --git a/cache2/src/region/staging.rs b/cache2/src/region/staging.rs index bc229e6..95faaf0 100644 --- a/cache2/src/region/staging.rs +++ b/cache2/src/region/staging.rs @@ -27,7 +27,7 @@ use crate::managed_memory::BUFFER_ALIGNMENT; use crate::managed_memory::BufferLease; use crate::managed_memory::ManagedMemory; use crate::managed_memory::ManagedMemoryError; -use crate::managed_memory::RuntimeMemoryReservation; +use crate::managed_memory::MemoryReservation; use crate::region::index::packed::IndexEntry; use crate::region::index::packed::MAX_RECORD_LEN; use crate::region::index::packed::PackedLocation; @@ -252,7 +252,11 @@ pub struct AppendStaging { shards: Vec, chunk_bytes: usize, region_size: u64, - _memory: RuntimeMemoryReservation, + #[expect( + dead_code, + reason = "Returns the aggregate charge when staging is dropped." + )] + reservation: MemoryReservation, } impl AppendStaging { @@ -301,7 +305,7 @@ impl AppendStaging { .ok_or(ManagedMemoryError::Allocation)?; // Keep the aggregate reservation alive so eager buffers and record // vectors participate in the hard memory limit. - let memory = managed_memory.reserve_runtime_memory(reserved)?; + let reservation = managed_memory.reserve(reserved)?; let mut shards = Vec::new(); shards @@ -329,7 +333,7 @@ impl AppendStaging { shards, chunk_bytes, region_size, - _memory: memory, + reservation, }) }