diff --git a/crates/fff-core/src/dbs/db_healthcheck.rs b/crates/fff-core/src/dbs/db_healthcheck.rs index 77d834994..6ed8fd537 100644 --- a/crates/fff-core/src/dbs/db_healthcheck.rs +++ b/crates/fff-core/src/dbs/db_healthcheck.rs @@ -14,7 +14,7 @@ pub struct DbHealth { } pub trait DbHealthChecker { - fn get_env(&self) -> &heed::Env; + fn get_env(&self) -> &heed::Env; fn is_healthy(&self) -> bool; /// Entries per database, each group has a static string label fn count_entries(&self) -> Result>; diff --git a/crates/fff-core/src/dbs/env_pool.rs b/crates/fff-core/src/dbs/env_pool.rs index 0b8e1507d..73962ffdb 100644 --- a/crates/fff-core/src/dbs/env_pool.rs +++ b/crates/fff-core/src/dbs/env_pool.rs @@ -1,4 +1,4 @@ -use heed::{Env, EnvOpenOptions}; +use heed::{Env, EnvOpenOptions, WithoutTls}; use std::collections::HashMap; use std::fs; use std::ops::Deref; @@ -19,7 +19,7 @@ pub(crate) struct EnvSpec { } pub(crate) struct PooledEnv { - env: Env, + env: Env, key: PathBuf, /// lmdb's env spec label label: &'static str, @@ -47,8 +47,8 @@ impl Drop for PooledEnv { pub(crate) struct SharedEnv(Arc); impl Deref for SharedEnv { - type Target = Env; - fn deref(&self) -> &Env { + type Target = Env; + fn deref(&self) -> &Env { &self.0.env } } @@ -92,8 +92,11 @@ impl SharedEnv { erase_if_oversized(&path, spec); let result = unsafe { - let mut opts = EnvOpenOptions::new(); + // MDB_NOTLS: reader slots are tied to txn objects (freed on + // commit/abort) instead of pinned per thread for its lifetime (#783). + let mut opts = EnvOpenOptions::new().read_txn_without_tls(); opts.map_size(spec.map_size); + opts.max_readers(max_readers()); if spec.max_dbs > 0 { opts.max_dbs(spec.max_dbs); } @@ -219,6 +222,23 @@ const MAX_TRANSIENT_RETRIES: u32 = 8; // Concurrent mdb_env_open calls on the same path can race on macOS // this is for some reason fixable by simple retry of the open +// heed's default reader table is 126 slots. In TLS mode each thread pins a slot +// for its lifetime, so long-lived embedders (Neovim, node agents) that share one +// lock file across many processes/threads exhaust it (#783). Reader slots are +// tiny (~64B), so raise the ceiling; `FFF_LMDB_MAX_READERS` lets hosts tune it. +const DEFAULT_MAX_READERS: u32 = 1024; + +fn max_readers() -> u32 { + parse_max_readers(std::env::var("FFF_LMDB_MAX_READERS").ok()) +} + +// Never drop below heed's default 126; ignore missing/garbage/too-small values. +fn parse_max_readers(raw: Option) -> u32 { + raw.and_then(|v| v.trim().parse::().ok()) + .filter(|&n| n >= 126) + .unwrap_or(DEFAULT_MAX_READERS) +} + fn is_transient_env_open_error(err: &heed::Error) -> bool { match err { heed::Error::Io(io) => matches!( @@ -248,3 +268,17 @@ fn erase_if_oversized(db_path: &Path, spec: &EnvSpec) { let _ = fs::remove_file(&data); let _ = fs::remove_file(db_path.join("lock.mdb")); } + +#[cfg(test)] +mod tests { + use super::{DEFAULT_MAX_READERS, parse_max_readers}; + + #[test] + fn max_readers_parsing() { + assert_eq!(parse_max_readers(None), DEFAULT_MAX_READERS); + assert_eq!(parse_max_readers(Some("nan".into())), DEFAULT_MAX_READERS); + assert_eq!(parse_max_readers(Some("64".into())), DEFAULT_MAX_READERS); // below 126 floor + assert_eq!(parse_max_readers(Some(" 512 ".into())), 512); + assert_eq!(parse_max_readers(Some("126".into())), 126); + } +} diff --git a/crates/fff-core/src/dbs/frecency.rs b/crates/fff-core/src/dbs/frecency.rs index de8f9469e..863ce3ff2 100644 --- a/crates/fff-core/src/dbs/frecency.rs +++ b/crates/fff-core/src/dbs/frecency.rs @@ -43,7 +43,7 @@ const AI_MODIFICATION_THRESHOLDS: [(i64, u64); 5] = [ ]; impl DbHealthChecker for FrecencyTracker { - fn get_env(&self) -> &heed::Env { + fn get_env(&self) -> &heed::Env { &self.env } diff --git a/crates/fff-core/src/dbs/lmdb.rs b/crates/fff-core/src/dbs/lmdb.rs index 04361b370..842136412 100644 --- a/crates/fff-core/src/dbs/lmdb.rs +++ b/crates/fff-core/src/dbs/lmdb.rs @@ -1,4 +1,4 @@ -use heed::{Database, Env}; +use heed::{Database, Env, WithoutTls}; use std::path::Path; use std::sync::Arc; use std::sync::RwLock; @@ -124,7 +124,7 @@ pub(crate) trait LmdbStore: Sized + Send + Sync + 'static { fn health(&self) -> &DbHealth; /// Borrow the raw heed env. - fn env(&self) -> &Env { + fn env(&self) -> &Env { self.shared_env() } diff --git a/crates/fff-core/src/dbs/query_tracker.rs b/crates/fff-core/src/dbs/query_tracker.rs index e7ddf1994..a4b9eadcb 100644 --- a/crates/fff-core/src/dbs/query_tracker.rs +++ b/crates/fff-core/src/dbs/query_tracker.rs @@ -39,7 +39,7 @@ pub struct QueryTracker { } impl DbHealthChecker for QueryTracker { - fn get_env(&self) -> &Env { + fn get_env(&self) -> &Env { &self.env } @@ -198,7 +198,7 @@ impl QueryTracker { /// offset=0 returns most recent, offset=1 returns 2nd most recent, etc. fn read_history_at_offset( db: &Database>>, - env: &Env, + env: &Env, project_key: &[u8; 32], offset: usize, ) -> Result, Error> { diff --git a/crates/fff-core/tests/lmdb_readers_full_repro.rs b/crates/fff-core/tests/lmdb_readers_full_repro.rs new file mode 100644 index 000000000..4db9850fe --- /dev/null +++ b/crates/fff-core/tests/lmdb_readers_full_repro.rs @@ -0,0 +1,163 @@ +// Repro for #783: fff opens LMDB envs with only map_size set, leaving heed's +// default max_readers (126) and default TLS mode. Long-lived threads each pin a +// reader slot for the thread's lifetime, so >126 live reader threads exhaust the +// table with MDB_READERS_FULL. + +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Barrier, mpsc}; +use std::time::Duration; + +use heed::EnvOpenOptions; + +// This binary links heed directly without the fff lib, so nothing pulls in +// advapi32 for lmdb's security-descriptor calls in mdb_env_setup_locks. +#[cfg(windows)] +#[link(name = "advapi32")] +unsafe extern "C" {} + +fn temp_env_dir(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("fff-readers-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +// Regression for #783: with only map_size set (pre-fix), heed's default 126 +// reader slots are exhausted once >126 live threads each hold a read txn. fff now +// raises max_readers, so this many live readers must all get a slot. +const FFF_MAX_READERS: u32 = 1024; + +#[test] +fn raised_max_readers_admits_more_than_126_live_readers() { + let dir = temp_env_dir("raised"); + let env = unsafe { + EnvOpenOptions::new() + .map_size(10 * 1024 * 1024) + .max_readers(FFF_MAX_READERS) + .open(&dir) + } + .unwrap(); + + const THREADS: usize = 200; + let stop = Arc::new(AtomicBool::new(false)); + let ready = Arc::new(Barrier::new(THREADS + 1)); + let readers_full = Arc::new(AtomicBool::new(false)); + let (tx, rx) = mpsc::channel::(); // true = read txn acquired + + let mut handles = Vec::new(); + for _ in 0..THREADS { + let env = env.clone(); + let stop = stop.clone(); + let ready = ready.clone(); + let readers_full = readers_full.clone(); + let tx = tx.clone(); + handles.push(std::thread::spawn(move || { + match env.read_txn() { + Ok(txn) => { + tx.send(true).ok(); + ready.wait(); + while !stop.load(Ordering::Relaxed) { + std::thread::park_timeout(Duration::from_millis(5)); + } + drop(txn); // hold the slot for the whole test + } + Err(e) => { + if e.to_string().contains("MDB_READERS_FULL") { + readers_full.store(true, Ordering::Relaxed); + } + tx.send(false).ok(); + ready.wait(); + } + } + })); + } + drop(tx); + + // Collect exactly one result per thread; parked threads keep their tx clone + // alive, so we must not wait for the channel to close. + let acquired = AtomicUsize::new(0); + for _ in 0..THREADS { + if rx.recv().unwrap() { + acquired.fetch_add(1, Ordering::Relaxed); + } + } + ready.wait(); + + let acquired = acquired.load(Ordering::Relaxed); + stop.store(true, Ordering::Relaxed); + for h in handles { + h.join().unwrap(); + } + let _ = std::fs::remove_dir_all(&dir); + + // With max_readers raised, all 200 live reader threads must get a slot and + // none may see MDB_READERS_FULL. On the pre-fix default of 126 this plateaus + // at 126 and the rest fail. + assert!( + !readers_full.load(Ordering::Relaxed), + "MDB_READERS_FULL hit: only {acquired}/{THREADS} live reader threads got a slot" + ); + assert_eq!( + acquired, THREADS, + "all {THREADS} live reader threads should get a slot; got {acquired}" + ); +} + +// Structural fix for #783: with MDB_NOTLS a reader slot is tied to the txn +// object and freed on drop, not pinned per thread. 200 long-lived threads each +// open+drop a txn against the *default* 126-slot table; in TLS mode this +// plateaus at 126, in NOTLS mode every thread must succeed. +#[test] +fn notls_releases_slots_of_live_threads() { + let dir = temp_env_dir("notls"); + let env = unsafe { + EnvOpenOptions::new() + .read_txn_without_tls() + .map_size(10 * 1024 * 1024) + .open(&dir) + } + .unwrap(); + + const THREADS: usize = 200; + let stop = Arc::new(AtomicBool::new(false)); + let ready = Arc::new(Barrier::new(THREADS + 1)); + // Serialize txns so the test measures slot *release*, not concurrency. + let txn_gate = Arc::new(std::sync::Mutex::new(())); + let acquired = Arc::new(AtomicUsize::new(0)); + + let mut handles = Vec::new(); + for _ in 0..THREADS { + let env = env.clone(); + let stop = stop.clone(); + let ready = ready.clone(); + let txn_gate = txn_gate.clone(); + let acquired = acquired.clone(); + handles.push(std::thread::spawn(move || { + { + let _gate = txn_gate.lock().unwrap(); + if let Ok(txn) = env.read_txn() { + acquired.fetch_add(1, Ordering::Relaxed); + drop(txn); // NOTLS: slot returns to the pool here + } + } + // Stay alive: in TLS mode this thread would keep its slot pinned. + ready.wait(); + while !stop.load(Ordering::Relaxed) { + std::thread::park_timeout(Duration::from_millis(5)); + } + })); + } + + ready.wait(); + let got = acquired.load(Ordering::Relaxed); + stop.store(true, Ordering::Relaxed); + for h in handles { + h.join().unwrap(); + } + let _ = std::fs::remove_dir_all(&dir); + + assert_eq!( + got, THREADS, + "NOTLS must free slots on txn drop; only {got}/{THREADS} live threads got one" + ); +}