diff --git a/Cargo.lock b/Cargo.lock index dd78db15..1f7181d2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -41,6 +41,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "0.6.21" @@ -923,6 +929,12 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cbindgen" version = "0.29.2" @@ -990,6 +1002,33 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clap" version = "4.6.1" @@ -1169,6 +1208,37 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3bf7af66b0989381bd0be551bd7cc91912a655a58c6918420c9527b1fd8b4679" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "itertools 0.13.0", + "num-traits", + "oorandom", + "regex", + "serde", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + [[package]] name = "crossbeam" version = "0.8.4" @@ -1225,6 +1295,12 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-common" version = "0.1.7" @@ -1882,6 +1958,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + [[package]] name = "hashbrown" version = "0.14.5" @@ -2288,6 +2375,24 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.14.0" @@ -2860,6 +2965,7 @@ dependencies = [ "bitflags", "blake3", "bytes", + "criterion", "crossbeam", "dashmap", "fastcdc", @@ -3244,6 +3350,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "open" version = "5.3.5" @@ -3563,7 +3675,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -3584,7 +3696,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn", @@ -4052,6 +4164,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "scc" version = "2.4.0" @@ -4646,6 +4767,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tinyvec" version = "1.11.0" @@ -5275,6 +5406,16 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index d01709bd..c464ee89 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,6 +57,7 @@ clap = { version = "4.5.47", features = ["derive", "string", "unicode"] } clap-markdown = "0.1.5" config = { version = "0.15.15", features = ["toml"], default-features = false } ctor = "0.2" +criterion = { version = "0.6", default-features = false } crossbeam = "0.8.4" dashmap = "6.1.0" diffy = "0.4.2" diff --git a/lore-storage/Cargo.toml b/lore-storage/Cargo.toml index 43346398..2124e94d 100644 --- a/lore-storage/Cargo.toml +++ b/lore-storage/Cargo.toml @@ -38,6 +38,13 @@ uuid = { workspace = true, features = ["zerocopy"] } xxhash-rust = { workspace = true } zerocopy = { workspace = true } +[dev-dependencies] +criterion = { workspace = true } + +[[bench]] +name = "chunking" +harness = false + [features] default = [] failure_generator = [] diff --git a/lore-storage/benches/chunking.rs b/lore-storage/benches/chunking.rs new file mode 100644 index 00000000..9683bdb4 --- /dev/null +++ b/lore-storage/benches/chunking.rs @@ -0,0 +1,123 @@ +use std::sync::Arc; + +use bytes::Bytes; +use criterion::{criterion_group, criterion_main, Criterion}; +use rand::Rng; + +use lore_storage::fragment_engine::chunk_boundaries; +use lore_storage::FRAGMENT_SIZE_EXPECTED; +use lore_storage::FRAGMENT_SIZE_MINIMUM; +use lore_storage::FRAGMENT_SIZE_THRESHOLD; + +fn make_random_buffer(size: usize) -> Bytes { + let mut data = vec![0u8; size]; + rand::rng().fill(&mut data[..]); + Bytes::from(data) +} + +/// Simulates the old per-chunk approach: one compute-pool spawn + oneshot per boundary. +fn old_per_chunk_boundaries(buffer: &Bytes) -> Vec<(usize, usize)> { + let size = buffer.len(); + let buffer_guard = buffer.clone(); + + let chunker = { + let slice: &[u8] = unsafe { &*(buffer.as_ref() as *const [u8]) }; + Arc::new(fastcdc::v2020::FastCDC::with_level( + slice, + FRAGMENT_SIZE_MINIMUM as u32, + FRAGMENT_SIZE_EXPECTED as u32, + FRAGMENT_SIZE_THRESHOLD as u32, + fastcdc::v2020::Normalization::Level1, + )) + }; + + let mut offset = 0; + let mut boundaries = Vec::new(); + + while offset < size { + let remain = size - offset; + let chunker = chunker.clone(); + let guard = buffer_guard.clone(); + let (tx, rx) = tokio::sync::oneshot::channel(); + lore_base::runtime::compute_pool().spawn(move || { + let (_, end) = chunker.cut(offset, remain); + drop(guard); + let _ = tx.send(end); + }); + let end = rx.blocking_recv().expect("chunker task failed"); + boundaries.push((offset, end)); + offset = end; + } + + boundaries +} + +fn bench_chunking(c: &mut Criterion) { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + + let sizes: &[usize] = &[ + 64 * 1024, + 1024 * 1024, + 16 * 1024 * 1024, + 64 * 1024 * 1024, + ]; + + let buffers: Vec<(String, Bytes)> = sizes + .iter() + .map(|&size| { + let label = if size >= 1024 * 1024 { + format!("{}MiB", size / (1024 * 1024)) + } else { + format!("{}KiB", size / 1024) + }; + (label, make_random_buffer(size)) + }) + .collect(); + + let mut group = c.benchmark_group("fastcdc_chunking"); + + for (label, buffer) in &buffers { + group.bench_function(format!("batched_{label}"), |b| { + b.iter(|| { + rt.block_on(async { + chunk_boundaries( + std::hint::black_box(buffer.clone()), + std::hint::black_box(0), + ) + .await + .expect("chunking succeeds"); + }) + }) + }); + + group.bench_function(format!("per_chunk_{label}"), |b| { + b.iter(|| { + std::hint::black_box(old_per_chunk_boundaries(std::hint::black_box(buffer))); + }) + }); + } + + group.finish(); + + { + let buffer_16mb = make_random_buffer(16 * 1024 * 1024); + c.bench_function("fixed_size/16MiB", |b| { + b.iter(|| { + rt.block_on(async { + chunk_boundaries( + std::hint::black_box(buffer_16mb.clone()), + std::hint::black_box(FRAGMENT_SIZE_EXPECTED), + ) + .await + .expect("chunking succeeds"); + }) + }) + }); + } +} + +criterion_group!(benches, bench_chunking); +criterion_main!(benches); diff --git a/lore-storage/src/fragment_engine.rs b/lore-storage/src/fragment_engine.rs index 07b3c434..c13fb98a 100644 --- a/lore-storage/src/fragment_engine.rs +++ b/lore-storage/src/fragment_engine.rs @@ -24,12 +24,62 @@ use crate::types::FragmentReference; use crate::types::Partition; use crate::write::store_fragment; -/// Splits content into fragments using `FastCDC` and stores them via -/// [`store_fragment`]. +/// Figure out where to cut `buffer` into chunks, all in one go. +pub async fn chunk_boundaries( + buffer: Bytes, + fixed_size_chunk: usize, +) -> Result, StorageError> { + let size = buffer.len(); + if fixed_size_chunk > 0 { + let step = fixed_size_chunk; + Ok((0..size) + .step_by(step) + .map(|offset| (offset, (offset + step).min(size))) + .collect()) + } else { + let chunker = { + // SAFETY: The chunker borrows `buffer` via a forged `'static` slice (see + // `extend_lifetime`). The compute-pool task is detached and is not + // cancelled if this future is dropped at the `rx.await` below, so + // we must keep the buffer allocation alive for the whole task. + // `Bytes::clone` bumps the refcount with a stable data pointer, + // guaranteeing the slice stays valid until the task finishes. + let slice: &[u8] = unsafe { extend_lifetime(buffer.as_ref()) }; + fastcdc::v2020::FastCDC::with_level( + slice, + FRAGMENT_SIZE_MINIMUM as u32, + FRAGMENT_SIZE_EXPECTED as u32, + FRAGMENT_SIZE_THRESHOLD as u32, + fastcdc::v2020::Normalization::Level1, + ) + }; + let buffer_guard = buffer.clone(); + let (tx, rx) = tokio::sync::oneshot::channel(); + lore_base::runtime::compute_pool().spawn(move || { + let _ = tx.send( + chunker + .map(|c| (c.offset, c.offset + c.length)) + .collect::>(), + ); + drop(buffer_guard); + }); + let boundaries = rx + .await + .map_err(|e| StorageError::internal_with_context(e, "chunker task failed"))?; + Ok(boundaries) + } +} + +/// Cuts `buffer` into chunks and stores each one via [`store_fragment`]. +/// +/// Uses `FastCDC` (content-defined chunking) when `flags.fixed_size_chunk` is 0, +/// or fixed-size chunking when it is >0. Each chunk is hashed, stored as a +/// content-addressed fragment in the immutable `store`, and assembled into a +/// fragment list (Merklized). Returns the root address and fragment. /// -/// For each chunk the function calls [`store_fragment`] directly with the -/// provided store, partition, and optional remote session. This replaces the -/// previous closure-based `store_fn` pattern. +/// When the entire buffer fits in a single chunk, the single-fragment fast path +/// is used and no fragment list is created. In `hash_only` mode, fragments are +/// not actually stored — only their hashes are computed. #[allow(clippy::too_many_arguments)] pub async fn write_fragmented( store: Arc, @@ -41,56 +91,18 @@ pub async fn write_fragmented( remote_session: Option>, tracker: Option>, ) -> Result<(Address, Fragment), StorageError> { - // Raw data, use content defined chunking with FastCDC let size = buffer.len(); - let mut chunk_offset = 0usize; let mut tasks = JoinSet::>::new(); - let mut chunk_index = 0usize; - - let chunker = Arc::new({ - // SAFETY: we await on all spawned tasks using this chunker before buffer - // is dropped, so the lifetime will not escape the scope of the buffer. - let buffer: &[u8] = unsafe { extend_lifetime(buffer.as_ref()) }; - fastcdc::v2020::FastCDC::with_level( - buffer, - FRAGMENT_SIZE_MINIMUM as u32, - FRAGMENT_SIZE_EXPECTED as u32, - FRAGMENT_SIZE_THRESHOLD as u32, - fastcdc::v2020::Normalization::Level1, - ) - }); lore_base::lore_trace!( "Write and fragment buffer to immutable store: {size} bytes representing {size} bytes (flags {flags:?})", ); - while chunk_offset < size { - let chunk_remain = size - chunk_offset; - let chunk_end = if flags.fixed_size_chunk > 0 { - chunk_offset + std::cmp::min(flags.fixed_size_chunk, chunk_remain) - } else { - // FastCDC chunking is CPU-bound (rolling hash over the buffer). - // Run it on the shared compute pool alongside compression so it - // doesn't contend with blocking IO on tokio's blocking pool. - let chunker = chunker.clone(); - // The chunker borrows `buffer` via a forged `'static` slice (see - // `extend_lifetime` above). The rayon task is detached and is not - // cancelled if this future is dropped at the `rx.await` below, so - // we must keep the buffer allocation alive for the whole task. - // `Bytes::clone` bumps the refcount with a stable data pointer, - // guaranteeing the slice stays valid until the task finishes. - let buffer_guard = buffer.clone(); - let (tx, rx) = tokio::sync::oneshot::channel(); - lore_base::runtime::compute_pool().spawn(move || { - let _ = tx.send(chunker.cut(chunk_offset, chunk_remain)); - drop(buffer_guard); - }); - let (_, chunk_end) = rx - .await - .map_err(|e| StorageError::internal_with_context(e, "chunker task failed"))?; - chunk_end - }; + + let chunk_boundaries = chunk_boundaries(buffer.clone(), flags.fixed_size_chunk).await?; + + for (chunk_index, (chunk_offset, chunk_end)) in chunk_boundaries.into_iter().enumerate() { let chunk_size = chunk_end - chunk_offset; - let chunk_buffer = buffer.slice(chunk_offset..(chunk_offset + chunk_size)); + let chunk_buffer = buffer.slice(chunk_offset..chunk_end); let fragment = Fragment { flags: flags.into(), @@ -154,12 +166,7 @@ pub async fn write_fragmented( }; Ok((chunk_index, chunk_offset, chunk_address)) }); - - chunk_offset += chunk_size; - chunk_index += 1; } - drop(chunker); - drop(buffer); let mut list_buffer = BytesMut::with_capacity(tasks.len() * std::mem::size_of::()); @@ -407,3 +414,78 @@ where { unsafe { &*(data as *const T) } } + +#[cfg(test)] +mod tests { + use bytes::Bytes; + + use super::*; + + /// Make a buffer of random bytes so chunking has to actually cut it. + fn mixed_pattern_buffer(size: usize) -> Bytes { + use rand::Rng; + let mut data = vec![0u8; size]; + rand::rng().fill(&mut data[..]); + Bytes::from(data) + } + + #[tokio::test] + async fn fastcdc_batch_handles_buffer_smaller_than_min_chunk() { + // Tiny buffer — should be one chunk covering the whole thing. + let buffer = mixed_pattern_buffer(1024); + let actual = chunk_boundaries(buffer.clone(), 0) + .await + .expect("chunking succeeds"); + assert_eq!(actual, vec![(0, buffer.len())]); + } + + #[tokio::test] + async fn fastcdc_batch_handles_empty_buffer() { + let buffer = Bytes::new(); + let actual = chunk_boundaries(buffer, 0) + .await + .expect("chunking succeeds"); + assert!(actual.is_empty()); + } + + #[tokio::test] + async fn fastcdc_batch_handles_pathological_all_zero_buffer() { + // All-zero input — the rolling hash never matches, but the split should still be clean. + let buffer = Bytes::from(vec![0u8; 512 * 1024]); + let actual = chunk_boundaries(buffer.clone(), 0) + .await + .expect("chunking succeeds"); + let reconstructed_size: usize = actual.iter().map(|(s, e)| e - s).sum(); + assert_eq!(reconstructed_size, buffer.len()); + assert!(actual.iter().all(|&(s, e)| s < e)); + assert_eq!(actual.first().copied(), Some((0, actual[0].1))); + assert_eq!(actual.last().unwrap().1, buffer.len()); + } + + #[tokio::test] + async fn fixed_size_chunking_covers_whole_buffer() { + let buffer = mixed_pattern_buffer(10 * 1024 + 17); + let step = 1024; + let actual = chunk_boundaries(buffer.clone(), step) + .await + .expect("chunking succeeds"); + + let expected: Vec<(usize, usize)> = (0..buffer.len()) + .step_by(step) + .map(|o| (o, (o + step).min(buffer.len()))) + .collect(); + assert_eq!(actual, expected); + + let reconstructed_size: usize = actual.iter().map(|(s, e)| e - s).sum(); + assert_eq!(reconstructed_size, buffer.len()); + } + + #[tokio::test] + async fn fixed_size_chunking_handles_buffer_smaller_than_step() { + let buffer = Bytes::from(vec![1u8; 100]); + let actual = chunk_boundaries(buffer.clone(), 1024) + .await + .expect("chunking succeeds"); + assert_eq!(actual, vec![(0, buffer.len())]); + } +}