diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index b3fb746..17eb190 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -468,6 +468,8 @@ Three invariants bind every implementation: The two LSM backends are interchangeable in contract and share the value-record codec (`snapshot_record.rs`); fjall keeps the crate pure-Rust, RocksDB trades a C++ build dependency for the battle-tested engine and its operational tooling (`ldb`, `sst_dump`). Both keep the cursor in the same atomic batch as the data it names, so under NO_SYNC a crash can lose the un-synced tail but never desynchronize cursor from data — on reopen the recovered cursor is consistent and the watch re-folds the tail. The rest of this section describes the **append-log backend** (the default), whose on-disk format is below. +**fjall write-halt watchdog.** fjall halts a writer while L0 holds 30 runs, but enqueues compaction only from flushes, and a halted writer makes no flushes: if the queued compactions finish without getting L0 under 30, `apply` parks forever (measured on fjall 3.1.4 at 190M entries and 3.1.5 at 282M entries with 4 workers and 64 MiB memtables, L0 at 33 to 35 tables, no compaction running; `check_write_halt` is unchanged through 3.1.13). `FjallSnapshot` runs a watchdog thread that samples once a second and, after 5 consecutive samples with L0 at 30 or more tables, no running compaction, no queued flush and a sequence number that has not moved (a halted writer commits nothing; L0 *tables* run far past 30 on a healthy fold, so the table count alone would fire needless full rewrites), forces `major_compact` on the data partition, which needs no queue message. The thread stops and is joined when the snapshot drops. `FjallStats` (`FjallSnapshot::stats`, or `FjallReader::stats` from a scraper thread, since `apply` holds `&mut self` while parked) exposes `l0_tables`, `outstanding_flushes`, `active_compactions` and `halt_rescues`; a non-zero `halt_rescues` means the wedge was hit and recovered. `FjallConfig::worker_threads` and `memtable_bytes` override the tuned defaults (`examples/halt_bench.rs` is the bulk-apply harness). + ### File Format ``` diff --git a/examples/halt_bench.rs b/examples/halt_bench.rs new file mode 100644 index 0000000..d485691 --- /dev/null +++ b/examples/halt_bench.rs @@ -0,0 +1,127 @@ +//! Bulk-apply bench for fjall's write-halt wedge (J11). +//! +//! Applies `ENTRIES` puts through `FjallSnapshot::apply` in 1024-entry batches +//! as fast as possible, printing one line per `REPORT` entries: entries, +//! interval writes/s, L0 tables, active compactions, watchdog rescues. Exits 2 +//! with `WEDGED` when the interval rate stays under `WEDGE_RATE` writes/s for +//! `WEDGE_SECS` seconds. +//! +//! Env: ENTRIES, KEYS=rand|seq, DIR, REPORT, WEDGE_RATE, WEDGE_SECS, +//! WORKERS and MEMTABLE_MB (0 = shipped tuning; 4 and 64 is the config that +//! wedged raw fjall 3.1.5 at 257M entries). +//! +//! cargo run --release --example halt_bench --features fjall + +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::time::{Duration, Instant}; + +use slipstream::{ + FjallConfig, FjallSnapshot, KvEntry, KvUpdate, SnapshotStore, VersionToken, WatchCursor, +}; + +const BATCH: usize = 1024; + +fn env(k: &str, d: T) -> T { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} + +fn mix(mut x: u64) -> u64 { + x ^= x >> 33; + x = x.wrapping_mul(0xff51_afd7_ed55_8ccd); + x ^= x >> 33; + x = x.wrapping_mul(0xc4ce_b9fe_1a85_ec53); + x ^ (x >> 33) +} + +fn key(i: u64, rand: bool) -> String { + let k = if rand { mix(i) } else { i }; + format!("route.svc-{:08}.{:08}", k / 1000, k % 1000) +} + +fn main() { + let entries: u64 = env("ENTRIES", 300_000_000); + let rand = env::("KEYS", "rand".into()) == "rand"; + let report: u64 = env("REPORT", 5_000_000); + let wedge_rate: f64 = env("WEDGE_RATE", 2000.0); + let wedge_secs: u64 = env("WEDGE_SECS", 120); + let dir: PathBuf = env::("DIR", "/tmp/halt_bench".into()).into(); + let _ = std::fs::remove_dir_all(&dir); + let mut value = vec![0u8; 200]; + let mut s = 0x5EEDu64; + for b in &mut value { + s = mix(s); + *b = s as u8; + } + + let config = FjallConfig { + worker_threads: env("WORKERS", 0), + memtable_bytes: env::("MEMTABLE_MB", 0) << 20, + ..FjallConfig::default() + }; + let (_, mut snap) = FjallSnapshot::open(&dir, config).unwrap(); + let rd = snap.reader(); + let applied = Arc::new(AtomicU64::new(0)); + + let a2 = applied.clone(); + let t0 = Instant::now(); + std::thread::spawn(move || { + let (mut last_n, mut last_t, mut next) = (0u64, Instant::now(), report); + let mut slow_since = None::; + loop { + std::thread::sleep(Duration::from_secs(1)); + let n = a2.load(Relaxed); + let now = Instant::now(); + let rate = (n - last_n) as f64 / now.duration_since(last_t).as_secs_f64(); + let st = rd.stats(); + let line = format!( + "entries={n} writes/s={rate:.0} l0={} active_compactions={} halt_rescues={} t={}s", + st.l0_tables, + st.active_compactions, + st.halt_rescues, + t0.elapsed().as_secs() + ); + if n >= next { + println!("{line}"); + next += report; + } + if n >= entries { + println!("DONE {line} wedged=no"); + std::process::exit(0); + } + if rate < wedge_rate { + let since = *slow_since.get_or_insert(now); + if now.duration_since(since).as_secs() >= wedge_secs { + println!("WEDGED {line} wedged=yes"); + std::process::exit(2); + } + } else { + slow_since = None; + } + last_n = n; + last_t = now; + } + }); + + let mut batch = Vec::with_capacity(BATCH); + let mut i = 0u64; + while i < entries { + batch.clear(); + let end = (i + BATCH as u64).min(entries); + for j in i..end { + batch.push(KvUpdate::Put(KvEntry { + key: key(j, rand), + value: value.clone(), + version: VersionToken::from_u64(j + 1), + })); + } + snap.apply(&batch, &WatchCursor::from_u64(end)).unwrap(); + applied.store(end, Relaxed); + i = end; + } + std::thread::sleep(Duration::from_secs(2)); +} diff --git a/src/lib.rs b/src/lib.rs index 0a69642..acf1404 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,7 +55,7 @@ pub use nats::{NatsConnection, NatsConnectionConfig, nats_connect}; pub use repair::{ExpiryRepair, RestoreSource, RestoredFold}; pub use snapshot::{AppendLogSnapshot, SnapshotStore}; #[cfg(feature = "fjall")] -pub use snapshot_fjall::{FjallConfig, FjallReader, FjallSnapshot}; +pub use snapshot_fjall::{FjallConfig, FjallReader, FjallSnapshot, FjallStats}; #[cfg(feature = "rocksdb")] pub use snapshot_rocksdb::{RocksDbConfig, RocksDbReader, RocksDbSnapshot}; pub use stores::{ diff --git a/src/snapshot_fjall.rs b/src/snapshot_fjall.rs index d4d307a..0ef5d1d 100644 --- a/src/snapshot_fjall.rs +++ b/src/snapshot_fjall.rs @@ -58,6 +58,9 @@ //! cold-read folds favor RocksDB. use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc; use std::time::{Duration, Instant}; use fjall::config::{BlockSizePolicy, PinningPolicy}; @@ -99,6 +102,18 @@ const META_MEMTABLE_BYTES: u64 = 8 << 20; /// instead of 4 KiB on a cache-miss point read. const DATA_BLOCK_SIZE: u32 = 16 * 1024; +/// L0 table count at which fjall's write halt can engage (it counts L0 *runs*, +/// which are never more than tables, so tables is the conservative trigger). +const L0_HALT_TABLES: usize = 30; + +/// How often the watchdog samples the data partition. +const WATCHDOG_INTERVAL: Duration = Duration::from_secs(1); + +/// Consecutive samples of "no commits, L0 at the halt line, compactor idle" before the +/// watchdog acts. Compaction picks come in bursts; 5 s idle behind 30 L0 tables +/// is the wedge, not a gap between picks. +const WATCHDOG_STRIKES: u32 = 5; + /// Durability and read-cache configuration for [`FjallSnapshot`]. /// /// Defaults to NO_SYNC (`sync: false`) — same cache philosophy as the append @@ -119,6 +134,14 @@ pub struct FjallConfig { /// the cache to the hot set so hydrations stay cache-resident. `0` falls back to /// fjall's 32 MiB default. Set this to roughly the resident working-set size. pub cache_size_bytes: u64, + + /// fjall flush/compaction worker threads. `0` (the default) uses + /// `min(cores, 16)`; on a small host that is `cores`, so hydration shares + /// few workers with serving. + pub worker_threads: usize, + + /// Data-partition memtable size in bytes. `0` (the default) uses 256 MiB. + pub memtable_bytes: u64, } impl Default for FjallConfig { @@ -128,10 +151,32 @@ impl Default for FjallConfig { // 1 GiB: holds index/data blocks for a ~1e6-service working set // resident, matching the routing registries' default resident cap. cache_size_bytes: 1024 * 1024 * 1024, + worker_threads: 0, + memtable_bytes: 0, } } } +/// Point-in-time LSM health of a [`FjallSnapshot`], for a consumer's gauges. +/// +/// `l0_tables >= 30` with `active_compactions == 0` is the shape of fjall's +/// write-halt wedge: the writer is parked in `apply` and nothing drains L0. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FjallStats { + /// L0 tables in the data partition. fjall throttles writes at 20 runs and halts at 30. + pub l0_tables: usize, + /// All tables in the data partition. + pub tables: usize, + /// Memtable flushes queued or running. + pub outstanding_flushes: usize, + /// Compactions running now. + pub active_compactions: usize, + /// Times the write-halt watchdog forced a major compaction to drain a wedged + /// L0. Non-zero means fjall's write-halt livelock was hit and recovered; a + /// rate of it means this node ingests faster than it compacts. + pub halt_rescues: u64, +} + /// On-disk durable fold backed by fjall. See the module docs in `src/snapshot_fjall.rs`. pub struct FjallSnapshot { // fjall 3 renamed its types: the database root is `Database` (was `Keyspace`) @@ -144,6 +189,57 @@ pub struct FjallSnapshot { // fjall's `Config.path` is `pub(crate)`, so the export path keeps its own // copy of the DB directory for the artifact copy. path: PathBuf, + rescues: Arc, + /// Dropping the sender stops the watchdog; `Drop` joins it. + watchdog: Option<(mpsc::Sender<()>, std::thread::JoinHandle<()>)>, +} + +impl Drop for FjallSnapshot { + fn drop(&mut self) { + if let Some((stop, handle)) = self.watchdog.take() { + drop(stop); + let _ = handle.join(); + } + } +} + +/// Un-wedge fjall's write halt. +/// +/// fjall halts a writer while L0 holds `>= 30` runs, but enqueues compaction only +/// from flushes (`try_send`, dropped when the queue is full) — and a halted +/// writer makes no flushes. If the queued compactions finish without bringing L0 +/// under 30 the writer is parked forever (measured: raw 4-worker fjall 3.1.5 at +/// 257M entries, L0 at 34 tables, no compaction running). A major compaction +/// needs no queue message and drains L0, so run one when the condition persists. +fn watchdog(db: Database, data: Keyspace, rescues: Arc, stop: mpsc::Receiver<()>) { + let mut strikes = 0u32; + let mut last_seqno = db.seqno(); + // `recv_timeout` returns `Disconnected` the moment the snapshot drops. + while let Err(mpsc::RecvTimeoutError::Timeout) = stop.recv_timeout(WATCHDOG_INTERVAL) { + // A halted writer commits nothing, so a moving sequence number means L0 + // is high (tables, unlike runs, can be far past 30 on a healthy fold) + // but writes are flowing: not the wedge. + let seqno = db.seqno(); + let stalled = seqno == last_seqno; + last_seqno = seqno; + let wedged = stalled + && data.l0_table_count() >= L0_HALT_TABLES + && db.active_compactions() == 0 + && db.outstanding_flushes() == 0; + strikes = if wedged { strikes + 1 } else { 0 }; + if strikes < WATCHDOG_STRIKES { + continue; + } + strikes = 0; + tracing::warn!( + l0_tables = data.l0_table_count(), + "fjall L0 at the write-halt line with an idle compactor; forcing a major compaction" + ); + rescues.fetch_add(1, Ordering::Relaxed); + if let Err(e) = data.major_compact() { + tracing::error!(error = %e, "fjall write-halt rescue compaction failed"); + } + } } impl FjallSnapshot { @@ -153,10 +249,13 @@ impl FjallSnapshot { /// persisted resume cursor — [`WatchCursor::none`] when fresh — and the store. pub fn open(path: &Path, config: FjallConfig) -> Result<(WatchCursor, Self), SnapshotError> { std::fs::create_dir_all(path)?; - let workers = std::thread::available_parallelism() - .map(std::num::NonZero::get) - .unwrap_or(4) - .min(MAX_WORKER_THREADS); + let workers = match config.worker_threads { + 0 => std::thread::available_parallelism() + .map(std::num::NonZero::get) + .unwrap_or(4) + .min(MAX_WORKER_THREADS), + n => n, + }; let mut builder = Database::builder(path).worker_threads(workers); // Size the LSM block cache to the working set (default 1 GiB). fjall's own // default is 32 MiB, far too small for the fold — see @@ -168,7 +267,10 @@ impl FjallSnapshot { let data = db .keyspace(DATA_PARTITION, || { KeyspaceCreateOptions::default() - .max_memtable_size(DATA_MEMTABLE_BYTES) + .max_memtable_size(match config.memtable_bytes { + 0 => DATA_MEMTABLE_BYTES, + n => n, + }) .data_block_size_policy(BlockSizePolicy::all(DATA_BLOCK_SIZE)) // Last-level filters are kept (no `expect_point_read_hits`): // they are the only in-memory rejection for absent-key @@ -198,6 +300,15 @@ impl FjallSnapshot { None => WatchCursor::none(), }; + let rescues = Arc::new(AtomicU64::new(0)); + let (stop_tx, stop_rx) = mpsc::channel(); + let handle = std::thread::Builder::new() + .name("fjall-halt-watchdog".into()) + .spawn({ + let (db, data, rescues) = (db.clone(), data.clone(), rescues.clone()); + move || watchdog(db, data, rescues, stop_rx) + })?; + Ok(( cursor.clone(), Self { @@ -207,6 +318,8 @@ impl FjallSnapshot { config, cursor, path: path.to_path_buf(), + rescues, + watchdog: Some((stop_tx, handle)), }, )) } @@ -222,10 +335,19 @@ impl FjallSnapshot { /// cache miss — without the serving path ever touching the writer. pub fn reader(&self) -> FjallReader { FjallReader { + db: self.db.clone(), data: self.data.clone(), + rescues: self.rescues.clone(), } } + /// Current LSM health. Cheap. A scraper should poll this on a + /// [`reader`](Self::reader) handle: `apply` takes `&mut self`, so the + /// snapshot itself is unreachable while a halted write is parked inside it. + pub fn stats(&self) -> FjallStats { + self.reader().stats() + } + /// Force a major compaction of the data partition, blocking until done. /// /// fjall's background compaction is write-driven: after a bulk hydration @@ -284,6 +406,7 @@ impl FjallSnapshot { FjallConfig { sync: true, cache_size_bytes: 0, + ..FjallConfig::default() }, )?; if staged_cursor != manifest.cursor { @@ -399,10 +522,23 @@ fn copy_db_dir(src: &Path, dst: &Path) -> Result<(), SnapshotError> { /// are safe to run concurrently with it. #[derive(Clone)] pub struct FjallReader { + db: Database, data: Keyspace, + rescues: Arc, } impl FjallReader { + /// Current LSM health of the data partition; see [`FjallStats`]. + pub fn stats(&self) -> FjallStats { + FjallStats { + l0_tables: self.data.l0_table_count(), + tables: self.data.table_count(), + outstanding_flushes: self.db.outstanding_flushes(), + active_compactions: self.db.active_compactions(), + halt_rescues: self.rescues.load(Ordering::Relaxed), + } + } + /// Live entry for `key`, or `None` if absent/deleted. pub fn get(&self, key: &str) -> Result, SnapshotError> { match self.data.get(key.as_bytes()).map_err(map_fjall)? { @@ -573,6 +709,7 @@ impl SnapshotStore for FjallSnapshot { FjallConfig { sync: true, cache_size_bytes: 0, + ..FjallConfig::default() }, )?; if staged_cursor != self.cursor { diff --git a/tests/bootstrap.rs b/tests/bootstrap.rs index 8c5e5e3..ab1fd93 100644 --- a/tests/bootstrap.rs +++ b/tests/bootstrap.rs @@ -283,6 +283,7 @@ mod fjall_bootstrap { FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() }, ) .expect("open fjall") @@ -294,6 +295,7 @@ mod fjall_bootstrap { FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() }, ) }, diff --git a/tests/multi_export.rs b/tests/multi_export.rs index d8816d3..950598a 100644 --- a/tests/multi_export.rs +++ b/tests/multi_export.rs @@ -752,6 +752,7 @@ mod fjall_multi { FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() } } diff --git a/tests/repair_dst.rs b/tests/repair_dst.rs index 01f9993..f3cc169 100644 --- a/tests/repair_dst.rs +++ b/tests/repair_dst.rs @@ -105,6 +105,7 @@ impl Fjall { slipstream::FjallConfig { sync: false, cache_size_bytes: 4 << 20, + ..Default::default() } } } diff --git a/tests/resync.rs b/tests/resync.rs index 13ee286..0164fee 100644 --- a/tests/resync.rs +++ b/tests/resync.rs @@ -422,6 +422,7 @@ mod fjall_resync { FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() } } diff --git a/tests/transport_s3.rs b/tests/transport_s3.rs index 564c341..91b30e4 100644 --- a/tests/transport_s3.rs +++ b/tests/transport_s3.rs @@ -257,6 +257,7 @@ async fn minio_full_bootstrap_loop_fjall() { let cfg = FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() }; full_bootstrap_loop( move |path| FjallSnapshot::open(path, cfg).expect("open fjall"), @@ -267,6 +268,7 @@ async fn minio_full_bootstrap_loop_fjall() { FjallConfig { sync: false, cache_size_bytes: 64 << 20, + ..Default::default() }, ) },