* fix: raise LMDB max_readers to avoid MDB_READERS_FULL (#783) heed's default reader table is 126 slots and fff opened envs in default TLS mode, so each long-lived reader thread pinned a slot for its lifetime. Long-lived embedders (Neovim, node agents) sharing one lock file across many processes/threads exhausted the table with MDB_READERS_FULL. Raise max_readers to 1024 (slots are ~64B, cost negligible) and expose FFF_LMDB_MAX_READERS for hosts to tune. NOTLS left for maintainer. Closes #783 * fix: open LMDB envs with MDB_NOTLS so reader slots free on txn drop (#783) Reader slots are now tied to txn objects instead of pinned per thread for the thread's lifetime, so long-lived embedders no longer accumulate slots. Env/RoTxn become WithoutTls-typed; RwTxn is unaffected. * fix(build): link advapi32 on Windows for lmdb-master-sys mdb_env_setup_locks references InitializeSecurityDescriptor / SetSecurityDescriptorDacl but lmdb-master-sys's build script never links advapi32; minimal test binaries fail with LNK2019 without it. * fix(test): link advapi32 in lmdb repro test binary on Windows The test links heed directly and rustc elides the unused fff lib, so build-script link flags never reach this binary; declare the dependency on advapi32 (mdb_env_setup_locks security-descriptor APIs) in the test. --------- Co-authored-by: Dmitriy Kovalenko <dmtr.kovalenko@outlook.com>
This commit is contained in:
@@ -14,7 +14,7 @@ pub struct DbHealth {
|
||||
}
|
||||
|
||||
pub trait DbHealthChecker {
|
||||
fn get_env(&self) -> &heed::Env;
|
||||
fn get_env(&self) -> &heed::Env<heed::WithoutTls>;
|
||||
fn is_healthy(&self) -> bool;
|
||||
/// Entries per database, each group has a static string label
|
||||
fn count_entries(&self) -> Result<Vec<(&'static str, u64)>>;
|
||||
|
||||
@@ -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<WithoutTls>,
|
||||
key: PathBuf,
|
||||
/// lmdb's env spec label
|
||||
label: &'static str,
|
||||
@@ -47,8 +47,8 @@ impl Drop for PooledEnv {
|
||||
pub(crate) struct SharedEnv(Arc<PooledEnv>);
|
||||
|
||||
impl Deref for SharedEnv {
|
||||
type Target = Env;
|
||||
fn deref(&self) -> &Env {
|
||||
type Target = Env<WithoutTls>;
|
||||
fn deref(&self) -> &Env<WithoutTls> {
|
||||
&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<String>) -> u32 {
|
||||
raw.and_then(|v| v.trim().parse::<u32>().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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<heed::WithoutTls> {
|
||||
&self.env
|
||||
}
|
||||
|
||||
|
||||
@@ -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<WithoutTls> {
|
||||
self.shared_env()
|
||||
}
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ pub struct QueryTracker {
|
||||
}
|
||||
|
||||
impl DbHealthChecker for QueryTracker {
|
||||
fn get_env(&self) -> &Env {
|
||||
fn get_env(&self) -> &Env<heed::WithoutTls> {
|
||||
&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<Bytes, SerdeBincode<VecDeque<HistoryEntry>>>,
|
||||
env: &Env,
|
||||
env: &Env<heed::WithoutTls>,
|
||||
project_key: &[u8; 32],
|
||||
offset: usize,
|
||||
) -> Result<Option<String>, Error> {
|
||||
|
||||
@@ -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::<bool>(); // 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"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user