From ba5090e73b636dc87a2dd7e3e5e77baa905aca03 Mon Sep 17 00:00:00 2001 From: Valter Balegas Date: Mon, 13 Jul 2026 21:25:11 +0100 Subject: [PATCH] =?UTF-8?q?perf(durable-streams-rust):=20--stream-lanes=20?= =?UTF-8?q?=E2=80=94=20hash=20stream=20data=20files=20across=20per-device?= =?UTF-8?q?=20lane=20dirs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ~1M-stream wall (wal-1m-diag): with every stream file on ONE data device, a checkpoint's syncfs writeback of ~200k dirty files took 60-74s per shard — tiny appends to distinct files amplify to ~8-12KB of data+inode+journal writeback each, saturating the single device and starving append staging (batch_avg collapsed 53->2-6). --stream-lanes N hashes stream files across streams/<0..N>/ subdirs, one per device: N× writeback capacity, N parallel syncfs barriers (one per lane, threaded), and no single ext4 dir holds every stream. Default 1 = byte-identical historical layout. Recovery walks all lanes; e2e covers lanes=3 crash recovery + layout spread. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y3x7bcT9vLGiT4tXZeQpnk --- packages/durable-streams-rust/src/main.rs | 14 +++ packages/durable-streams-rust/src/store.rs | 106 +++++++++++++++++- .../durable-streams-rust/src/wal/e2e_tests.rs | 62 ++++++++++ .../durable-streams-rust/src/wal/shard.rs | 5 +- 4 files changed, 183 insertions(+), 4 deletions(-) diff --git a/packages/durable-streams-rust/src/main.rs b/packages/durable-streams-rust/src/main.rs index 19b341ffe3..d874aef155 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -311,6 +311,20 @@ fn main() { } } } + // Stream data lanes: hash stream files across streams/<0..N>/ subdirs, + // one per (intended) device, so checkpoint writeback spreads over N + // devices with N parallel syncfs barriers (the ~1M-stream wall fix). + // A LAYOUT choice: must match the on-disk layout across restarts. + "--stream-lanes" => { + let v = val(args.next(), "--stream-lanes"); + match v.parse::() { + Ok(n) if n >= 1 => store::set_stream_lanes(n), + _ => { + eprintln!("--stream-lanes must be a positive integer"); + std::process::exit(2); + } + } + } // Checkpoint size trigger: checkpoint a shard as soon as its retained // WAL exceeds this many bytes (0 = disabled). An explicit replay-time // budget that also self-staggers shards by their own write rates. diff --git a/packages/durable-streams-rust/src/store.rs b/packages/durable-streams-rust/src/store.rs index 6bc2fc927e..fdb0b749e6 100644 --- a/packages/durable-streams-rust/src/store.rs +++ b/packages/durable-streams-rust/src/store.rs @@ -186,6 +186,89 @@ pub(crate) fn syncfs_barrier(_file: &File) -> std::io::Result<()> { )) } +/// Stream-lane count (`--stream-lanes`, default 1 = the flat `streams/` layout). +/// With N > 1, stream data files are hashed across `streams/<0..N>/` subdirs so +/// each lane can be mounted on its OWN device: the checkpoint's dirty-file +/// writeback (the ~1M-stream wall — one `syncfs` measured at 60–74 s when every +/// stream shared one device, wal-1m-diag 2026-07-13) spreads over N devices and +/// runs N barriers in parallel, and no single ext4 directory holds every stream. +/// Must be set BEFORE `Store::open` and match the on-disk layout across restarts +/// (same N or files won't be found — a layout choice, not a runtime tunable). +static STREAM_LANES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(1); +pub fn set_stream_lanes(n: usize) { + STREAM_LANES.store(n.max(1), Ordering::Relaxed); +} +pub fn stream_lanes() -> usize { + STREAM_LANES.load(Ordering::Relaxed) +} + +/// Stable lane for a stream data-file name (FNV-1a; the fname embeds the stream +/// id, so this is fixed for the stream's lifetime and recomputable anywhere). +fn lane_of(fname: &str) -> usize { + let lanes = stream_lanes(); + if lanes <= 1 { + return 0; + } + let mut h: u64 = 0xcbf2_9ce4_8422_2325; + for b in fname.as_bytes() { + h ^= *b as u64; + h = h.wrapping_mul(0x1000_0000_01b3); + } + (h % lanes as u64) as usize +} + +/// Directory of lane `lane`: the flat `streams/` when lanes == 1 (byte-identical +/// to the historical layout), else `streams//`. +fn lane_dir(data_dir: &std::path::Path, lane: usize) -> PathBuf { + let root = data_dir.join("streams"); + if stream_lanes() <= 1 { + root + } else { + root.join(lane.to_string()) + } +} + +/// Open directory fds, one per stream lane, registered at `Store::open` — the +/// checkpoint's syncfs must barrier EVERY lane's filesystem (touched files can +/// live on any lane), and a dir fd is a valid syncfs target. Replaced (not +/// appended) per open so tests that build many stores target the latest layout. +static LANE_SYNC_FDS: StdMutex>>> = StdMutex::new(None); + +/// Checkpoint durability barrier across all stream lanes: one `syncfs` per lane, +/// parallelized (each is a full device writeback and the lanes are independent +/// devices in the intended deployment). Falls back to a single barrier on +/// `fallback`'s fs when no lane registry exists (e.g. shard-only unit tests). +pub(crate) fn syncfs_stream_lanes(fallback: &File) -> std::io::Result<()> { + let fds = LANE_SYNC_FDS.lock().unwrap().clone(); + match fds { + Some(fds) if !fds.is_empty() => { + if fds.len() == 1 { + return syncfs_barrier(&fds[0]); + } + std::thread::scope(|s| { + let handles: Vec<_> = fds + .iter() + .map(|f| s.spawn(move || syncfs_barrier(f))) + .collect(); + let mut first_err = None; + for h in handles { + if let Err(e) = h + .join() + .unwrap_or_else(|_| Err(std::io::Error::other("syncfs thread panicked"))) + { + first_err.get_or_insert(e); + } + } + match first_err { + None => Ok(()), + Some(e) => Err(e), + } + }) + } + _ => syncfs_barrier(fallback), + } +} + pub struct StreamState { pub id: u64, pub path: String, @@ -472,6 +555,17 @@ impl Store { ) -> std::io::Result { let streams_dir = data_dir.join("streams"); std::fs::create_dir_all(&streams_dir)?; + // Create every stream-lane dir and register their dir fds for the + // checkpoint's per-lane syncfs barrier (see `syncfs_stream_lanes`). + { + let mut lane_fds = Vec::with_capacity(stream_lanes()); + for lane in 0..stream_lanes() { + let d = lane_dir(&data_dir, lane); + std::fs::create_dir_all(&d)?; + lane_fds.push(File::open(&d)?); + } + *LANE_SYNC_FDS.lock().unwrap() = Some(Arc::new(lane_fds)); + } // Stream data can be sensitive; keep the data dir owner-only (best-effort). #[cfg(unix)] { @@ -511,10 +605,16 @@ impl Store { /// everything else. Orphan files (crash between create and meta write) are /// discarded. fn recover(&self, streams_dir: &std::path::Path) -> std::io::Result<()> { + let _ = streams_dir; // root; per-lane dirs derived below (lane 0 == root when lanes == 1) let mut metas: HashMap = HashMap::new(); let mut data_files: Vec = Vec::new(); - for entry in std::fs::read_dir(streams_dir)? { - let p = entry?.path(); + let mut entries: Vec = Vec::new(); + for lane in 0..stream_lanes() { + for entry in std::fs::read_dir(lane_dir(&self.data_dir, lane))? { + entries.push(entry?.path()); + } + } + for p in entries { let name = p.file_name().and_then(|n| n.to_str()).unwrap_or(""); if name.ends_with(".meta.tmp") { let _ = std::fs::remove_file(&p); @@ -859,7 +959,7 @@ impl Store { } let id = self.next_id.fetch_add(1, Ordering::Relaxed); let fname = format!("{}~{}", encode_path(path), id); - let file_path = self.data_dir.join("streams").join(fname); + let file_path = lane_dir(&self.data_dir, lane_of(&fname)).join(fname); let file = Arc::new( OpenOptions::new() .create(true) diff --git a/packages/durable-streams-rust/src/wal/e2e_tests.rs b/packages/durable-streams-rust/src/wal/e2e_tests.rs index 135313459f..80606ff1d6 100644 --- a/packages/durable-streams-rust/src/wal/e2e_tests.rs +++ b/packages/durable-streams-rust/src/wal/e2e_tests.rs @@ -952,6 +952,68 @@ async fn e2e_recycled_first_segment_acked_records_survive_crash() { let _ = std::fs::remove_dir_all(&dir); } +/// `--stream-lanes N`: stream data files hash across `streams/<0..N>/` subdirs +/// (one per device in the intended deployment — the ~1M-stream writeback-wall +/// fix). Crash recovery must find every file in its lane dir, and the +/// checkpoint's per-lane syncfs must preserve durability-before-recycle exactly +/// as the single-lane layout does. Guarded by DurabilityGuard (serialized) since +/// stream-lanes is process-global state; reset to 1 before releasing the guard. +#[tokio::test] +async fn e2e_stream_lanes_recover_acked_records() { + let _guard = DurabilityGuard::wal(); + crate::store::set_stream_lanes(3); + crate::wal::shard::set_checkpoint_syncfs(true); + const SEG: u64 = 4096; + let dir = tmp("stream-lanes"); + + let h = Harness::boot_with_segment_size(&dir, Some(1), 1, SEG).unwrap(); + // Enough streams that the FNV lane hash populates more than one lane. + let names: Vec = (0..12).map(|i| format!("lane-s{i}")).collect(); + for n in &names { + create_stream(&h.store, n, OCTET).await; + } + let mut expected: std::collections::HashMap> = Default::default(); + for round in 0..40usize { + for n in &names { + let rec = format!("{n}-r{round:03}|").into_bytes(); + append_acked(&h.store, n, OCTET, &rec).await; + expected.entry(n.clone()).or_default().extend_from_slice(&rec); + } + } + // Checkpoint (per-lane syncfs + recycle), then more acked appends on top. + h.walset.shards()[0].checkpoint().await.unwrap(); + for n in &names { + let rec = format!("{n}-post|").into_bytes(); + append_acked(&h.store, n, OCTET, &rec).await; + expected.entry(n.clone()).or_default().extend_from_slice(&rec); + } + + h.crash(); + + let h2 = Harness::boot_with_segment_size(&dir, None, 1, SEG).unwrap(); + // Layout sanity: files actually spread across lane subdirs. + let lanes_used = (0..3) + .filter(|l| { + std::fs::read_dir(dir.join("streams").join(l.to_string())) + .map(|d| d.flatten().next().is_some()) + .unwrap_or(false) + }) + .count(); + assert!(lanes_used >= 2, "expected streams spread over lanes, got {lanes_used}"); + for n in &names { + let got = stream_file_bytes(&h2.store, n); + assert_eq!( + &got, + expected.get(n).unwrap(), + "stream {n} recovers byte-identical across lanes" + ); + } + h2.crash(); + crate::wal::shard::set_checkpoint_syncfs(false); + crate::store::set_stream_lanes(1); + let _ = std::fs::remove_dir_all(&dir); +} + /// Cardinality-cliff #1: with `--wal-checkpoint-syncfs on`, the checkpoint makes /// touched per-stream files durable via ONE `syncfs()` barrier instead of the /// per-stream `fdatasync` loop. This must preserve the durability-before-recycle diff --git a/packages/durable-streams-rust/src/wal/shard.rs b/packages/durable-streams-rust/src/wal/shard.rs index 4d94591b8e..909b1f03b8 100644 --- a/packages/durable-streams-rust/src/wal/shard.rs +++ b/packages/durable-streams-rust/src/wal/shard.rs @@ -919,7 +919,10 @@ impl Shard { // dirty page on it — all touched files become durable at once. Still // strictly before persist_durable_tails/recycle, so the // durability-before-recycle ordering is unchanged. - crate::store::syncfs_barrier(&touched[0].2)?; + // One syncfs per stream lane (touched files may span lanes on + // multi-device layouts); falls back to touched[0]'s fs when no + // lane registry exists (shard-only unit tests). + crate::store::syncfs_stream_lanes(&touched[0].2)?; } else if fanout <= 1 { for (_, _, f) in &touched { crate::store::barrier_fsync(f)?;