From 18dc51a4eff065f38bbec383de266842cb4c7c2d Mon Sep 17 00:00:00 2001 From: leiysky Date: Wed, 28 Oct 2026 21:04:30 +0800 Subject: [PATCH] perf: reduce redundant cache path work --- ARCHITECTURE.md | 4 +- CHANGELOG.md | 2 +- cache2/src/cache/runtime/mod.rs | 37 ++-- cache2/src/memory/mod.rs | 4 +- cache2/src/region/index/storage/mod.rs | 11 +- cache2/src/region/mod.rs | 45 +++-- cache2/src/region/persistence/tests.rs | 251 +++++++++++++------------ 7 files changed, 183 insertions(+), 171 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 4d7e218..873f560 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -126,7 +126,9 @@ Delete allocates a sequence number, performs one bounded in-memory index removal When free capacity is low, a reclaim worker takes the oldest sealed Region, reads its used prefix sequentially, and walks the self-sized records. A mapping is changed only if the index still points at that exact Region address. -The first L2 candidate access sets `seen`; a later access sets `hot`. Reclaim clears these volatile bits. Cold current mappings are removed. Hot current records may be rewritten through an existing append shard while preserving their logical sequence number. Payload validation produces a borrowed payload and checksum that staging reuses without scanning the payload again; the replacement header receives its own checksum. Reinsertion is capped at roughly one eighth of the reclaimed bytes and rotates successive source Regions across append shards. An empty Free queue disables reinsertion for that reclaim pass so foreground write capacity is restored first; staging or budget pressure drops the remaining candidates. +The first L2 candidate access sets `seen`; a later access sets `hot`. Reclaim clears these volatile bits. Cold current mappings are removed. Hot current records may be rewritten through an existing append shard while preserving their logical sequence number. Payload validation produces a borrowed payload and checksum that staging reuses without scanning the payload again; the replacement header receives its own checksum. + +Reinsertion is capped at roughly one eighth of the reclaimed bytes and rotates successive source Regions across append shards. An empty Free queue disables reinsertion for that reclaim pass so foreground write capacity is restored first. Staging pressure stops further reinsertion from that source Region; the byte budget is checked per record. Once reinsertion stops, records within the byte budget skip payload checksums but retain header validation and address-conditional removal. Payloads exceeding the budget are still validated so the budget-skipped count includes only valid records. The source Region stays pinned until accepted replacement writes finish. Conditional index replacement lets a newer foreground put or delete win over a delayed reinsert. Only then is the Region returned to the free FIFO. Reclaim uses background I/O lanes and bounded single-pass work. `RuntimeOptions::reclaim_io_timeout` bounds each reclaim read from I/O admission through completion, defaulting to five seconds; foreground I/O deadlines are independent. Expiration enters reversible background I/O recovery. Long recovery can exhaust free capacity, but close interrupts recovery before draining workers. diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a3aae8..03254d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ - A zero-capacity L1 bypasses shard locking and directory work for lookups, fills, promotions, and deletes. - Memory accounting skips redundant atomic updates when transient reads stay within the recorded peak or L1 transfers leave the charged byte total unchanged. - L2 index updates and cold-record reclamation avoid redundant atomic writes for already-clear heat bits and validated index pages. -- Hot-record reclamation reuses the checksum computed during payload validation for its rewrite, avoiding a second payload scan while preserving record validation and the on-disk format. +- Hot-record reclamation reuses validated payload checksums for rewrites and skips payload checksum scans for records within the byte budget after reinsertion stops, while preserving record header validation, conditional removal, budget statistics, and the on-disk format. - L1 replacement and eviction reuse detached, exclusively owned allocations of the same size under CLOCK and S3-FIFO, reducing repeated allocation during fixed-size turnover without an additional buffer pool. Caller-held values remain immutable and retain their separate memory charge. When activity statistics are enabled, mutations with no eviction also avoid a shared zero-value counter update. ## v0.6.0 (2026-09-21) diff --git a/cache2/src/cache/runtime/mod.rs b/cache2/src/cache/runtime/mod.rs index 7b38b39..5adecc4 100644 --- a/cache2/src/cache/runtime/mod.rs +++ b/cache2/src/cache/runtime/mod.rs @@ -1871,29 +1871,20 @@ fn reclaim_worker_result( } else { None }; - let mut accepting_reinserts = reinsert_operation.is_some(); - let mut staged_reinsert = false; - let stats = state.regions.scan_reclaim(receipt, bytes, |record| { - if !accepting_reinserts { - return Ok(false); - } - match state - .regions - .try_stage_reinsert(&state.staging, reinsert_shard, record)? - { - RegionStageValue::Staged { .. } => { - staged_reinsert = true; - Ok(true) - } - RegionStageValue::Contended - | RegionStageValue::NeedsFlush - | RegionStageValue::NeedsRotation => { - accepting_reinserts = false; - Ok(false) - } - } - })?; - if staged_reinsert { + let stats = state.regions.scan_reclaim( + receipt, + bytes, + reinsert_operation.is_some(), + |record| { + Ok(matches!( + state + .regions + .try_stage_reinsert(&state.staging, reinsert_shard, record,)?, + RegionStageValue::Staged { .. } + )) + }, + )?; + if stats.reinsert_records != 0 { let generation = state.append_controls[reinsert_shard].request_drain(false)?; state.append_controls[reinsert_shard].wait_for_drain(generation)?; } diff --git a/cache2/src/memory/mod.rs b/cache2/src/memory/mod.rs index fcaefb5..d529b73 100644 --- a/cache2/src/memory/mod.rs +++ b/cache2/src/memory/mod.rs @@ -729,7 +729,7 @@ impl MemoryShard { let Ok(memory_value) = MemoryValue::try_new(key, value, seqno, charge, recycled) else { return MemoryInsertResult::rejected(evictions); }; - let previous_head = self.directory_head(hash); + let previous_head = self.directory_upsert(hash, packed_index); let entry = MemoryEntry { value: memory_value, hash_next: previous_head.unwrap_or(NO_SLOT_INDEX), @@ -741,8 +741,6 @@ impl MemoryShard { self.resident_bytes = self.resident_bytes.saturating_add(charged_bytes); self.eviction .insert(&mut self.policy_slots, index, hash, charged_bytes); - let previous = self.directory_upsert(hash, packed_index); - debug_assert_eq!(previous, previous_head); MemoryInsertResult { slot: Some(packed_index), evictions, diff --git a/cache2/src/region/index/storage/mod.rs b/cache2/src/region/index/storage/mod.rs index 148ee6f..0be7cbc 100644 --- a/cache2/src/region/index/storage/mod.rs +++ b/cache2/src/region/index/storage/mod.rs @@ -764,15 +764,14 @@ impl IndexStorage { // The exclusive partition guard keeps the state observed by the probe // stable, so updating its physical count needs no second slot decode. let global_slot = self.global_slot(slot)?; - let previous = IndexSlot::from_state(previous); + let previous_kind = match previous { + IndexSlotState::Empty => SlotPhysicalKind::Empty, + IndexSlotState::Value { .. } => SlotPhysicalKind::Value, + }; let value = IndexSlot::from_state(state); let next_stats = self .physical_stats - .transitioned( - previous.physical_kind(), - value.physical_kind(), - self.range.slot_count, - ) + .transitioned(previous_kind, value.physical_kind(), self.range.slot_count) .ok_or(IndexStorageError::InvalidPhysicalStats)?; self.core.write_slot(global_slot, value)?; self.physical_stats = next_stats; diff --git a/cache2/src/region/mod.rs b/cache2/src/region/mod.rs index 0653447..f11c90f 100644 --- a/cache2/src/region/mod.rs +++ b/cache2/src/region/mod.rs @@ -461,12 +461,14 @@ impl RegionStore { } /// Scans one exact sealed prefix, removes cold mappings, and offers each - /// hot current record once to a bounded reinsertion sink. The - /// source Region remains exclusively pinned until [`Self::complete_reclaim`]. + /// hot current record once to a bounded reinsertion sink. A false sink + /// result stops further offers from this source Region. The source stays + /// exclusively pinned until [`Self::complete_reclaim`]. pub fn scan_reclaim( &self, receipt: RegionReclaimReceipt, bytes: &[u8], + mut accepting_reinserts: bool, mut try_reinsert: impl FnMut(RegionReinsertRecord<'_>) -> io::Result, ) -> io::Result { if bytes.len() as u64 != receipt.used_offset { @@ -563,22 +565,29 @@ impl RegionStore { "reclaim replacement record is too large", ) })?; - let payload = RecordPayload::validate(key, value, header.payload_crc); let within_budget = u64::from(rewrite_bytes) <= reinsert_budget; - let budget_exhausted = payload.is_some() && !within_budget; - let reinserted = if let Some(payload) = payload - && within_budget + // Once the sink stops, in-budget records will only be removed. + // Over-budget records still need validation because the + // budget-skipped counter excludes corrupt payloads. + let mut reinserted = false; + if (accepting_reinserts || !within_budget) + && let Some(payload) = + RecordPayload::validate(key, value, header.payload_crc) { - try_reinsert(RegionReinsertRecord { - hash: header.key_hash, - previous_location: location, - logical_seqno: header.seqno, - record_bytes: rewrite_bytes, - payload, - })? - } else { - false - }; + if within_budget { + reinserted = try_reinsert(RegionReinsertRecord { + hash: header.key_hash, + previous_location: location, + logical_seqno: header.seqno, + record_bytes: rewrite_bytes, + payload, + })?; + accepting_reinserts = reinserted; + } else { + stats.reinsert_budget_skipped = + stats.reinsert_budget_skipped.saturating_add(1); + } + } if reinserted { reinsert_budget = reinsert_budget.saturating_sub(u64::from(rewrite_bytes)); stats.reinsert_records = stats.reinsert_records.saturating_add(1); @@ -586,10 +595,6 @@ impl RegionStore { .reinsert_bytes .saturating_add(u64::from(rewrite_bytes)); } else { - if budget_exhausted { - stats.reinsert_budget_skipped = - stats.reinsert_budget_skipped.saturating_add(1); - } if self .index .remove_if_match(header.key_hash, location) diff --git a/cache2/src/region/persistence/tests.rs b/cache2/src/region/persistence/tests.rs index 0c0e625..fe1b4f1 100644 --- a/cache2/src/region/persistence/tests.rs +++ b/cache2/src/region/persistence/tests.rs @@ -1004,134 +1004,151 @@ fn completed_owned_span_publishes_index_without_a_steady_state_sync() { #[test] fn reclaim_validates_payload_before_staging_and_counts_rewrites() { let key = b"retained-reclaim"; - for (value_len, corrupt_byte, reinserted, budget_skips) in [ + for (value_len, corrupt_byte, payload_eligible, budget_skips) in [ (16 * 1024 + 7, None, true, 0), (16 * 1024 + 7, Some(0), false, 0), (16 * 1024 + 7, Some(key.len()), false, 0), (64 * 1024, None, false, 1), (64 * 1024, Some(key.len()), false, 0), ] { - let (data, regions, staging) = foreground_stage_fixture(); - let value = vec![0xa7; value_len]; - let cold_key = b"cold-reclaim"; - let cold_value = vec![0x31; 192 * 1024]; - let cold_len = required_record_bytes(cold_key.len(), cold_value.len()).unwrap() as usize; - let rewrite_bytes = required_record_bytes(key.len(), value.len()).unwrap(); - let used = (cold_len + rewrite_bytes as usize).next_multiple_of(RECOVERY_PAGE_SIZE); - let mut bytes = vec![0_u8; used]; - // Model an already-read source prefix. Region selection and release are - // covered separately; this test exercises scanning through staged I/O. - let receipt = RegionReclaimReceipt { - region_id: 1, - created_seqno: 2, - used_offset: used as u64, - physical_record_count: 2, - }; - for (offset, length, key, value, seqno) in [ - (0, cold_len, &cold_key[..], &cold_value[..], 2), - (cold_len, used - cold_len, &key[..], &value[..], 3), - ] { - let payload = [key, value].concat(); - let header = RecordHeader { - key_len: key.len() as u16, - value_len: value.len() as u32, - seqno, - key_hash: hash_key(data.hash_seed, key), - payload_crc: crc_fast::crc32_iscsi(&payload), - region_generation: receipt.created_seqno, - record_len: length as u32, + // Exercise each payload with an accepting, rejecting, and disabled sink. + for (accepting_reinserts, sink_accepts) in [(true, true), (true, false), (false, false)] { + let reinserted = accepting_reinserts && sink_accepts && payload_eligible; + let (data, regions, staging) = foreground_stage_fixture(); + let value = vec![0xa7; value_len]; + let cold_key = b"cold-reclaim"; + let cold_value = vec![0x31; 192 * 1024]; + let cold_len = + required_record_bytes(cold_key.len(), cold_value.len()).unwrap() as usize; + let rewrite_bytes = required_record_bytes(key.len(), value.len()).unwrap(); + let used = (cold_len + rewrite_bytes as usize).next_multiple_of(RECOVERY_PAGE_SIZE); + let mut bytes = vec![0_u8; used]; + // Model an already-read source prefix. Region selection and release are + // covered separately; this test exercises scanning through staged I/O. + let receipt = RegionReclaimReceipt { + region_id: 1, + created_seqno: 2, + used_offset: used as u64, + physical_record_count: 2, }; - bytes[offset..offset + RECORD_HEADER_SIZE].copy_from_slice(&header.encode()); - bytes[offset + RECORD_HEADER_SIZE..offset + RECORD_HEADER_SIZE + payload.len()] - .copy_from_slice(&payload); + for (offset, length, key, value, seqno) in [ + (0, cold_len, &cold_key[..], &cold_value[..], 2), + (cold_len, used - cold_len, &key[..], &value[..], 3), + ] { + let payload = [key, value].concat(); + let header = RecordHeader { + key_len: key.len() as u16, + value_len: value.len() as u32, + seqno, + key_hash: hash_key(data.hash_seed, key), + payload_crc: crc_fast::crc32_iscsi(&payload), + region_generation: receipt.created_seqno, + record_len: length as u32, + }; + bytes[offset..offset + RECORD_HEADER_SIZE].copy_from_slice(&header.encode()); + bytes[offset + RECORD_HEADER_SIZE..offset + RECORD_HEADER_SIZE + payload.len()] + .copy_from_slice(&payload); + regions + .index + .upsert( + header.key_hash, + IndexEntry { + location: PackedLocation::new( + receipt.region_id, + offset as u32, + length as u32, + ) + .unwrap(), + }, + ) + .unwrap(); + } + let hash = hash_key(data.hash_seed, key); + assert!(regions.begin_point_read(hash).is_some()); + assert!(regions.begin_point_read(hash).is_some()); + let previous = regions.lookup_snapshot(hash).unwrap(); + if let Some(corrupt_byte) = corrupt_byte { + bytes[cold_len + RECORD_HEADER_SIZE + corrupt_byte] ^= 1; + } + let mut offered = 0; + let stats = regions + .scan_reclaim(receipt, &bytes, accepting_reinserts, |record| { + offered += 1; + assert!( + payload_eligible, + "invalid or over-budget payload reached staging" + ); + if !sink_accepts { + return Ok(false); + } + assert!(matches!( + regions.try_stage_reinsert(&staging, 0, record)?, + RegionStageValue::Staged { seqno, .. } if seqno > 3 + )); + Ok(true) + }) + .unwrap(); + assert_eq!( + offered, + usize::from(accepting_reinserts && payload_eligible) + ); + assert_eq!(stats.bytes_read, receipt.used_offset); + assert_eq!(stats.records_scanned, 2); + assert_eq!(stats.records_removed, 2 - u64::from(reinserted)); + assert_eq!(stats.reinsert_records, u64::from(reinserted)); + assert_eq!( + stats.reinsert_bytes, + u64::from(reinserted) * u64::from(rewrite_bytes) + ); + assert_eq!(stats.reinsert_skipped, u64::from(!reinserted)); + assert_eq!(stats.reinsert_budget_skipped, budget_skips); + if !reinserted { + assert_eq!(regions.lookup_snapshot(hash).unwrap(), None); + assert!(staging.shard_fill_snapshot(0).unwrap().is_none()); + continue; + } + assert_eq!(regions.lookup_snapshot(hash).unwrap(), previous); + assert!( + used - cold_len > rewrite_bytes as usize, + "source has flush padding" + ); + // Destroy the source before I/O: staging must own the validated bytes. + bytes.fill(0xcc); + drop(bytes); + let directory = TestDirectory::new(); + let (file, _) = FaultFile::open(&directory.paths.data).unwrap(); + file.set_len(data.geometry.data_file_len).unwrap(); + let engine = IoEngine::for_test(Arc::new(file), 1).unwrap(); regions - .index - .upsert( - header.key_hash, - IndexEntry { - location: PackedLocation::new( - receipt.region_id, - offset as u32, - length as u32, - ) - .unwrap(), - }, + .flush_staging_shard( + &staging, + &engine, + 0, + crate::io::background::BackgroundIoAttempt::new( + &crate::io::background::IoRecovery::new(Some(Duration::ZERO)), + None, + ), ) + .unwrap() .unwrap(); + let current = regions.begin_point_read(hash).unwrap(); + assert_ne!(current.region_generation, receipt.created_seqno); + let desc = describe_read(data.geometry, hash, current, false).unwrap(); + let memory = data_path_memory(); + let hit = regions + .read_value_from_desc( + &engine, + engine.try_reserve_read().unwrap(), + memory.try_read_buffer(desc.read_len).unwrap(), + desc, + key, + ) + .unwrap() + .unwrap(); + assert_eq!(hit.value(), value); + assert_eq!(hit.seqno(), 3); + engine.shutdown().unwrap(); } - let hash = hash_key(data.hash_seed, key); - assert!(regions.begin_point_read(hash).is_some()); - assert!(regions.begin_point_read(hash).is_some()); - let previous = regions.lookup_snapshot(hash).unwrap(); - if let Some(corrupt_byte) = corrupt_byte { - bytes[cold_len + RECORD_HEADER_SIZE + corrupt_byte] ^= 1; - } - let stats = regions - .scan_reclaim(receipt, &bytes, |record| { - assert!(reinserted, "invalid or over-budget payload reached staging"); - assert!(matches!( - regions.try_stage_reinsert(&staging, 0, record)?, - RegionStageValue::Staged { seqno, .. } if seqno > 3 - )); - Ok(true) - }) - .unwrap(); - assert_eq!(stats.bytes_read, receipt.used_offset); - assert_eq!(stats.records_scanned, 2); - assert_eq!(stats.records_removed, 2 - u64::from(reinserted)); - assert_eq!(stats.reinsert_records, u64::from(reinserted)); - assert_eq!( - stats.reinsert_bytes, - u64::from(reinserted) * u64::from(rewrite_bytes) - ); - assert_eq!(stats.reinsert_skipped, u64::from(!reinserted)); - assert_eq!(stats.reinsert_budget_skipped, budget_skips); - if !reinserted { - assert_eq!(regions.lookup_snapshot(hash).unwrap(), None); - assert!(staging.shard_fill_snapshot(0).unwrap().is_none()); - continue; - } - assert_eq!(regions.lookup_snapshot(hash).unwrap(), previous); - assert!( - used - cold_len > rewrite_bytes as usize, - "source has flush padding" - ); - // Destroy the source before I/O: staging must own the validated bytes. - bytes.fill(0xcc); - drop(bytes); - let directory = TestDirectory::new(); - let (file, _) = FaultFile::open(&directory.paths.data).unwrap(); - file.set_len(data.geometry.data_file_len).unwrap(); - let engine = IoEngine::for_test(Arc::new(file), 1).unwrap(); - regions - .flush_staging_shard( - &staging, - &engine, - 0, - crate::io::background::BackgroundIoAttempt::new( - &crate::io::background::IoRecovery::new(Some(Duration::ZERO)), - None, - ), - ) - .unwrap() - .unwrap(); - let current = regions.begin_point_read(hash).unwrap(); - assert_ne!(current.region_generation, receipt.created_seqno); - let desc = describe_read(data.geometry, hash, current, false).unwrap(); - let memory = data_path_memory(); - let hit = regions - .read_value_from_desc( - &engine, - engine.try_reserve_read().unwrap(), - memory.try_read_buffer(desc.read_len).unwrap(), - desc, - key, - ) - .unwrap() - .unwrap(); - assert_eq!(hit.value(), value); - assert_eq!(hit.seqno(), 3); - engine.shutdown().unwrap(); } }