From f66e2546625c63ded5ad52451f20c71cd209005c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 18 Jul 2026 20:59:32 +0700 Subject: [PATCH] perf(shard): re-pin auxiliary threads off pinned shard cores (O5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mechanism (confirmed on real HW via ps -eLo psr): manifest-sync-N, spill-N, and moon-wal-sync-N are spawned FROM the shard's own pinned event-loop thread, and Linux threads inherit the parent's affinity mask at pthread_create — these aux threads weren't merely sharing a shard core, they were BORN wearing its exact single-core mask and could never leave (digest finding 6: ~12% CPU steal on shard cores under load; tmp/CPU-CACHE-DIGEST.md). Fix: when shards < cores, every auxiliary thread re-pins itself to the non-shard remainder (round-robin over cores shards..cores) as the first act of its closure — the same "escape the inherited mask" idiom the parallel-HNSW build workers already use. Covered: main accept/dispatch listener, manifest-sync-N, spill-N, moon-wal-sync-N, aof-writer[-N], auto-save, embedded AOF writer. No-op when shards >= cores, on non-Linux, or with MOON_NO_AUX_PIN=1 (bench A/B knob + operator escape hatch). Two deliberate exclusions: - The vector segment-reload pool stays scheduler-placed: reloads are latency-critical (queries await them) and the cold-start reload storm wants ALL cores while shards are idle — confining it to the small remainder risks the time-to-green regression PR #237 fixed. - The MAIN thread re-pins at the LAST moment (right before entering the accept loop, per-runtime), not at init: children spawned from main inherit its mask, so an early self-pin would have confined the reload/search pools and every other main-spawned thread to ONE aux core with no escape. Ordering reviewed and documented at both call sites. Adversarial review round widened coverage to every other pinned-parent spawn site in the tree: moon-heap-orphan-sweep-N (~40s crash-recovery I/O burst previously confined to its shard's core), the lazy moon-cold-read-N pool (previously born on whichever ONE shard core served the first cold read, stuck there for the process lifetime while serving ALL shards), moon-vec-snapshot-N, moon-vec-idx-gc, and moon-vec-compact-N (its last-core-downward heuristic landed workers on shard cores whenever shards ~= cores; now prefers the non-shard set via new numa::aux_core_at, heuristic kept as fallback). cpuset caveat documented: machine-wide core IDs can fail to pin under a restrictive cgroup cpuset - the thread keeps its inherited mask (debug-logged), never worse than before. jemalloc background-reclaim threads are a documented gap (spawned internally by jemalloc, no Rust-side affinity hook). Validation (GCE t2a-standard-8, 8 dedicated ARM vCPUs, shards=4, appendonly yes, 4 interleaved rounds, benchmark client tasksetted to cores 6-7 — conservative: pinned aux threads share those cores): - SET median 730.3K (pin) vs 719.6K (no-pin): +1.5% - GET median 918.5K vs 936.8K: -1.9%, inside the pre-registered -2% gate; GET spread tightened 23% -> 7% under pinning - Placement proof: pinned legs put all 12 aux threads on cores 4-7; unpinned legs show manifest-sync-N/spill-N exactly on shard cores 0-3 - A/B harness traps burned and fixed along the way: SO_REUSEPORT lets an orphan server silently split benchmark traffic (added pgrep clean-state aborts); redis-benchmark -q emits NO final summary once a run exceeds ~1s (switched to --csv); appendfsync everysec at 700K+ pipelined writes/s trips AOF backpressure and aborts the SET bench (A/B runs appendfsync no; the aof-writer threads still carry the full append stream). Refs tmp/CPU-CACHE-DIGEST.md (O5), task #26 author: Tin Dang --- CHANGELOG.md | 44 ++++++ src/main.rs | 31 ++++ src/persistence/manifest_sync.rs | 5 + src/persistence/wal_v3/sync_agent.rs | 5 + src/server/embedded.rs | 10 ++ src/shard/event_loop.rs | 7 + src/shard/numa.rs | 208 +++++++++++++++++++++++++++ src/storage/tiered/cold_read_pool.rs | 7 + src/storage/tiered/spill_thread.rs | 5 + src/vector/background_compact.rs | 17 ++- src/vector/persistence/manifest.rs | 5 + src/vector/reload_pool.rs | 14 ++ src/vector/store.rs | 3 + 13 files changed, 355 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 247d2c52..bd7c2b46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,50 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Both flags are CLI-only: a `moon.conf` value cannot reach jemalloc (its config is read at process start, before the conf file is parsed) and now triggers a loud startup warning instead of a silent no-op. +- **Auxiliary threads no longer contend with pinned shard cores (O5).** + `manifest-sync-N`, `spill-N`, and `moon-wal-sync-N` are spawned from + inside the shard's own (pinned) event-loop thread and, on Linux, + inherit that thread's exact single-core affinity mask at + `pthread_create` time — they weren't merely sharing the shard's core, + they were born wearing its mask and could never leave it (confirmed via + `ps -eLo psr`: `manifest-sync-N`/`spill-N` co-located on shard N's + core; the main accept/dispatch thread also floated onto a shard core, + ~12% CPU steal observed under load). When `--shards` leaves a + remainder of un-pinned cores, every auxiliary thread (main + accept/dispatch, `manifest-sync-N`, `spill-N`, `moon-wal-sync-N`, + the AOF writer(s), auto-save) now re-pins itself to that non-shard + core set (round-robin) as the first act of its closure — except the + main thread itself, which re-pins at the LAST moment before entering + the accept loop, because children spawned from main inherit its mask + and the vector pools must keep the full-machine mask. Adversarial + review widened coverage to every OTHER pinned-parent spawn site in + the tree: `moon-heap-orphan-sweep-N` (a ~40s crash-recovery I/O burst + previously confined to its shard's own core), the lazily-spawned + `moon-cold-read-N` pool (previously born on whichever ONE shard core + served the first cold read and stuck there for the process lifetime, + serving all shards), `moon-vec-snapshot-N`, `moon-vec-idx-gc`, and + the `moon-vec-compact-N` pool (its old last-core-downward heuristic + landed workers on shard cores whenever shards ≈ cores; it now prefers + the non-shard set via `aux_core_at`, keeping the heuristic as + fallback). cpuset caveat: core IDs come from the machine-wide online + list, so under a restrictive cgroup cpuset the pin can fail — the + thread then keeps its inherited mask (debug-logged), never worse than + before. No-op when + `--shards >= cores` (no remainder to place them on) or on non-Linux. + `MOON_NO_AUX_PIN=1` disables it. The vector segment-reload pool + (`moon-vec-reload-N`) is deliberately left scheduler-placed — reloads + are latency-critical and the cold-start/crash-recovery reload storm + wants every core while shard threads are idle; confining it to the + small non-shard remainder risks 3x worse cold-reload latency, the + regression PR #237 fixed. jemalloc's background-reclaim threads are + unaffected — jemalloc spawns them internally with no Rust-side + affinity hook. A/B on GCE t2a-standard-8 (8 dedicated ARM vCPUs, + shards=4, AOF on, 4 interleaved rounds vs `MOON_NO_AUX_PIN=1`): + SET median +1.5%, GET median −1.9% (inside noise; GET run-to-run + spread tightened 23%→7% under pinning), with `ps -eLo psr` placement + proof — pinned legs show all 12 aux threads on the non-shard cores + 4-7, unpinned legs show `manifest-sync-N`/`spill-N` sitting exactly + on shard cores 0-3. - **Shard offload paths are precomputed at shard init — the recurring tick paths no longer allocate (#45).** The 100ms eviction tick, the memory-pressure cascade, the 10s warm-transition check, and the diff --git a/src/main.rs b/src/main.rs index 3e0c0a0f..d5f4d943 100644 --- a/src/main.rs +++ b/src/main.rs @@ -543,6 +543,19 @@ fn main() -> anyhow::Result<()> { info!("Starting with {} shards", num_shards); + // O5: record the non-shard core set BEFORE any auxiliary thread spawns + // (AOF writers, the vector pools below, and — indirectly — the + // manifest-sync/spill/wal-sync threads each shard spawns internally). + // + // Deliberately NOT re-pinning the main thread here: threads spawned from + // main inherit its affinity mask at spawn time, and the vector + // search/reload pools just below must stay scheduler-placed across ALL + // cores (see src/vector/reload_pool.rs) — pinning main first would have + // them born wearing one aux core's mask with no escape. Main re-pins + // itself as "main-listener" at the LAST moment instead, right before it + // enters the accept loop, after every child thread has been spawned. + moon::shard::numa::init_aux_pinning(num_shards); + // FT.SEARCH intra-query worker pool: fan per-segment HNSW searches of one // query across workers (threads spawn eagerly here — the pool is tiny and // parks on its channel when vector search is unused). Default OFF: each @@ -820,6 +833,9 @@ fn main() -> anyhow::Result<()> { std::thread::Builder::new() .name(thread_name) .spawn(move || { + // O5: escape the shard-core mask this thread would + // otherwise inherit from its spawning thread. + moon::shard::numa::pin_current_aux_thread(&thread_name_inner); RuntimeFactoryImpl::block_on_local( thread_name_inner, aof::per_shard_aof_writer_task( @@ -866,6 +882,9 @@ fn main() -> anyhow::Result<()> { std::thread::Builder::new() .name("aof-writer".to_string()) .spawn(move || { + // O5: escape the shard-core mask this thread would + // otherwise inherit from its spawning thread. + moon::shard::numa::pin_current_aux_thread("aof-writer"); RuntimeFactoryImpl::block_on_local( "aof-writer".to_string(), aof::aof_writer_task( @@ -1792,6 +1811,12 @@ fn main() -> anyhow::Result<()> { .build() .expect("failed to build listener runtime"); + // O5: every child thread is spawned by now — safe to confine the + // accept/dispatch loop (this thread) to the non-shard core set. The + // tokio::spawn calls inside the block_on run on this same + // current-thread runtime, not on new OS threads. + moon::shard::numa::pin_current_aux_thread("main-listener"); + listener_rt.block_on(async { // Set up auto-save timer if save rules are configured (sharded mode) if config.save.is_some() { @@ -1903,6 +1928,9 @@ fn main() -> anyhow::Result<()> { std::thread::Builder::new() .name("auto-save".to_string()) .spawn(move || { + // O5: escape the shard-core mask this thread would + // otherwise inherit from its spawning thread. + moon::shard::numa::pin_current_aux_thread("auto-save"); RuntimeFactoryImpl::block_on_local( "auto-save".to_string(), moon::persistence::auto_save::run_auto_save_sharded( @@ -1925,6 +1953,9 @@ fn main() -> anyhow::Result<()> { // kernel distributes connections across all bound sockets, per-shard handles some // directly (no MPSC hop), central forwards the rest via conn_txs. let per_shard_accept = false; + // O5: every child thread is spawned by now — safe to confine the + // accept/dispatch loop (this thread) to the non-shard core set. + moon::shard::numa::pin_current_aux_thread("main-listener"); RuntimeFactoryImpl::block_on_local("listener".to_string(), async move { if let Err(e) = server::listener::run_sharded( config, diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs index 55b61f97..c8f94d9e 100644 --- a/src/persistence/manifest_sync.rs +++ b/src/persistence/manifest_sync.rs @@ -96,6 +96,11 @@ impl ManifestSyncAgent { match std::thread::Builder::new() .name(format!("manifest-sync-{shard_id}")) .spawn(move || { + // O5: this thread is spawned from the shard's own (pinned) + // event-loop thread and would otherwise inherit its exact + // single-core mask — re-pin to the non-shard core set as the + // first act, before anything else runs. + crate::shard::numa::pin_current_aux_thread(&format!("manifest-sync-{shard_id}")); // The spawner sends the io immediately after a successful // spawn; Err here means the agent was dropped first — exit. let Ok(io) = io_rx.recv() else { return }; diff --git a/src/persistence/wal_v3/sync_agent.rs b/src/persistence/wal_v3/sync_agent.rs index cadb2d7f..2a7cd8e5 100644 --- a/src/persistence/wal_v3/sync_agent.rs +++ b/src/persistence/wal_v3/sync_agent.rs @@ -116,6 +116,11 @@ impl WalSyncAgent { let thread = std::thread::Builder::new() .name(format!("moon-wal-sync-{shard_id}")) .spawn(move || { + // O5: this thread is spawned from the shard's own (pinned) + // event-loop thread and would otherwise inherit its exact + // single-core mask — re-pin to the non-shard core set as the + // first act, before anything else runs. + crate::shard::numa::pin_current_aux_thread(&format!("moon-wal-sync-{shard_id}")); // Requests arrive in LSN order (single producer); fetch_max // in publish() guards the watermark even if that ever // changes. diff --git a/src/server/embedded.rs b/src/server/embedded.rs index d6642b4b..19c8477f 100644 --- a/src/server/embedded.rs +++ b/src/server/embedded.rs @@ -134,6 +134,13 @@ pub async fn run_embedded( num_shards, config.bind, config.port ); + // O5: record the non-shard core set for this process before any + // auxiliary thread spawns below (AOF writer, and — indirectly — the + // manifest-sync/spill/wal-sync threads each shard spawns internally). + // Unlike main.rs, the CALLING thread here belongs to the embedding + // application, not to us — it is deliberately left untouched. + crate::shard::numa::init_aux_pinning(num_shards); + // One-time global init that the production binary normally performs. crate::admin::metrics_setup::init_global_slowlog( config.slowlog_max_len, @@ -165,6 +172,9 @@ pub async fn run_embedded( let handle = std::thread::Builder::new() .name("embedded-moon-aof".to_string()) .spawn(move || { + // O5: escape the shard-core mask this thread would otherwise + // inherit from its spawning thread. + crate::shard::numa::pin_current_aux_thread("embedded-moon-aof"); RuntimeFactoryImpl::block_on_local( "embedded-moon-aof".to_string(), aof::aof_writer_task(rx, aof_file_path, fsync, aof_token, None), diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index ae0682e8..3a4d216a 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -666,6 +666,13 @@ impl super::Shard { std::thread::Builder::new() .name(format!("moon-heap-orphan-sweep-{shard_id}")) .spawn(move || { + // O5: spawned from the pinned shard thread — escape + // the inherited single-core mask (this sweep can run + // ~40s of I/O; on the shard's own core it would + // contend with command processing the whole time). + crate::shard::numa::pin_current_aux_thread(&format!( + "moon-heap-orphan-sweep-{shard_id}" + )); for path in &pending_heap_orphans { crate::storage::tiered::kv_spill::remove_orphan_heap_file(path); } diff --git a/src/shard/numa.rs b/src/shard/numa.rs index 944d5497..b7122b26 100644 --- a/src/shard/numa.rs +++ b/src/shard/numa.rs @@ -100,6 +100,169 @@ pub fn pin_worker_to_core(core_id: usize) { } } +// ── O5: auxiliary-thread affinity ─────────────────────────────────────────── +// +// Shard threads pin to cores `0..num_shards` (see `pin_to_core`) and hold +// that pinning perfectly under load (0 cpu-migrations, measured). Everything +// else — the main accept/dispatch thread, manifest-sync-N, spill-N, +// moon-wal-sync-N, and the AOF writer / auto-save threads — is +// scheduler-placed. Several of these are spawned FROM an already-pinned +// shard thread (`ManifestSyncAgent::spawn`, `SpillThread::new`, +// `WalSyncAgent::spawn` all run inside the shard's own event loop), and Linux +// threads inherit their parent's affinity mask at `pthread_create` time — so +// without an explicit re-pin these auxiliary threads don't just "share" the +// shard core, they are BORN wearing that shard's exact single-core mask and +// can never leave it. That is the mechanism behind digest finding 6 (~12% +// steal observed on a shard core). +// +// The fix: when `shards < cores`, every auxiliary thread re-pins itself (as +// the first act of its closure, the same "escape the inherited mask" idiom +// `pin_worker_to_core` already uses for parallel HNSW build workers) to a +// core drawn round-robin from the non-shard remainder `shards..cores`. When +// `shards >= cores` there is no remainder — `aux_core_set` returns `None` and +// every call below becomes a no-op, deliberately: pinning to an empty set +// would fight the scheduler instead of helping it. +// +// EXCEPTION: the vector segment-reload pool (`moon-vec-reload-N`, +// src/vector/reload_pool.rs) is deliberately excluded from this fix. See the +// comment at its spawn site for why — short version: reloads are +// latency-critical and the reload storm at cold-start/crash-recovery wants +// every core while shard threads are still idle; confining it to the small +// non-shard remainder risks 3x worse cold-reload latency, the exact +// regression PR #237 (parallel HNSW / time-to-green) fixed. +// +// jemalloc's background-reclaim threads are NOT covered here: they are +// spawned internally by jemalloc itself (`background_thread:true` in +// `_rjem_malloc_conf`, see src/main.rs), with no Rust-side spawn hook or +// public API to redirect their affinity. Cross-thread `mallctl` affinity +// control does not exist in tikv-jemallocator/jemalloc's public surface. +// Left as a documented gap, not a silent omission. + +use std::sync::OnceLock; +use std::sync::atomic::{AtomicUsize, Ordering}; + +/// Non-shard core IDs for `num_shards` shards (pinned to cores +/// `0..num_shards`) out of `total_cores` machine cores. `None` when there is +/// no remainder (`num_shards >= total_cores`) — pure function, unit-testable +/// without touching real system topology. +fn aux_core_set_for(num_shards: usize, total_cores: usize) -> Option> { + if num_shards >= total_cores { + return None; + } + Some((num_shards..total_cores).collect()) +} + +/// Non-shard core IDs for `num_shards` shards on this machine +/// (`system_parallelism()` cores). See [`aux_core_set_for`]. +pub fn aux_core_set(num_shards: usize) -> Option> { + aux_core_set_for(num_shards, system_parallelism()) +} + +/// `MOON_NO_AUX_PIN=1` disables auxiliary-thread pinning entirely — a bench +/// A/B knob and operator escape hatch, mirroring the posture of +/// `--io-busy-poll-us` / `MOON_XSHARD_SPIN_BUDGET`: measured opt-in behavior +/// stays overridable, never a silent, un-disable-able tax. +fn aux_pinning_disabled() -> bool { + std::env::var_os("MOON_NO_AUX_PIN").is_some_and(|v| v == "1") +} + +/// Process-wide non-shard core set, computed once. `None` means "don't pin" +/// (shards >= cores, `MOON_NO_AUX_PIN=1`, or [`init_aux_pinning`] was never +/// called — e.g. library-embedded tests that skip startup wiring). +static AUX_CORES: OnceLock>> = OnceLock::new(); + +/// Round-robin cursor spreading auxiliary threads evenly across the +/// non-shard core set instead of piling every one of them onto its first +/// entry. +static AUX_RR: AtomicUsize = AtomicUsize::new(0); + +/// Record the non-shard core set for this process. Call once at startup, +/// after `num_shards` is resolved and BEFORE spawning any auxiliary thread +/// (the AOF pool, the vector reload/search pools, and the shard threads +/// themselves — the last of which spawn manifest-sync/spill/wal-sync +/// threads internally). First call wins (idempotent), matching the +/// `OnceLock` init pattern used by `reload_pool::init_global` / +/// `search_pool::init_global` elsewhere in this module's callers. +/// +/// Safe to call before any thread is spawned even though shard threads +/// inherit whatever affinity the calling thread has at spawn time: every +/// shard thread calls `pin_to_core` as the first act of its own closure, +/// unconditionally overriding whatever it inherited. +pub fn init_aux_pinning(num_shards: usize) { + let _ = AUX_CORES.get_or_init(|| { + if aux_pinning_disabled() { + None + } else { + aux_core_set(num_shards) + } + }); +} + +/// Deterministic per-index pick from the process's non-shard core set, for +/// pools that place worker `i` on a stable core instead of using the shared +/// round-robin cursor (e.g. the background-compact pool). `None` when +/// pinning is off for any of the [`pin_current_aux_thread`] reasons — +/// callers fall back to their own heuristic. +pub fn aux_core_at(i: usize) -> Option { + let Some(Some(cores)) = AUX_CORES.get() else { + return None; + }; + if cores.is_empty() { + return None; + } + Some(cores[i % cores.len()]) +} + +/// Pin the CURRENT thread onto the process's non-shard core set (round-robin +/// across it). Call as the first statement of every auxiliary thread's +/// closure — main accept/dispatch, manifest-sync-N, spill-N, +/// moon-wal-sync-N, aof-writer[-N], auto-save, moon-heap-orphan-sweep-N, +/// moon-cold-read-N, moon-vec-snapshot-N, moon-vec-idx-gc. (The +/// background-compact pool `moon-vec-compact-N` uses [`aux_core_at`] with +/// its own fallback instead.) NOT called by the vector segment-reload pool +/// (`moon-vec-reload-N`) — see the comment at its spawn site in +/// src/vector/reload_pool.rs for why that pool stays scheduler-placed. +/// +/// cpuset caveat: the core set comes from the machine-wide online-CPU list +/// (`system_parallelism`), so under a restrictive cgroup cpuset the chosen +/// core may be outside the allowed set — `set_for_current` then fails and +/// the thread simply KEEPS its inherited mask (debug-logged). Graceful: +/// never worse than the pre-pinning behavior. +/// +/// No-op when [`init_aux_pinning`] was never called, `shards >= cores`, or +/// `MOON_NO_AUX_PIN=1` — in all three cases this deliberately leaves +/// whatever affinity the thread already has (inherited or default) alone. +pub fn pin_current_aux_thread(label: &str) { + let Some(Some(cores)) = AUX_CORES.get() else { + return; + }; + // Defensive: aux_core_set_for never returns Some(vec![]), but a bare + // index % 0 would panic if that invariant ever slipped. + if cores.is_empty() { + return; + } + let idx = AUX_RR.fetch_add(1, Ordering::Relaxed) % cores.len(); + let core_id = cores[idx]; + + #[cfg(target_os = "linux")] + { + use core_affinity::CoreId; + if core_affinity::set_for_current(CoreId { id: core_id }) { + tracing::debug!("{label} pinned to non-shard core {core_id}"); + } else { + tracing::debug!("{label} failed to pin to non-shard core {core_id}"); + } + } + + #[cfg(not(target_os = "linux"))] + { + // No strict core-pinning equivalent on macOS (thread_policy_set is a + // QoS hint, not affinity) — no-op by design, per CLAUDE.md's + // Linux-only-code rule (compile guard + stub is sufficient). + let _ = (label, core_id); + } +} + /// Detect which NUMA node owns the given CPU core (Linux only). /// /// Reads `/sys/devices/system/node/nodeN/cpulist` to find the node. @@ -191,3 +354,48 @@ mod tests { assert!(system_parallelism() >= 1); } } + +// Not Linux-gated: pure arithmetic, no libc/core_affinity calls — must hold +// on macOS dev machines too. +#[cfg(test)] +mod aux_core_set_tests { + use super::*; + + #[test] + fn shards_less_than_cores_returns_remainder() { + assert_eq!(aux_core_set_for(4, 6), Some(vec![4, 5])); + assert_eq!(aux_core_set_for(1, 6), Some(vec![1, 2, 3, 4, 5])); + } + + #[test] + fn shards_equal_cores_is_noop() { + assert_eq!(aux_core_set_for(4, 4), None); + assert_eq!(aux_core_set_for(1, 1), None); + } + + #[test] + fn shards_greater_than_cores_is_noop() { + assert_eq!(aux_core_set_for(6, 4), None); + } + + #[test] + fn single_core_machine_is_always_noop() { + // shards=1, cores=1: the only case a 1-core box can be in (shards is + // clamped to >=1 elsewhere) — must never pin to an empty set. + assert_eq!(aux_core_set_for(1, 1), None); + } + + #[test] + fn zero_shards_edge_case() { + // Defensive: num_shards=0 shouldn't happen in practice (main.rs + // resolves auto-detect to >=1), but the function must not panic and + // must return the full core range rather than treat 0 as "unbounded". + assert_eq!(aux_core_set_for(0, 4), Some(vec![0, 1, 2, 3])); + } + + #[test] + fn zero_cores_edge_case_never_panics() { + assert_eq!(aux_core_set_for(0, 0), None); + assert_eq!(aux_core_set_for(4, 0), None); + } +} diff --git a/src/storage/tiered/cold_read_pool.rs b/src/storage/tiered/cold_read_pool.rs index 6b591243..cdf1e733 100644 --- a/src/storage/tiered/cold_read_pool.rs +++ b/src/storage/tiered/cold_read_pool.rs @@ -100,6 +100,13 @@ fn job_sender() -> &'static flume::Sender { // `read_cold_entry_async`'s send-failure branch) -- slow, but // still correct. Not worth propagating a spawn error here. let _ = build.spawn(move || { + // O5: this pool is lazily created on the FIRST cold read, + // from a pinned shard thread — without a re-pin every worker + // inherits that one shard's single-core mask and serves cold + // reads for ALL shards from it, forever. The non-shard + // remainder is a strict improvement; scheduler placement is + // not reachable from here (the inherited mask IS one core). + crate::shard::numa::pin_current_aux_thread(&format!("moon-cold-read-{i}")); for job in rx.iter() { let outcome = read_cold_entry(&job.shard_dir, job.location, job.now_ms, None); // If the requester's task was cancelled/dropped, this diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index 19a9469b..c47e3f26 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -476,6 +476,11 @@ impl SpillThread { let join_handle = std::thread::Builder::new() .name(format!("spill-{shard_id}")) .spawn(move || { + // O5: this thread is spawned from the shard's own (pinned) + // event-loop thread and would otherwise inherit its exact + // single-core mask — re-pin to the non-shard core set as the + // first act, before anything else runs. + crate::shard::numa::pin_current_aux_thread(&format!("spill-{shard_id}")); Self::run(request_rx, completion_tx, stop_flag_bg); }) .expect("failed to spawn spill thread"); diff --git a/src/vector/background_compact.rs b/src/vector/background_compact.rs index 84a558b0..d0301937 100644 --- a/src/vector/background_compact.rs +++ b/src/vector/background_compact.rs @@ -109,12 +109,17 @@ impl BackgroundCompactor { // SINGLE-core affinity mask on Linux — concurrent // small (sequential) builds then time-slice one core // (measured: 3 × 1K builds at ~1s each vs ~0.6s - // alone). Spread workers round-robin from the LAST - // core downward, away from shard 0. No-op on macOS. - let cores = crate::shard::numa::system_parallelism(); - crate::shard::numa::pin_worker_to_core( - cores.saturating_sub(1 + (i % cores)), - ); + // alone). O5: prefer the process's non-shard core + // set when aux pinning is initialized (the old + // last-core-downward heuristic lands on SHARD cores + // whenever shards ≈ cores, e.g. shards=6/cores=8 put + // workers 2-3 on shard cores 4-5); fall back to that + // heuristic when aux pinning is off. No-op on macOS. + let core = crate::shard::numa::aux_core_at(i).unwrap_or_else(|| { + let cores = crate::shard::numa::system_parallelism(); + cores.saturating_sub(1 + (i % cores)) + }); + crate::shard::numa::pin_worker_to_core(core); while let Ok(job) = rx.recv() { let result = match job.op { BuildOp::Compact { diff --git a/src/vector/persistence/manifest.rs b/src/vector/persistence/manifest.rs index 3a13691e..33999081 100644 --- a/src/vector/persistence/manifest.rs +++ b/src/vector/persistence/manifest.rs @@ -597,6 +597,11 @@ impl SnapshotPool { std::thread::Builder::new() .name(format!("moon-vec-snapshot-{i}")) .spawn(move || { + // O5: lazily spawned from a pinned shard thread — + // escape the inherited single-core mask. + crate::shard::numa::pin_current_aux_thread(&format!( + "moon-vec-snapshot-{i}" + )); loop { let next_job = { let mut guard = pending.lock(); diff --git a/src/vector/reload_pool.rs b/src/vector/reload_pool.rs index 2a112d67..510bf9ad 100644 --- a/src/vector/reload_pool.rs +++ b/src/vector/reload_pool.rs @@ -79,6 +79,20 @@ impl SegmentReloadPool { let st = Arc::clone(&state); std::thread::Builder::new() .name(format!("moon-vec-reload-{i}")) + // O5: deliberately NOT pinned via + // `numa::pin_current_aux_thread`, unlike every other + // auxiliary thread in this codebase. Segment reloads are + // latency-critical (a query awaits its own reload to + // finish — see SegmentReloadPool's callers) and the + // heaviest reload storm happens exactly when shard + // threads are idle: cold-start / crash recovery, where + // many segments reload in parallel to reach + // time-to-green. Confining that burst to the small + // non-shard remainder (`shards..cores`) instead of + // letting the scheduler spread it across ALL cores + // risks ~3x worse cold-reload latency — the exact + // regression PR #237 (parallel HNSW / time-to-green) + // fixed. Leave this pool scheduler-placed. .spawn(move || worker_loop(&rx, &st)) .expect("failed to spawn vector reload worker thread") }) diff --git a/src/vector/store.rs b/src/vector/store.rs index 87c3f8f0..af60db41 100644 --- a/src/vector/store.rs +++ b/src/vector/store.rs @@ -2027,6 +2027,9 @@ impl VectorStore { let spawned = std::thread::Builder::new() .name("moon-vec-idx-gc".to_owned()) .spawn(move || { + // O5: spawned from the owning (pinned) shard thread — + // escape the inherited single-core mask. + crate::shard::numa::pin_current_aux_thread("moon-vec-idx-gc"); if let Err(e) = std::fs::remove_dir_all(&idx_dir) { if e.kind() != std::io::ErrorKind::NotFound { tracing::warn!(