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..bae071fa6 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,48 @@ 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 +674,73 @@ 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 +757,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 +772,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 +783,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 +800,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 +812,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..773d2fb7c 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,165 @@ 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; +} + +// 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 +// 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 = settledOsThreadCount(); + + // 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_TRUE(waitForOsThreadCount(baseline)) + << "reclaimed threads still in /proc/self/task: " << osThreadCount() << " vs baseline " + << 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 = settledOsThreadCount(); + + 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; }