From 9bebb4136853ac68f8cd6f038ae2a6cbdf211031 Mon Sep 17 00:00:00 2001 From: Dor Forer Date: Tue, 7 Jul 2026 14:14:52 +0300 Subject: [PATCH 1/2] [MOD-16610] Lazy spawn + logical shrink for shared SVS pool: unblock CONFIG SET WORKERS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FT.CONFIG SET WORKERS N resizes the shared SVS thread pool synchronously on the Redis main thread. Each slot eagerly constructed an svs::threads::Thread — a serialized pthread_create plus a boot handshake that blocks until the worker reaches its spinning state — and freshly booted threads burn a 1M-pause idle spin with no work, oversubscribing the CPU so the handshakes stretch superlinearly: 52.5 s at N=2000 (vs ~60 us on 8.6, which only toggled the write mode). On Redis Enterprise this stalls the shard past CNM's 3 s socket timeout. Shrinking a warmed pool had the symmetric cost: a shutdown handshake plus join per spawned thread on the main thread (~3.9 s at N=2000). Make both resize directions O(slots) on the main thread — never O(threads): * Lazy spawn (grow): ThreadSlot holds std::optional, empty at creation. Resize only allocates slots; the OS thread is spawned by the renter that wins the occupancy CAS (ensureThread), interleaved with assign so the new thread's first spin window immediately catches its partition. Spawn cost moves to the first job that uses the slot, on a background thread. * Logical shrink: physical slots are never destroyed. Shrink lowers logical_size_ (tracked separately from capacity; grows/shrinks are classified against it) through the existing deferred-resize protocol; rent() only hands out slots below the limit. Excess threads park asleep (one final idle-spin window, no periodic wakeup) and are reused warm if the pool grows again. This mirrors the logical-only resize the per-index pools used before the shared pool existed, minus its defect: capacity can still grow above the high-water mark. Slots are heap-allocated and immortal, so renters' raw ThreadSlot pointers are structurally safe. * Thread reclaim ("GC"): reclaimExcessThreads() — noexcept, detach-under-lock / join-outside-lock — joins parked threads above the logical limit. Triggered only from the worker-executed completion path of a scheduled job (gated on isScheduled saved before the job self-deletes), so the periodic SVS GC and update jobs drive reclamation. Deliberately NOT in endScheduledJob(): that also fires from JobsRegistry teardown of unexecuted jobs on the Redis main thread, which must stay counter-only. Accepted gap: in write-in-place mode (WORKERS 0) or after the last SVS index is dropped, parked threads persist until async SVS activity resumes. * parallel_for gains a unified degraded-execution path: a spawn failure (std::system_error — thread exhaustion now happens at first use, mid-job), an assign failure (crashed worker), or a rent shortfall all fall back to running the missing partitions on the calling thread — partitions are never dropped (previously a shortfall silently ran only partition 0 in release builds). Crashed workers are retired to the unspawned state and lazily respawn on the next rent. * ExecuteMultiThreadJobImpl gets an RAII completion guard: a throwing task can no longer leave reserve jobs blocked forever or wedge the pending-jobs reservation that deferred shrink and reclaim depend on. Task failures are logged with index context by the typed wrappers. * VecSim_GetSharedMemory() contract change: slot capacity is a high-water mark (~64 bytes/slot), so the reported allocation no longer drops on shrink; sharedMemoryTracksThreadPoolResize is updated accordingly. Measured on 14 cores (pool-level bench): grow to N=2000 drops from 52.5 s to 0.19 ms on the main thread; warmed shrink from ~3.9 s to ~1 us (a size-field update). The joins (~3.1 s at N=2000 worst case) run on the job-completion path off the main thread; the first big rent pays the spawn cost off the main thread; reuse and warm regrow pay nothing. --- src/VecSim/algorithms/svs/svs_tiered.h | 77 +++++-- src/VecSim/algorithms/svs/svs_utils.h | 266 ++++++++++++++++++++----- tests/unit/test_svs.cpp | 26 ++- tests/unit/test_svs_threadpool.cpp | 139 ++++++++++++- tests/unit/test_svs_tiered.cpp | 105 ++++++++++ 5 files changed, 540 insertions(+), 73 deletions(-) diff --git a/src/VecSim/algorithms/svs/svs_tiered.h b/src/VecSim/algorithms/svs/svs_tiered.h index 535920365..42c31b1b0 100644 --- a/src/VecSim/algorithms/svs/svs_tiered.h +++ b/src/VecSim/algorithms/svs/svs_tiered.h @@ -149,16 +149,53 @@ class SVSMultiThreadJob : public AsyncJob { static void ExecuteMultiThreadJobImpl(AsyncJob *job) { auto *jobPtr = static_cast(job); auto controlBlock = jobPtr->controlBlock; - size_t num_threads = 1; - if (controlBlock) { - num_threads = controlBlock->waitForThreads(); + // Saved before the guard scope: the guard deletes the job, so jobPtr must not + // be read afterwards. + const bool should_reclaim = jobPtr->isScheduled; + + { + // Completion guard: whatever the task does — including throwing (e.g. a + // lazily spawned SVS pool thread failing with std::system_error, or an + // allocation failure mid-update) — the reserve jobs must be released and + // the job deleted. Otherwise reserved RediSearch workers block on the + // control block forever and the job's pending-scheduled-job reservation + // (endScheduledJob() in the dtor) never drops, wedging deferred pool + // resizes and thread reclamation. + struct CompletionGuard { + SVSMultiThreadJob *job; + const std::shared_ptr &controlBlock; + ~CompletionGuard() { + if (controlBlock) { + controlBlock->markJobDone(); + } + job->jobsRegistry->delete_job(job); + } + } guard{jobPtr, controlBlock}; + + size_t num_threads = 1; + if (controlBlock) { + num_threads = controlBlock->waitForThreads(); + } + assert(num_threads > 0); + try { + jobPtr->task(jobPtr->index, num_threads); + } catch (...) { + // Swallow after the fact: an async job has no caller to rethrow to (a + // throw here would unwind into the worker thread pool). Observability + // is handled by the typed task wrappers, which log with index context + // before rethrowing. + } } - assert(num_threads > 0); - jobPtr->task(jobPtr->index, num_threads); - if (controlBlock) { - jobPtr->controlBlock->markJobDone(); + + // Worker-executed completion path — the ONLY reclaim trigger. Runs after the + // guard released the job (so endScheduledJob() has applied any deferred + // logical shrink) and after the task released its index locks. Deliberately + // NOT in endScheduledJob() itself: that also fires from JobsRegistry teardown + // of unexecuted jobs on the Redis main thread, which must never join threads. + // reclaimExcessThreads() is noexcept. + if (should_reclaim) { + VecSimSVSThreadPoolImpl::instance()->reclaimExcessThreads(); } - jobPtr->jobsRegistry->delete_job(job); } SVSMultiThreadJob(std::shared_ptr allocator, JobType jobType, @@ -570,8 +607,16 @@ class TieredSVSIndex : public VecSimTieredIndex { std::lock_guard lock(index->updateJobMutex); // Release the scheduled flag to allow scheduling again index->indexUpdateScheduled.clear(); - // Update the SVS index - index->updateSVSIndex(availableThreads); + // Update the SVS index. A failed update must be visible, not a silent data lag: + // log with index context here (the generic job runner has no typed index access), + // then rethrow into the job's completion guard. + try { + index->updateSVSIndex(availableThreads); + } catch (const std::exception &error) { + index->backendIndex->log(VecSimCommonStrings::LOG_WARNING_STRING, + "tiered SVS index update job failed: %s", error.what()); + throw; + } } /** @@ -608,8 +653,16 @@ class TieredSVSIndex : public VecSimTieredIndex { } index->executeTracingCallback("GCJob::before_run_gc"); svs_index->setParallelism(std::min(availableThreads, index->backendIndex->indexSize())); - // VecSimIndexAbstract::runGC() is protected - static_cast(index->backendIndex)->runGC(); + // A failed GC must be visible: log with index context, then rethrow into the + // job's completion guard (see updateSVSIndexWrapper). + try { + // VecSimIndexAbstract::runGC() is protected + static_cast(index->backendIndex)->runGC(); + } catch (const std::exception &error) { + index->backendIndex->log(VecSimCommonStrings::LOG_WARNING_STRING, + "tiered SVS index GC job failed: %s", error.what()); + throw; + } } #ifdef BUILD_TESTS diff --git a/src/VecSim/algorithms/svs/svs_utils.h b/src/VecSim/algorithms/svs/svs_utils.h index 8dfa23d53..b0e176c98 100644 --- a/src/VecSim/algorithms/svs/svs_utils.h +++ b/src/VecSim/algorithms/svs/svs_utils.h @@ -347,11 +347,23 @@ struct SVSGraphBuilder { }; // A slot in the shared SVS thread pool. Wraps an SVS Thread with an occupancy flag -// used by the rental mechanism. Stored as shared_ptr in the pool so that deferred -// resize can safely shrink. Renters hold raw pointers (safe because the deferred-resize -// protocol prevents slot destruction while jobs are in flight). +// used by the rental mechanism. Slots are individually heap-allocated and are never +// destroyed (the pool only grows physically; shrink is logical), so renters can hold +// raw pointers safely. +// +// The OS thread is spawned lazily on first rent, not at slot creation. Growing the +// pool (VecSim_UpdateThreadPoolSize on `CONFIG SET WORKERS N`) runs on the Redis main +// thread; spawning N threads there is O(N) serialized thread-creates + boot handshakes +// (each svs::threads::Thread ctor blocks until its worker reaches the Spinning state), +// and freshly booted threads burn their idle-spin budget with no work, oversubscribing +// the CPU so the handshakes stretch superlinearly — seconds for large N (MOD-16610). +// Slot allocation is trivial, so resize returns immediately and each thread's spawn +// cost is paid by the renter on the first job that actually uses the slot — at which +// point the new thread's first spin window immediately catches its assigned partition. struct ThreadSlot { - svs::threads::Thread thread; + // Engaged on first ensureThread() call. Slots that are never rented never spawn + // an OS thread (and are destroyed for free on shrink — no shutdown handshake). + std::optional thread; std::atomic occupied{false}; ThreadSlot() = default; @@ -361,6 +373,18 @@ struct ThreadSlot { ThreadSlot &operator=(const ThreadSlot &) = delete; ThreadSlot(ThreadSlot &&) = delete; ThreadSlot &operator=(ThreadSlot &&) = delete; + + // Spawn the OS thread if not spawned yet. Must only be called by the renter that + // won the `occupied` compare-and-swap — that renter has exclusive ownership of the + // slot until it releases it, so no synchronization on `thread` is needed. + // May throw std::system_error if the OS refuses a new thread; callers must treat + // that as a degraded-execution trigger, not a fatal error. + svs::threads::Thread &ensureThread() { + if (!thread.has_value()) { + thread.emplace(); + } + return *thread; + } }; // Shared thread pool for SVS indexes with rental model. @@ -370,12 +394,14 @@ struct ThreadSlot { // * Multiple callers can rent disjoint subsets of threads concurrently // * Shrinking while threads are rented is safe (shared_ptr lifecycle) class VecSimSVSThreadPoolImpl { + using SlotPtr = std::shared_ptr; + // RAII guard for threads rented from the shared pool. On destruction, marks all - // rented slots as unoccupied (lock-free atomic stores). Uses raw pointers to - // avoid shared_ptr ref-counting overhead on the hot path. - // Safety: raw pointers are safe because the deferred-resize protocol ensures the - // pool cannot shrink (destroy slots) while scheduled jobs are in flight, and all - // multi-threaded SVS operations run within scheduled jobs. + // rented slots as unoccupied (lock-free atomic stores). Holds raw pointers — + // safe because slots are never destroyed: the pool only grows physically (shrink + // is logical), slots are individually heap-allocated (stable addresses across the + // vector's reallocation on grow), and the pool singleton is deliberately leaked + // at exit. class RentedThreads { public: RentedThreads() = default; @@ -392,9 +418,24 @@ class VecSimSVSThreadPoolImpl { size_t count() const { return slots_.size(); } + // Lazily spawn the OS thread for rented slot `i` (see ThreadSlot::ensureThread). + // May throw std::system_error on thread-resource exhaustion. + svs::threads::Thread &ensureThreadAt(size_t i) { + assert(i < slots_.size()); + return slots_[i]->ensureThread(); + } + + // Destroy slot `i`'s OS thread (joins it); the slot reverts to the unspawned + // state and will lazily respawn on a future rent. Used to retire crashed threads. + void resetThreadAt(size_t i) { + assert(i < slots_.size()); + slots_[i]->thread.reset(); + } + svs::threads::Thread &operator[](size_t i) { assert(i < slots_.size()); - return slots_[i]->thread; + assert(slots_[i]->thread.has_value() && "Rented slot must have a spawned thread"); + return *slots_[i]->thread; } private: @@ -408,15 +449,15 @@ class VecSimSVSThreadPoolImpl { std::vector slots_; }; - using SlotPtr = std::shared_ptr; - // Create a pool with `num_threads` total parallelism (including the calling thread). - // Spawns `num_threads - 1` worker OS threads. num_threads must be >= 1. + // Allocates `num_threads - 1` worker slots; OS threads are spawned lazily on first + // rent (see ThreadSlot). num_threads must be >= 1. // In write-in-place mode, the pool is created with num_threads == 1 (0 worker threads, // only the calling thread participates). // Private — use instance() to access the shared singleton. explicit VecSimSVSThreadPoolImpl(size_t num_threads = 1) - : allocator_(VecSimAllocator::newVecsimAllocator()), slots_(allocator_) { + : allocator_(VecSimAllocator::newVecsimAllocator()), slots_(allocator_), + logical_size_(num_threads - 1) { assert(num_threads && "VecSimSVSThreadPoolImpl should not be created with 0 threads"); slots_.reserve(num_threads - 1); for (size_t i = 0; i < num_threads - 1; ++i) { @@ -451,10 +492,12 @@ class VecSimSVSThreadPoolImpl { // Returns true iff instance() has ever been called (singleton constructed). static bool isInitialized() { return initialized_flag().load(std::memory_order_acquire); } - // Total parallelism: worker slots + 1 (the calling thread always participates). + // Total parallelism: logical worker count + 1 (the calling thread always + // participates). May be smaller than the physical slot capacity after a shrink — + // slots above the logical limit are retired lazily (see reclaimExcessThreads). size_t size() const { std::lock_guard lock{pool_mutex_}; - return slots_.size() + 1; + return logical_size_ + 1; } // Bytes currently allocated through the pool's internal allocator (the slots vector @@ -523,12 +566,27 @@ class VecSimSVSThreadPoolImpl { // has_attached_index_, deferred_size_, and pending_jobs_. Intended for unit // tests that need a clean baseline (the singleton itself is process-wide // and cannot be torn down). Caller must ensure no jobs are in flight. + // Number of slots whose OS thread is currently spawned (parked or in use). + // Test-only introspection: lets integration tests assert spawn/park/reclaim + // transitions without counting process-wide OS threads (which is noisy in a + // binary that runs thousands of unrelated tests). + size_t spawnedThreadCountForTest() const { + std::lock_guard lock{pool_mutex_}; + size_t count = 0; + for (const auto &slot : slots_) { + count += slot->thread.has_value() ? 1 : 0; + } + return count; + } + void resetForTest() { std::lock_guard lock{pool_mutex_}; assert(pending_jobs_ == 0 && "resetForTest called with jobs in flight"); // Swap with a fresh empty vector to release the capacity allocation - // (clear() destroys elements but retains capacity). + // (clear() destroys elements but retains capacity). Destroying the slots here + // joins any spawned threads — acceptable in tests. vecsim_stl::vector(allocator_).swap(slots_); + logical_size_ = 0; deferred_size_.reset(); has_attached_index_ = false; } @@ -538,10 +596,14 @@ class VecSimSVSThreadPoolImpl { size_t beginScheduledJob() { std::lock_guard lock{pool_mutex_}; ++pending_jobs_; - return slots_.size() + 1; + return logical_size_ + 1; } - // Decrement the pending-jobs counter. When it reaches zero, apply any deferred resize. + // Decrement the pending-jobs counter. When it reaches zero, apply any deferred + // resize. Counter-and-size bookkeeping ONLY — never joins threads: this also runs + // from JobsRegistry teardown of unexecuted jobs (FT.DROPINDEX, main thread). + // Thread reclamation happens separately, on the worker-executed job completion + // path (see reclaimExcessThreads). void endScheduledJob() { std::lock_guard lock{pool_mutex_}; assert(pending_jobs_ > 0 && "endScheduledJob called without matching beginScheduledJob"); @@ -551,6 +613,49 @@ class VecSimSVSThreadPoolImpl { } } + // Join and destroy the OS threads of slots parked above the logical limit — the + // "GC" that trims the pool back to its configured size after a shrink. The joins + // happen on the calling thread; call only from background contexts (a worker that + // just completed a scheduled job), never from the resize caller (the Redis main + // thread). Cheap no-op when nothing is parked. + // + // Safety without a pending-jobs gate: rent() only hands out slots below + // logical_size_, and a lowered logical_size_ only takes effect at a point where no + // job holds an older, larger size snapshot (the deferred-shrink protocol). So an + // unoccupied slot at index >= logical_size_ can never become rented; the set is + // stable once observed under the lock. Slots still occupied by a pre-shrink rental + // are skipped and reclaimed on a later pass. + // + // Best-effort and non-throwing: runs after the job completion guard, where an + // exception would unwind into the C worker-thread boundary. The local container is + // reserved before any thread is detached, so an allocation failure aborts the pass + // cleanly (threads simply stay parked for the next pass). + void reclaimExcessThreads() noexcept { + try { + std::vector doomed; + { + std::lock_guard lock{pool_mutex_}; + if (slots_.size() <= logical_size_) { + return; + } + doomed.reserve(slots_.size() - logical_size_); + for (size_t i = logical_size_; i < slots_.size(); ++i) { + auto &slot = *slots_[i]; + if (slot.thread.has_value() && + !slot.occupied.load(std::memory_order_acquire)) { + doomed.push_back(std::move(*slot.thread)); + slot.thread.reset(); + } + } + } + // ~Thread joins each detached thread (a parked worker wakes, sees + // RequestShutdown and exits promptly — no idle-spin burn), outside + // pool_mutex_ so a concurrent CONFIG SET resize is never blocked. + } catch (...) { + // Allocation failure — skip this pass; threads stay parked. + } + } + // Execute `f` in parallel with `n` partitions. The calling thread runs partition 0, // and up to `n-1` worker threads are rented for partitions 1..n-1. // Same signature as the SVS ThreadPool concept. @@ -570,25 +675,74 @@ class VecSimSVSThreadPoolImpl { return; } - // Rent n-1 worker threads + // Rent n-1 worker threads. A shortfall (fewer slots than requested) is handled + // below by the degraded-execution path, like every other dispatch failure. auto rented = rent(n - 1, log_ctx); - // Assign work to rented workers (partitions 1..n-1) + // Dispatch partitions 1..n-1 to rented workers: for each slot, lazily spawn its + // OS thread and immediately assign its partition (interleaved, so the fresh + // thread's first spin window catches the work; runs outside pool_mutex_ so + // background spawns never block a concurrent CONFIG SET resize). + // + // Degraded execution: if a spawn throws (std::system_error — thread-resource + // exhaustion) or an assign throws (worker crashed between rents), stop + // dispatching. Partitions [dispatched+1, n) run on the calling thread below — + // partitions are never dropped, whatever the trigger (spawn failure, assign + // failure, or rent shortfall). + size_t dispatched = 0; + std::string dispatch_error; for (size_t i = 0; i < rented.count(); ++i) { - rented[i].assign({&f, i + 1}); + try { + rented.ensureThreadAt(i); + rented[i].assign({&f, i + 1}); + } catch (const std::exception &error) { + dispatch_error = error.what(); + // If the failure came from a crashed worker, retire it so the slot + // lazily respawns a healthy thread on a future rent. Never throws + // past this point (joining an exited thread is cheap and safe). + rented.resetThreadAt(i); + break; + } + ++dispatched; } - // Run partition 0 on the calling thread - std::string main_thread_error; + if (!dispatch_error.empty() || dispatched < n - 1) { + auto msg = fmt::format("SVS thread pool: dispatched {} of {} partitions to " + "workers (rented {}); running the rest on the calling " + "thread.{}{}", + dispatched, n - 1, rented.count(), + dispatch_error.empty() ? "" : " Dispatch error: ", + dispatch_error); + if (VecSimIndexInterface::logCallback && log_ctx) { + VecSimIndexInterface::logCallback(log_ctx, "warning", msg.c_str()); + } + } + + // Run partition 0 on the calling thread, then any partitions that were not + // dispatched. Errors are collected per-partition (parity with worker behavior: + // one failing partition does not prevent the others from running). + auto message = std::string{}; + auto inserter = std::back_inserter(message); + bool has_error = false; try { f(0); } catch (const std::exception &error) { - main_thread_error = error.what(); + has_error = true; + fmt::format_to(inserter, "Thread 0: {}\n", error.what()); + } + for (size_t p = dispatched + 1; p < n; ++p) { + try { + f(p); + } catch (const std::exception &error) { + has_error = true; + fmt::format_to(inserter, "Partition {} (on calling thread): {}\n", p, + error.what()); + } } - // Wait for all rented workers and collect errors. + // Wait for all dispatched workers and collect errors. // RentedThreads destructor will release the slots after this block. - manage_workers_after_run(main_thread_error, rented); + manage_workers_after_run(has_error, std::move(message), dispatched, rented); } private: @@ -605,7 +759,10 @@ class VecSimSVSThreadPoolImpl { std::lock_guard lock{pool_mutex_}; size_t rented_count = 0; - for (auto &slot : slots_) { + // Only slots below the logical limit are rentable — slots above it (parked + // after a shrink) are reserved for reclamation and must not gain new renters. + for (size_t i = 0; i < logical_size_; ++i) { + auto &slot = slots_[i]; bool expected = false; if (slot->occupied.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) { rented.add(slot.get()); @@ -617,8 +774,8 @@ class VecSimSVSThreadPoolImpl { if (rented.count() < count) { auto msg = fmt::format("SVS thread pool: rented {} threads out of {} requested " - "(pool has {} slots). This should not happen.", - rented.count(), count, slots_.size()); + "(pool has {} rentable slots). This should not happen.", + rented.count(), count, logical_size_); if (VecSimIndexInterface::logCallback) { assert(log_ctx && "Log context must be provided when logging is available"); VecSimIndexInterface::logCallback(log_ctx, "warning", msg.c_str()); @@ -628,18 +785,14 @@ class VecSimSVSThreadPoolImpl { return rented; } - // Wait for all rented workers to finish. If any worker (or the main thread) threw, - // restart crashed workers and throw a combined exception. - void manage_workers_after_run(const std::string &main_thread_error, RentedThreads &rented) { - auto message = std::string{}; + // Wait for the first `dispatched` rented workers to finish (only those actually got + // a partition assigned). If any worker (or the calling thread) threw, retire crashed + // workers to the unspawned state and throw a combined exception. + void manage_workers_after_run(bool has_error, std::string message, size_t dispatched, + RentedThreads &rented) { auto inserter = std::back_inserter(message); - bool has_error = !main_thread_error.empty(); - if (has_error) { - fmt::format_to(inserter, "Thread 0: {}\n", main_thread_error); - } - - for (size_t i = 0; i < rented.count(); ++i) { + for (size_t i = 0; i < dispatched; ++i) { auto &thread = rented[i]; thread.wait(); if (!thread.is_okay()) { @@ -649,9 +802,8 @@ class VecSimSVSThreadPoolImpl { } catch (const std::exception &error) { fmt::format_to(inserter, "Thread {}: {}\n", i + 1, error.what()); } - // Restart the crashed thread so the slot is usable again. - thread.shutdown(); - thread = svs::threads::Thread{}; + // Retire the crashed thread; the slot lazily respawns on a future rent. + rented.resetThreadAt(i); } } @@ -662,33 +814,49 @@ class VecSimSVSThreadPoolImpl { // Actual resize logic. Caller must hold pool_mutex_. // Grow is always applied immediately. Shrink is deferred if pending_jobs_ > 0. + // + // Resize is LOGICAL: it moves logical_size_ (the limit rent() enforces) and only + // ever extends the physical slot vector — never destroys slots. Classification is + // against logical_size_, not slots_.size(): a grow from logical 1 to 4 inside a + // 2000-slot high-water pool is a grow. Shrinking parks the threads of slots above + // the new limit (they sleep; one final idle-spin window, no periodic wakeup); + // they are joined later by reclaimExcessThreads() on a background thread, or + // reused warm if the pool grows again first. This keeps both resize directions + // O(slots) on the Redis main thread — never O(thread create/join). void resize_locked(size_t new_size) { size_t target_workers = new_size - 1; - if (target_workers >= slots_.size()) { + if (target_workers >= logical_size_) { // Grow (or same size): apply immediately, cancel any pending deferred shrink. deferred_size_.reset(); for (size_t i = slots_.size(); i < target_workers; ++i) { slots_.push_back( std::allocate_shared(VecsimSTLAllocator(allocator_))); } + logical_size_ = target_workers; } else { // Shrink. if (pending_jobs_ > 0) { - // Defer shrink — jobs in flight may still need these threads. + // Defer shrink — jobs in flight snapshotted the old size and may + // still rent up to it. deferred_size_ = new_size; } else { - // Safe to shrink now — no jobs in flight. - // Occupied threads (held by renters) survive via shared_ptr. - // Idle threads are destroyed immediately. - slots_.resize(target_workers); + // Safe to shrink now — no jobs in flight. Purely logical: slots above + // the limit become unrentable; their threads (if spawned) stay parked + // until reclaimExcessThreads() joins them off the main thread. + logical_size_ = target_workers; } } } std::shared_ptr allocator_; // pool's own allocator for memory tracking mutable std::mutex pool_mutex_; + // Physical slots. Only grows (high-water mark); heap-allocated slots, so raw + // ThreadSlot* held by renters stay valid across vector reallocation on grow. vecsim_stl::vector slots_; + // Logical worker count = the rent() limit = configured pool size - 1. Slots at + // index >= logical_size_ are unrentable and eligible for thread reclamation. + size_t logical_size_ = 0; size_t pending_jobs_ = 0; // jobs currently scheduled / in-flight // Pending pool size to apply at the next safe point: either the first SVS index // attaches (onIndexAttached()) or pending_jobs_ drops to 0 (endScheduledJob()). diff --git a/tests/unit/test_svs.cpp b/tests/unit/test_svs.cpp index 415add1fc..5fae0acd1 100644 --- a/tests/unit/test_svs.cpp +++ b/tests/unit/test_svs.cpp @@ -3407,11 +3407,17 @@ TYPED_TEST(SVSTest, debugInfoSharedMemoryMatchesApi) { VecSimSVSThreadPool::resize(1); } -// VecSim shared memory must actually track the SVS thread-pool allocation: -// it grows when the pool grows and shrinks when the pool shrinks. Without this, -// SHARED_MEMORY could be a constant and debugInfoSharedMemoryMatchesApi would -// still pass (both readouts share the same getSharedAllocationSize() source). +// VecSim shared memory must track the SVS thread-pool slot allocation: it grows +// when the pool grows. Shrink is LOGICAL (MOD-16610): slot capacity persists at +// its high-water mark (slots are tiny structs; the OS threads — the expensive +// part — are reclaimed separately, off the main thread), so shared memory does +// NOT drop on shrink but must not grow either, and regrowing within the +// high-water mark must not allocate new slots. TYPED_TEST(SVSTest, sharedMemoryTracksThreadPoolResize) { + // Slot capacity is a process-wide high-water mark now (logical shrink), so a + // previous run of this typed test leaves the pool at capacity 8 and the grow + // below would allocate nothing. Reset to a clean pool first. + VecSimSVSThreadPoolImpl::instance()->resetForTest(); // With lazy init, resize() only records the requested size until an SVS index // has attached. Mark the pool attached up front so the resizes below apply // eagerly and their allocation effect is observable (idempotent, matches the @@ -3429,11 +3435,19 @@ TYPED_TEST(SVSTest, sharedMemoryTracksThreadPoolResize) { size_t mem_8 = VecSim_GetSharedMemory(); EXPECT_GT(mem_8, mem_baseline) << "shared memory must grow when the pool grows"; - // Shrink back to size 1 (still WriteAsync). + // Shrink back to size 1 (still WriteAsync). Logical shrink: capacity — and + // therefore the slot allocation — stays at the high-water mark. VecSim_UpdateThreadPoolSize(1); ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 1u); size_t mem_after = VecSim_GetSharedMemory(); - EXPECT_LT(mem_after, mem_8) << "shared memory must shrink when the pool shrinks"; + EXPECT_EQ(mem_after, mem_8) + << "logical shrink keeps slot capacity (high-water) — allocation must not change"; + + // Regrow within the high-water mark: reuses existing slots, no new allocation. + VecSim_UpdateThreadPoolSize(8); + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 8u); + EXPECT_EQ(VecSim_GetSharedMemory(), mem_8) + << "regrow within high-water capacity must not allocate new slots"; // Restore to default baseline. VecSim_UpdateThreadPoolSize(0); diff --git a/tests/unit/test_svs_threadpool.cpp b/tests/unit/test_svs_threadpool.cpp index cce1fc15c..e02e96ecb 100644 --- a/tests/unit/test_svs_threadpool.cpp +++ b/tests/unit/test_svs_threadpool.cpp @@ -16,6 +16,9 @@ #include #include +#ifdef __linux__ +#include +#endif #include "VecSim/algorithms/svs/svs.h" #include "VecSim/algorithms/svs/svs_utils.h" @@ -53,8 +56,10 @@ class SVSThreadPoolTest : public ::testing::Test { VecSimSVSThreadPool::resize(1); } void TearDown() override { - // Reset the shared singleton pool to size 1 so tests don't leak state. + // Reset the shared singleton pool to size 1 so tests don't leak state, and + // join any threads parked above the logical limit by the shrink. VecSimSVSThreadPool::resize(1); + VecSimSVSThreadPoolImpl::instance()->reclaimExcessThreads(); VecSimIndexInterface::logCallback = saved_callback_; } @@ -455,8 +460,8 @@ TEST_F(SVSThreadPoolTest, ConcurrentRentalFromTwoIndexes) { // Test 8: All threads occupied — graceful degradation // When all pool threads are rented by wrapper A, wrapper B's parallel_for // cannot rent any workers. In debug builds, rent() asserts. In release -// builds, parallel_for uses rented.count() (0 workers) and runs only -// partition 0 on the calling thread. +// builds, the degraded-execution path runs ALL partitions on the calling +// thread — a rent shortfall reduces parallelism, never the amount of work. // // NOTE: This should never happen in production. RediSearch's reserve job // mechanism guarantees that the number of concurrent renters never exceeds @@ -496,11 +501,12 @@ TEST_F(SVSThreadPoolTest, AllThreadsOccupied) { wrapperB.setParallelism(2); #ifdef NDEBUG - // Release: graceful degradation — rent() returns 0 workers, parallel_for - // runs only partition 0 on the calling thread. Work is silently dropped. + // Release: graceful degradation — rent() returns 0 workers, and the + // degraded-execution path runs both partitions on the calling thread. + // Partitions are never dropped. std::atomic_int resultB{0}; wrapperB.parallel_for([&](size_t) { resultB++; }, 2); - ASSERT_EQ(resultB, 1); // only partition 0 ran + ASSERT_EQ(resultB, 2); // all partitions ran (serially, on the caller) #else // Debug: rent() asserts because it can't fulfill the request. ASSERT_DEATH(wrapperB.parallel_for([&](size_t) {}, 2), @@ -513,4 +519,125 @@ TEST_F(SVSThreadPoolTest, AllThreadsOccupied) { ASSERT_EQ(resultA, 4); } +#ifdef __linux__ +// Count the OS threads of this process. Unlike the pool's allocation-size +// accounting (which only tracks slot objects, not OS stacks), this observes +// actual thread creation, which is what lazy spawn is about. +static size_t osThreadCount() { + size_t count = 0; + for ([[maybe_unused]] const auto &entry : + std::filesystem::directory_iterator("/proc/self/task")) { + ++count; + } + return count; +} + +// --------------------------------------------------------------------------- +// Test 9: Lazy spawn — resize allocates slots without creating OS threads; +// threads are spawned on first rent, reused on later rents, and shrink joins +// only the threads that were actually spawned (MOD-16610). +// --------------------------------------------------------------------------- +TEST_F(SVSThreadPoolTest, LazySpawnOnFirstRent) { + // Reach a steady state first: join any threads parked by earlier tests, so + // the baseline is stable. + auto pool = VecSimSVSThreadPoolImpl::instance(); + pool->reclaimExcessThreads(); + const size_t baseline = osThreadCount(); + + // Growing the pool must not spawn any OS thread — this is the main-thread + // cost CONFIG SET WORKERS pays. + VecSimSVSThreadPool::resize(9); // 8 worker slots + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 9); + ASSERT_EQ(osThreadCount(), baseline); + + // First parallel_for with 4 partitions rents 3 slots and spawns exactly + // 3 threads — the 5 never-rented slots stay unspawned. + std::atomic_int counter{0}; + pool->parallel_for([&](size_t) { counter++; }, 4); + ASSERT_EQ(counter, 4); + ASSERT_EQ(osThreadCount(), baseline + 3); + + // Second run at the same width reuses the spawned threads (rent() scans + // slots in order, so the same 3 slots are picked). + pool->parallel_for([&](size_t) { counter++; }, 4); + ASSERT_EQ(counter, 8); + ASSERT_EQ(osThreadCount(), baseline + 3); + + // Shrink to 1 is logical: no slot is destroyed and no thread is joined on + // the resize caller (that would stall the Redis main thread). The 3 spawned + // threads park above the logical limit. + VecSimSVSThreadPool::resize(1); + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 1); + ASSERT_EQ(osThreadCount(), baseline + 3); // parked, not yet joined + + // Reclamation joins them (in production this runs on the worker-executed + // job completion path; the integration test lives in test_svs_tiered.cpp). + pool->reclaimExcessThreads(); + ASSERT_EQ(osThreadCount(), baseline); +} + +// --------------------------------------------------------------------------- +// Test 11: Warm reuse across shrink/grow — threads parked by a logical shrink +// are reused (not respawned) if the pool grows again before any reclaim pass. +// Also: a deferred logical shrink applied at the endScheduledJob zero point +// joins nothing (endScheduledJob is counter-only; teardown paths run on the +// Redis main thread). +// --------------------------------------------------------------------------- +TEST_F(SVSThreadPoolTest, LogicalShrinkParksAndReusesThreads) { + auto pool = VecSimSVSThreadPoolImpl::instance(); + pool->reclaimExcessThreads(); + const size_t baseline = osThreadCount(); + + VecSimSVSThreadPool::resize(4); // 3 worker slots + std::atomic_int counter{0}; + pool->parallel_for([&](size_t) { counter++; }, 4); + ASSERT_EQ(counter, 4); + ASSERT_EQ(osThreadCount(), baseline + 3); + + // Deferred logical shrink: recorded while a scheduled job is pending, + // applied at the zero point — and applying it must NOT join any thread. + size_t snapshot = pool->beginScheduledJob(); + ASSERT_EQ(snapshot, 4); + VecSimSVSThreadPool::resize(1); + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 4); // deferred while pending + pool->endScheduledJob(); + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 1); // applied... + ASSERT_EQ(osThreadCount(), baseline + 3); // ...but nothing joined + + // Regrow before any reclaim pass: the parked threads are reused warm — + // the next parallel_for spawns nothing new. + VecSimSVSThreadPool::resize(4); + pool->parallel_for([&](size_t) { counter++; }, 4); + ASSERT_EQ(counter, 8); + ASSERT_EQ(osThreadCount(), baseline + 3); +} +#endif // __linux__ + +// --------------------------------------------------------------------------- +// Test 10: A worker that crashes (its partition throws) is retired to the +// unspawned state and the slot lazily respawns a healthy thread on the next +// rent — the pool stays fully usable after a partition failure. +// --------------------------------------------------------------------------- +TEST_F(SVSThreadPoolTest, CrashedWorkerRetiresAndRespawns) { + VecSimSVSThreadPool::resize(3); + auto pool = VecSimSVSThreadPoolImpl::instance(); + + // Partition 1 throws inside a worker thread; parallel_for collects the + // error and rethrows a combined ThreadingException. + ASSERT_THROW(pool->parallel_for( + [](size_t tid) { + if (tid == 1) { + throw std::runtime_error("partition failure"); + } + }, + 3), + svs::threads::ThreadingException); + + // The crashed worker was retired (joined, slot back to unspawned). The + // next parallel_for respawns it lazily and all partitions run. + std::atomic_int counter{0}; + pool->parallel_for([&](size_t) { counter++; }, 3); + ASSERT_EQ(counter, 3); +} + #endif // HAVE_SVS diff --git a/tests/unit/test_svs_tiered.cpp b/tests/unit/test_svs_tiered.cpp index 88f5b715a..a605cc73c 100644 --- a/tests/unit/test_svs_tiered.cpp +++ b/tests/unit/test_svs_tiered.cpp @@ -146,6 +146,111 @@ class SVSTieredIndexTest : public ::testing::Test { // TEST_DATA_T and TEST_DIST_T are defined in test_utils.h +// A task exception must not leak a reserve job: the completion guard in +// ExecuteMultiThreadJobImpl has to release the control block (and delete the +// job) even when the task throws — otherwise the reserved worker below would +// block forever and the test would time out. +TEST(SVSMultiThreadJobTest, CompletionGuardReleasesReserveJobsOnTaskException) { + auto allocator = VecSimAllocator::newVecsimAllocator(); + SVSMultiThreadJob::JobsRegistry registry(allocator); + auto jobs = SVSMultiThreadJob::createJobs( + allocator, SVS_BATCH_UPDATE_JOB, + [](VecSimIndex *, size_t) { throw std::runtime_error("task failure"); }, + /*index=*/nullptr, /*num_threads=*/2, std::chrono::milliseconds(100), ®istry); + ASSERT_EQ(jobs.size(), 2); + + std::atomic_bool reserve_done{false}; + std::thread reserver([&, job = jobs[1]] { + job->Execute(job); + reserve_done.store(true, std::memory_order_release); + }); + + // Run the main job on this thread. The task throws; the guard must swallow + // the exception (no caller to rethrow to) and release the reserve job. + ASSERT_NO_THROW(jobs[0]->Execute(jobs[0])); + + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!reserve_done.load(std::memory_order_acquire)) { + ASSERT_LT(std::chrono::steady_clock::now(), deadline) + << "Reserve job was never released after a task exception"; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + reserver.join(); +} + +// Reclaim of parked pool threads must happen through the REAL production trigger — +// the worker-executed completion path of a scheduled SVSMultiThreadJob — not only +// via a direct reclaimExcessThreads() call. This catches trigger-placement bugs +// such as gating reclaim on pending_jobs_ == 0 inside a scheduled job (self-block). +// Spawn/park/reclaim transitions are asserted via the pool's own spawned-thread +// accounting: process-wide OS thread counts are noisy in a binary that runs +// thousands of unrelated tests (the isolated SVSThreadPoolTest fixture covers the +// OS-thread-level laziness assertions). +TEST(SVSMultiThreadJobTest, ExecutedScheduledJobReclaimsParkedThreads) { + auto pool = VecSimSVSThreadPoolImpl::instance(); + // Full clean slate — earlier suites may have left spawned or parked slots. + pool->resetForTest(); + pool->onIndexAttached(); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 0u); + + // Warm 3 threads, then logically shrink — they park above the limit. + VecSimSVSThreadPool::resize(4); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 0u); // lazy: resize spawns nothing + std::atomic_int ran{0}; + pool->parallel_for([&](size_t) { ran++; }, 4); + ASSERT_EQ(ran, 4); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 3u); + VecSimSVSThreadPool::resize(1); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 3u); // parked, not joined + + // Execute a real scheduled job; its completion path must reclaim the parked + // threads (isScheduled gate + post-guard trigger). + auto allocator = VecSimAllocator::newVecsimAllocator(); + SVSMultiThreadJob::JobsRegistry registry(allocator); + auto jobs = SVSMultiThreadJob::createScheduledJobs( + allocator, SVS_BATCH_UPDATE_JOB, [](VecSimIndex *, size_t) {}, /*index=*/nullptr, + std::chrono::milliseconds(1), ®istry); + ASSERT_EQ(jobs.size(), 1); // pool size 1 → single job, no reserve jobs + jobs[0]->Execute(jobs[0]); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 0u); +} + +// Negative: destroying UNEXECUTED scheduled jobs (JobsRegistry teardown — the +// FT.DROPINDEX path, which runs on the Redis main thread) must apply a deferred +// logical shrink but must NOT join/reclaim any thread. +TEST(SVSMultiThreadJobTest, UnexecutedJobTeardownDoesNotReclaim) { + auto pool = VecSimSVSThreadPoolImpl::instance(); + // Full clean slate (see ExecutedScheduledJobReclaimsParkedThreads). + pool->resetForTest(); + pool->onIndexAttached(); + + VecSimSVSThreadPool::resize(4); + std::atomic_int ran{0}; + pool->parallel_for([&](size_t) { ran++; }, 4); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 3u); + + auto allocator = VecSimAllocator::newVecsimAllocator(); + { + SVSMultiThreadJob::JobsRegistry registry(allocator); + auto jobs = SVSMultiThreadJob::createScheduledJobs( + allocator, SVS_BATCH_UPDATE_JOB, [](VecSimIndex *, size_t) {}, + /*index=*/nullptr, std::chrono::milliseconds(1), ®istry); + ASSERT_GE(jobs.size(), 1); + // Shrink while the job is pending → deferred. + VecSimSVSThreadPool::resize(1); + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 4); + // Registry teardown deletes the unexecuted jobs; their dtors run + // endScheduledJob(), which applies the deferred logical shrink... + } + ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 1); + // ...but joins nothing — the parked threads are still alive. + ASSERT_EQ(pool->spawnedThreadCountForTest(), 3u); + + // Cleanup for subsequent tests. + pool->reclaimExcessThreads(); + ASSERT_EQ(pool->spawnedThreadCountForTest(), 0u); +} + template struct SVSIndexType { static constexpr VecSimType get_index_type() { return type; } From 7b401a927ee512e10477db1cbe30584143bc2950 Mon Sep 17 00:00:00 2001 From: Dor Forer Date: Wed, 8 Jul 2026 08:08:20 +0000 Subject: [PATCH 2/2] Fix CI: clang-format violations + flaky /proc/self/task assertions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pthread_join returning does not guarantee the joined thread's /proc/self/task entry is gone — the kernel wakes the joiner before it releases the task entry, so a just-joined thread can linger in the count briefly (caught by the sanitizer job's slowdown). Post-reclaim OS-thread counts now poll with a deadline, and baselines are captured only after the count holds steady; spawn-side assertions stay exact (thread creation is synchronous). --- src/VecSim/algorithms/svs/svs_utils.h | 18 +++++------ tests/unit/test_svs_threadpool.cpp | 46 +++++++++++++++++++++++++-- 2 files changed, 51 insertions(+), 13 deletions(-) diff --git a/src/VecSim/algorithms/svs/svs_utils.h b/src/VecSim/algorithms/svs/svs_utils.h index b0e176c98..bae071fa6 100644 --- a/src/VecSim/algorithms/svs/svs_utils.h +++ b/src/VecSim/algorithms/svs/svs_utils.h @@ -641,8 +641,7 @@ class VecSimSVSThreadPoolImpl { doomed.reserve(slots_.size() - logical_size_); for (size_t i = logical_size_; i < slots_.size(); ++i) { auto &slot = *slots_[i]; - if (slot.thread.has_value() && - !slot.occupied.load(std::memory_order_acquire)) { + if (slot.thread.has_value() && !slot.occupied.load(std::memory_order_acquire)) { doomed.push_back(std::move(*slot.thread)); slot.thread.reset(); } @@ -707,12 +706,12 @@ class VecSimSVSThreadPoolImpl { } if (!dispatch_error.empty() || dispatched < n - 1) { - auto msg = fmt::format("SVS thread pool: dispatched {} of {} partitions to " - "workers (rented {}); running the rest on the calling " - "thread.{}{}", - dispatched, n - 1, rented.count(), - dispatch_error.empty() ? "" : " Dispatch error: ", - dispatch_error); + auto msg = + fmt::format("SVS thread pool: dispatched {} of {} partitions to " + "workers (rented {}); running the rest on the calling " + "thread.{}{}", + dispatched, n - 1, rented.count(), + dispatch_error.empty() ? "" : " Dispatch error: ", dispatch_error); if (VecSimIndexInterface::logCallback && log_ctx) { VecSimIndexInterface::logCallback(log_ctx, "warning", msg.c_str()); } @@ -735,8 +734,7 @@ class VecSimSVSThreadPoolImpl { f(p); } catch (const std::exception &error) { has_error = true; - fmt::format_to(inserter, "Partition {} (on calling thread): {}\n", p, - error.what()); + fmt::format_to(inserter, "Partition {} (on calling thread): {}\n", p, error.what()); } } diff --git a/tests/unit/test_svs_threadpool.cpp b/tests/unit/test_svs_threadpool.cpp index e02e96ecb..773d2fb7c 100644 --- a/tests/unit/test_svs_threadpool.cpp +++ b/tests/unit/test_svs_threadpool.cpp @@ -532,6 +532,44 @@ static size_t osThreadCount() { return count; } +// pthread_join returning does NOT guarantee the joined thread's /proc/self/task +// entry is gone: the kernel wakes the joiner before it releases the task entry, +// so a just-joined thread can linger in the count briefly (observed under +// sanitizer slowdown in CI). Assertions about counts reached via joins must +// poll. Thread *creation* is synchronous (the entry exists when pthread_create +// returns), so upper-bound assertions after spawns can stay exact. +static bool waitForOsThreadCount(size_t expected) { + auto deadline = std::chrono::steady_clock::now() + kTestTimeout; + while (osThreadCount() != expected) { + if (std::chrono::steady_clock::now() >= deadline) { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return true; +} + +// Baseline capture after a reclaim has the same exit-lag problem: a lingering +// task entry would inflate the baseline and make later equality checks fail +// low. Poll until the count holds steady for a while before trusting it. +static size_t settledOsThreadCount() { + auto deadline = std::chrono::steady_clock::now() + kTestTimeout; + size_t last = osThreadCount(); + auto stable_since = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + size_t cur = osThreadCount(); + if (cur != last) { + last = cur; + stable_since = std::chrono::steady_clock::now(); + } else if (std::chrono::steady_clock::now() - stable_since >= + std::chrono::milliseconds(50)) { + break; + } + } + return last; +} + // --------------------------------------------------------------------------- // Test 9: Lazy spawn — resize allocates slots without creating OS threads; // threads are spawned on first rent, reused on later rents, and shrink joins @@ -542,7 +580,7 @@ TEST_F(SVSThreadPoolTest, LazySpawnOnFirstRent) { // the baseline is stable. auto pool = VecSimSVSThreadPoolImpl::instance(); pool->reclaimExcessThreads(); - const size_t baseline = osThreadCount(); + const size_t baseline = settledOsThreadCount(); // Growing the pool must not spawn any OS thread — this is the main-thread // cost CONFIG SET WORKERS pays. @@ -573,7 +611,9 @@ TEST_F(SVSThreadPoolTest, LazySpawnOnFirstRent) { // Reclamation joins them (in production this runs on the worker-executed // job completion path; the integration test lives in test_svs_tiered.cpp). pool->reclaimExcessThreads(); - ASSERT_EQ(osThreadCount(), baseline); + ASSERT_TRUE(waitForOsThreadCount(baseline)) + << "reclaimed threads still in /proc/self/task: " << osThreadCount() << " vs baseline " + << baseline; } // --------------------------------------------------------------------------- @@ -586,7 +626,7 @@ TEST_F(SVSThreadPoolTest, LazySpawnOnFirstRent) { TEST_F(SVSThreadPoolTest, LogicalShrinkParksAndReusesThreads) { auto pool = VecSimSVSThreadPoolImpl::instance(); pool->reclaimExcessThreads(); - const size_t baseline = osThreadCount(); + const size_t baseline = settledOsThreadCount(); VecSimSVSThreadPool::resize(4); // 3 worker slots std::atomic_int counter{0};