address review

Author: meiravgri

src/VecSim/algorithms/svs/svs.h

@@ -38,13 +38,6 @@ struct SVSIndexBase
     SVSIndexBase() : num_marked_deleted{0} {};
     virtual ~SVSIndexBase() = default;
 
-    // Singleton accessor for the shared SVS thread pool.
-    // Always valid — initialized with size 1 (write-in-place mode: 0 worker threads,
-    // only the calling thread participates). Resized on VecSim_UpdateThreadPoolSize() calls.
-    static std::shared_ptr<VecSimSVSThreadPoolImpl> getSharedThreadPool() {
-        static auto shared_pool = std::make_shared<VecSimSVSThreadPoolImpl>(1);
-        return shared_pool;
-    }
     virtual int addVectors(const void *vectors_data, const labelType *labels, size_t n) = 0;
     virtual int deleteVectors(const labelType *labels, size_t n) = 0;
     virtual bool isLabelExists(labelType label) const = 0;
@@ -368,8 +361,8 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
           leanvec_dim{
               svs_details::getOrDefault(params.leanvec_dim, SVS_VAMANA_DEFAULT_LEANVEC_DIM)},
           epsilon{svs_details::getOrDefault(params.epsilon, SVS_VAMANA_DEFAULT_EPSILON)},
-          is_two_level_lvq{isTwoLevelLVQ(params.quantBits)},
-          threadpool_{SVSIndexBase::getSharedThreadPool(), this->logCallbackCtx}, impl_{nullptr} {
+          is_two_level_lvq{isTwoLevelLVQ(params.quantBits)}, threadpool_{this->logCallbackCtx},
+          impl_{nullptr} {
         logger_ = makeLogger();
         if (params.num_threads != 0) {
             this->log(VecSimCommonStrings::LOG_WARNING_STRING,

src/VecSim/algorithms/svs/svs_utils.h

@@ -403,11 +403,11 @@ class RentedThreads {
 // * Multiple callers can rent disjoint subsets of threads concurrently
 // * Shrinking while threads are rented is safe (shared_ptr lifecycle)
 class VecSimSVSThreadPoolImpl {
-public:
     // Create a pool with `num_threads` total parallelism (including the calling thread).
     // Spawns `num_threads - 1` worker OS threads. 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) {
         assert(num_threads && "VecSimSVSThreadPoolImpl should not be created with 0 threads");
         slots_.reserve(num_threads - 1);
@@ -416,6 +416,16 @@ class VecSimSVSThreadPoolImpl {
         }
     }
 
+public:
+    // Singleton accessor for the shared SVS thread pool.
+    // Always valid — initialized with size 1 (write-in-place mode: 0 worker threads,
+    // only the calling thread participates). Resized on VecSim_UpdateThreadPoolSize() calls.
+    static std::shared_ptr<VecSimSVSThreadPoolImpl> instance() {
+        static auto shared_pool =
+            std::shared_ptr<VecSimSVSThreadPoolImpl>(new VecSimSVSThreadPoolImpl(1));
+        return shared_pool;
+    }
+
     // Total parallelism: worker slots + 1 (the calling thread always participates).
     size_t size() const {
         std::lock_guard lock{pool_mutex_};
@@ -468,10 +478,9 @@ class VecSimSVSThreadPoolImpl {
 
         // Rent n-1 worker threads
         auto rented = rent(n - 1, log_ctx);
-        size_t num_workers = rented.count();
 
         // Assign work to rented workers (partitions 1..n-1)
-        for (size_t i = 0; i < num_workers; ++i) {
+        for (size_t i = 0; i < rented.count(); ++i) {
             rented[i].assign({&f, i + 1});
         }
 
@@ -485,7 +494,7 @@ class VecSimSVSThreadPoolImpl {
 
         // Wait for all rented workers and collect errors.
         // RentedThreads destructor will release the slots after this block.
-        manage_workers_after_run(main_thread_error, rented, num_workers);
+        manage_workers_after_run(main_thread_error, rented);
     }
 
 private:
@@ -526,8 +535,7 @@ class VecSimSVSThreadPoolImpl {
 
     // 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,
-                                  size_t rented_count) {
+    void manage_workers_after_run(const std::string &main_thread_error, RentedThreads &rented) {
         auto message = std::string{};
         auto inserter = std::back_inserter(message);
         bool has_error = !main_thread_error.empty();
@@ -536,7 +544,7 @@ class VecSimSVSThreadPoolImpl {
             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 < rented.count(); ++i) {
             auto &thread = rented[i];
             thread.wait();
             if (!thread.is_okay()) {
@@ -582,16 +590,16 @@ class VecSimSVSThreadPool {
     void *log_ctx_ = nullptr; // per-index log context
 
 public:
-    // Construct with reference to the shared pool singleton.
+    // Construct using the shared pool singleton.
     // parallelism_ starts at 1 (the calling thread always participates), matching the
     // pool's minimum size. Safe for immediate use in write-in-place mode without an
     // explicit setParallelism() call.
-    explicit VecSimSVSThreadPool(std::shared_ptr<VecSimSVSThreadPoolImpl> pool,
-                                 void *log_ctx = nullptr)
-        : pool_(std::move(pool)), parallelism_(std::make_shared<std::atomic<size_t>>(1)),
-          log_ctx_(log_ctx) {
-        assert(pool_ && "Pool must not be null");
-    }
+    explicit VecSimSVSThreadPool(void *log_ctx = nullptr)
+        : pool_(VecSimSVSThreadPoolImpl::instance()),
+          parallelism_(std::make_shared<std::atomic<size_t>>(1)), log_ctx_(log_ctx) {}
+
+    // Resize the shared pool singleton. Delegates to VecSimSVSThreadPoolImpl::instance().
+    static void resize(size_t new_size) { VecSimSVSThreadPoolImpl::instance()->resize(new_size); }
 
     // Set the degree of parallelism for this index's next operation.
     // n must be the number of threads actually reserved by the caller (i.e., the

src/VecSim/vec_sim.cpp

@@ -15,7 +15,7 @@
 #include "VecSim/vec_sim_index.h"
 #include "VecSim/vec_sim_adhoc_bf_ctx.h"
 #include "VecSim/types/bfloat16.h"
-#include "VecSim/algorithms/svs/svs.h"
+#include "VecSim/algorithms/svs/svs_utils.h"
 #include <cassert>
 #include "memory.h"
 
@@ -44,7 +44,7 @@ extern "C" void VecSim_UpdateThreadPoolSize(size_t new_size) {
     // Resize the shared SVS thread pool. resize() clamps to minimum size 1.
     // new_size == 0 → pool size 1 (only calling thread, write-in-place).
     // new_size >  0 → pool size new_size (new_size - 1 worker threads).
-    SVSIndexBase::getSharedThreadPool()->resize(std::max(new_size, size_t{1}));
+    VecSimSVSThreadPool::resize(std::max(new_size, size_t{1}));
 }
 
 static VecSimResolveCode _ResolveParams_EFRuntime(VecSimAlgo index_type, VecSimRawParam rparam,

tests/unit/test_svs.cpp

@@ -3305,9 +3305,8 @@ TEST(SVSTest, compute_distance) {
 // ---------------------------------------------------------------------------
 TEST(SVSTest, NumThreadsParamIgnored) {
     // Resize the shared singleton pool to a known size.
-    auto pool = SVSIndexBase::getSharedThreadPool();
-    pool->resize(2);
-    ASSERT_EQ(pool->size(), 2);
+    VecSimSVSThreadPool::resize(2);
+    ASSERT_EQ(VecSimSVSThreadPoolImpl::instance()->size(), 2);
 
     // Capture warning logs emitted by VecSim.
     std::string captured_log;
@@ -3330,7 +3329,7 @@ TEST(SVSTest, NumThreadsParamIgnored) {
     ASSERT_NE(index, nullptr);
 
     // The shared pool size must remain at 2 — num_threads was ignored.
-    ASSERT_EQ(pool->size(), 2);
+    ASSERT_EQ(VecSimSVSThreadPoolImpl::instance()->size(), 2);
 
     // The index reports the shared pool size, not the deprecated param value.
     VecSimIndexDebugInfo info = VecSimIndex_DebugInfo(index);
@@ -3359,7 +3358,7 @@ TEST(SVSTest, NumThreadsParamIgnored) {
     VecSimIndex_Free(index2);
 
     // Restore pool to default size and clear log callback.
-    pool->resize(1);
+    VecSimSVSThreadPool::resize(1);
     VecSimIndexInterface::logCallback = nullptr;
 }
 

tests/unit/test_svs_threadpool.cpp

@@ -45,7 +45,11 @@ class SVSThreadPoolTest : public ::testing::Test {
         saved_callback_ = VecSimIndexInterface::logCallback;
         VecSimIndexInterface::logCallback = nullptr;
     }
-    void TearDown() override { VecSimIndexInterface::logCallback = saved_callback_; }
+    void TearDown() override {
+        // Reset the shared singleton pool to size 1 so tests don't leak state.
+        VecSimSVSThreadPool::resize(1);
+        VecSimIndexInterface::logCallback = saved_callback_;
+    }
 
 private:
     logCallbackFunction saved_callback_ = nullptr;
@@ -55,7 +59,7 @@ class SVSThreadPoolTest : public ::testing::Test {
 // Test 1: VecSimSVSThreadPoolImpl::resize — grow and shrink
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, ResizeGrowAndShrink) {
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(1);
+    auto pool = VecSimSVSThreadPoolImpl::instance();
     ASSERT_EQ(pool->size(), 1);
 
     // Grow 1 → 4
@@ -91,11 +95,12 @@ TEST_F(SVSThreadPoolTest, ResizeGrowAndShrink) {
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, ShrinkWhileRented) {
     // Pool size 5: 4 worker slots [s0, s1, s2, s3].
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(5);
+    auto pool = VecSimSVSThreadPoolImpl::instance();
+    pool->resize(5);
     ASSERT_EQ(pool->size(), 5);
 
     // Wrapper A uses parallelism 3 → rents 2 workers (s0, s1).
-    VecSimSVSThreadPool wrapperA(pool);
+    VecSimSVSThreadPool wrapperA;
     wrapperA.setParallelism(3);
 
     std::latch hold(1);          // blocks rented workers
@@ -133,7 +138,7 @@ TEST_F(SVSThreadPoolTest, ShrinkWhileRented) {
 
     // While wrapperA's threads are still alive (blocked on latch), run
     // parallel_for on the shrunk pool with a second wrapper using a free slot.
-    VecSimSVSThreadPool wrapperB(pool);
+    VecSimSVSThreadPool wrapperB;
     // Parallelism 2 = 1 rented worker + calling thread. The pool has 3 slots
     // [s0, s1, s2] after shrink; s0 and s1 are occupied by wrapperA, so the
     // single rented worker will get s2 (the only free slot).
@@ -158,11 +163,12 @@ TEST_F(SVSThreadPoolTest, ShrinkWhileRented) {
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, GrowWhileRented) {
     // Pool size 3: 2 worker slots [s0, s1].
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(3);
+    auto pool = VecSimSVSThreadPoolImpl::instance();
+    pool->resize(3);
     ASSERT_EQ(pool->size(), 3);
 
     // Wrapper A uses parallelism 3 → rents 2 workers (s0, s1).
-    VecSimSVSThreadPool wrapperA(pool);
+    VecSimSVSThreadPool wrapperA;
     wrapperA.setParallelism(3);
 
     std::latch hold(1);          // blocks rented workers
@@ -198,7 +204,7 @@ TEST_F(SVSThreadPoolTest, GrowWhileRented) {
     // Wrapper B uses parallelism 3 → rents 2 workers. s0, s1 are occupied by
     // wrapperA, so it gets the 2 newly created slots s2, s3... but we only
     // need 2 of the 3 free slots (s2, s3 are free, only need 2).
-    VecSimSVSThreadPool wrapperB(pool);
+    VecSimSVSThreadPool wrapperB;
     wrapperB.setParallelism(3);
     std::atomic_int resultB{0};
     wrapperB.parallel_for([&](size_t) { resultB++; }, 3);
@@ -227,9 +233,9 @@ TEST_F(SVSThreadPoolTest, GrowWhileRented) {
 // the original is visible to SVS's copy when it calls size().
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, ParallelismPropagationAcrossCopies) {
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(8);
+    VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool original(pool);
+    VecSimSVSThreadPool original;
     original.setParallelism(2);
     ASSERT_EQ(original.size(), 2);
 
@@ -259,10 +265,10 @@ TEST_F(SVSThreadPoolTest, ParallelismPropagationAcrossCopies) {
 // sequentially using disjoint thread budgets.
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, TwoIndexesIndependentParallelism) {
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(8);
+    VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool wrapperA(pool);
-    VecSimSVSThreadPool wrapperB(pool);
+    VecSimSVSThreadPool wrapperA;
+    VecSimSVSThreadPool wrapperB;
 
     wrapperA.setParallelism(2);
     wrapperB.setParallelism(5);
@@ -306,7 +312,7 @@ TEST_F(SVSThreadPoolTest, TwoIndexesIndependentParallelism) {
 // The C API sets write mode and resizes the shared singleton pool.
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, UpdateThreadPoolSizeModeTransitions) {
-    auto pool = SVSIndexBase::getSharedThreadPool();
+    auto pool = VecSimSVSThreadPoolImpl::instance();
 
     // 0 → 4: switch to async mode, pool resizes to 4.
     VecSim_UpdateThreadPoolSize(4);
@@ -345,11 +351,11 @@ TEST_F(SVSThreadPoolTest, UpdateThreadPoolSizeModeTransitions) {
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, ConcurrentRentalFromTwoIndexes) {
     // Pool size 8: wrappers A (4) and B (4) sum to exactly 8.
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(8);
+    VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool wrapperA(pool);
+    VecSimSVSThreadPool wrapperA;
     wrapperA.setParallelism(4);
-    VecSimSVSThreadPool wrapperB(pool);
+    VecSimSVSThreadPool wrapperB;
     wrapperB.setParallelism(4);
 
     std::atomic_int resultA{0};
@@ -398,7 +404,7 @@ TEST_F(SVSThreadPoolTest, ConcurrentRentalFromTwoIndexes) {
         << "Timed out waiting for concurrent parallel_for calls. "
            "resultA="
         << resultA << ", resultB=" << resultB << ", idsA.size=" << idsA.size()
-        << ", idsB.size=" << idsB.size() << ", pool_size=" << pool->size();
+        << ", idsB.size=" << idsB.size() << ", pool_size=" << wrapperA.poolSize();
 
     ASSERT_EQ(resultA, 4);
     ASSERT_EQ(resultB, 4);
@@ -424,9 +430,9 @@ TEST_F(SVSThreadPoolTest, ConcurrentRentalFromTwoIndexes) {
 // ---------------------------------------------------------------------------
 TEST_F(SVSThreadPoolTest, AllThreadsOccupied) {
     // Pool size 4 (3 worker slots). Wrapper A rents all 3.
-    auto pool = std::make_shared<VecSimSVSThreadPoolImpl>(4);
+    VecSimSVSThreadPool::resize(4);
 
-    VecSimSVSThreadPool wrapperA(pool);
+    VecSimSVSThreadPool wrapperA;
     wrapperA.setParallelism(4);
 
     std::latch hold(1);
@@ -449,10 +455,10 @@ TEST_F(SVSThreadPoolTest, AllThreadsOccupied) {
     ASSERT_TRUE(wait_with_timeout(workers_ready, kTestTimeout))
         << "Timed out waiting for wrapper A's 3 workers to start. "
            "resultA="
-        << resultA << ", pool_size=" << pool->size();
+        << resultA << ", pool_size=" << wrapperA.poolSize();
 
     // All 3 worker slots are occupied. Wrapper B tries to rent 1 worker.
-    VecSimSVSThreadPool wrapperB(pool);
+    VecSimSVSThreadPool wrapperB;
     wrapperB.setParallelism(2);
 
 #ifdef NDEBUG

tests/unit/test_svs_tiered.cpp

@@ -3801,8 +3801,8 @@ TYPED_TEST(SVSTieredIndexTestBasic, testDeletedJournalMulti) {
 TEST(SVSTieredIndexTest, testThreadPool) {
     // Test VecSimSVSThreadPool with shared pool
     const size_t num_threads = 4;
-    auto shared_pool = std::make_shared<VecSimSVSThreadPoolImpl>(num_threads);
-    auto pool = VecSimSVSThreadPool(shared_pool);
+    VecSimSVSThreadPool::resize(num_threads);
+    VecSimSVSThreadPool pool;
     ASSERT_EQ(pool.poolSize(), num_threads);
     ASSERT_EQ(pool.size(), 1); // parallelism starts at 1 (calling thread)
     ASSERT_EQ(pool.getParallelism(), 1);
@@ -3846,8 +3846,8 @@ TEST(SVSTieredIndexTest, testThreadPool) {
 #endif
 
     // Test write-in-place mode (pool with size 1)
-    auto inplace_shared = std::make_shared<VecSimSVSThreadPoolImpl>(1);
-    auto inplace_pool = VecSimSVSThreadPool(inplace_shared);
+    VecSimSVSThreadPool::resize(1);
+    VecSimSVSThreadPool inplace_pool;
     inplace_pool.setParallelism(1);
     ASSERT_EQ(inplace_pool.size(), 1);
     ASSERT_EQ(inplace_pool.poolSize(), 1);
@@ -3856,7 +3856,8 @@ TEST(SVSTieredIndexTest, testThreadPool) {
     ASSERT_EQ(counter, 1);
 
     // parallel_for works immediately with default parallelism 1
-    auto default_pool = VecSimSVSThreadPool(shared_pool);
+    VecSimSVSThreadPool::resize(num_threads);
+    VecSimSVSThreadPool default_pool;
     counter = 0;
     default_pool.parallel_for(task, 1);
     ASSERT_EQ(counter, 1); // 0+1 = 1
@@ -3868,6 +3869,9 @@ TEST(SVSTieredIndexTest, testThreadPool) {
     pool.setParallelism(num_threads);
     ASSERT_THROW(pool.parallel_for(err_task, 1), svs::threads::ThreadingException);
     ASSERT_THROW(pool.parallel_for(err_task, num_threads), svs::threads::ThreadingException);
+
+    // Restore pool to default size so we don't leak state to other tests.
+    VecSimSVSThreadPool::resize(1);
 }
 
 #else // HAVE_SVS