MOD-15578 Track shared SVS thread pool memory & expose it through public API

src/VecSim/algorithms/svs/svs.h

@@ -361,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_{this->logCallbackCtx},
-          impl_{nullptr} {
+          is_two_level_lvq{isTwoLevelLVQ(params.quantBits)},
+          threadpool_{this->allocator, this->logCallbackCtx}, impl_{nullptr} {
         logger_ = makeLogger();
         if (params.num_threads != 0) {
             this->log(VecSimCommonStrings::LOG_WARNING_STRING,

src/VecSim/algorithms/svs/svs_utils.h

@@ -407,35 +407,62 @@ class VecSimSVSThreadPoolImpl {
         std::vector<ThreadSlot *> slots_;
     };
 
+    // Allocator type for the slots vector.
+    using SlotPtr = std::shared_ptr<ThreadSlot>;
+    using SlotVecAllocator = VecsimSTLAllocator<SlotPtr>;
+
     // 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) {
+    explicit VecSimSVSThreadPoolImpl(size_t num_threads = 1)
+        : allocator_(VecSimAllocator::newVecsimAllocator()), slots_(SlotVecAllocator(allocator_)) {
         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) {
-            slots_.push_back(std::make_shared<ThreadSlot>());
+            slots_.push_back(
+                std::allocate_shared<ThreadSlot>(VecsimSTLAllocator<ThreadSlot>(allocator_)));
         }
     }
 
+    // Set to true the first time instance() constructs the singleton. Allows other
+    // code paths (e.g., global stats reporting) to query whether the pool has been
+    // touched without forcing its lazy construction.
+    static std::atomic<bool> &initialized_flag() {
+        static std::atomic<bool> flag{false};
+        return flag;
+    }
+
 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), [](VecSimSVSThreadPoolImpl *) { /* leak at exit */ });
+        static auto shared_pool = [] {
+            auto p = std::shared_ptr<VecSimSVSThreadPoolImpl>(
+                new VecSimSVSThreadPoolImpl(1),
+                [](VecSimSVSThreadPoolImpl *) { /* leak at exit */ });
+            initialized_flag().store(true, std::memory_order_release);
+            return p;
+        }();
         return shared_pool;
     }
 
+    // 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).
     size_t size() const {
         std::lock_guard lock{pool_mutex_};
         return slots_.size() + 1;
     }
 
+    // Bytes currently allocated through the pool's internal allocator (the slots vector
+    // and the ThreadSlot objects). Does not include allocations performed by SVS itself
+    // outside of the pool, nor per-index wrapper state.
+    size_t getAllocationSize() const { return allocator_->getAllocationSize(); }
+
     // Physically resize the pool. Creates new OS threads on grow, shuts down idle threads
     // on shrink. new_size is total parallelism including the calling thread (minimum 1).
     // Occupied threads (held by renters) survive shrink via the deferred-resize protocol —
@@ -599,7 +626,8 @@ class VecSimSVSThreadPoolImpl {
             // 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::make_shared<ThreadSlot>());
+                slots_.push_back(
+                    std::allocate_shared<ThreadSlot>(VecsimSTLAllocator<ThreadSlot>(allocator_)));
             }
         } else {
             // Shrink.
@@ -615,8 +643,9 @@ class VecSimSVSThreadPoolImpl {
         }
     }
 
+    std::shared_ptr<VecSimAllocator> allocator_; // pool's own allocator for memory tracking
     mutable std::mutex pool_mutex_;
-    std::vector<std::shared_ptr<ThreadSlot>> slots_;
+    std::vector<SlotPtr, SlotVecAllocator> slots_;
     size_t pending_jobs_ = 0;             // jobs currently scheduled / in-flight
     std::optional<size_t> deferred_size_; // resize target deferred until pending_jobs_ == 0
 };
@@ -646,9 +675,14 @@ class VecSimSVSThreadPool {
     // 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(void *log_ctx = nullptr)
+    // parallelism_ is allocated through the provided VecsimAllocator so that the
+    // allocation is tracked by the index's memory accounting.
+    explicit VecSimSVSThreadPool(const std::shared_ptr<VecSimAllocator> &allocator,
+                                 void *log_ctx = nullptr)
         : pool_(VecSimSVSThreadPoolImpl::instance()),
-          parallelism_(std::make_shared<std::atomic<size_t>>(1)), log_ctx_(log_ctx) {}
+          parallelism_(std::allocate_shared<std::atomic<size_t>>(
+              VecsimSTLAllocator<std::atomic<size_t>>(allocator), 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); }
@@ -677,6 +711,20 @@ class VecSimSVSThreadPool {
     // Shared pool size — used by scheduling to decide how many reserve jobs to submit.
     static size_t poolSize() { return VecSimSVSThreadPoolImpl::instance()->size(); }
 
+    // Bytes allocated by the shared pool singleton. Returns 0 if the singleton has
+    // never been constructed (e.g., no SVS index was ever created and
+    // VecSim_UpdateThreadPoolSize was never called). Safe to call from any context;
+    // does not force singleton construction.
+    static size_t getSharedAllocationSize() {
+        if (!VecSimSVSThreadPoolImpl::isInitialized()) {
+            return 0;
+        }
+        return VecSimSVSThreadPoolImpl::instance()->getAllocationSize();
+    }
+
+    // True iff the shared pool singleton has been constructed.
+    static bool isSharedPoolInitialized() { return VecSimSVSThreadPoolImpl::isInitialized(); }
+
     // Delegates to the shared pool's parallel_for, passing the per-index log context.
     // n may be less than parallelism_ when the problem size is smaller than the
     // thread count (SVS computes n = min(arg.size(), pool.size())).

src/VecSim/utils/vec_utils.cpp

@@ -89,6 +89,9 @@ const char *VecSimCommonStrings::TIERED_SVS_UPDATE_THRESHOLD_STRING = "TIERED_SV
 const char *VecSimCommonStrings::TIERED_SVS_THREADS_RESERVE_TIMEOUT_STRING =
     "TIERED_SVS_THREADS_RESERVE_TIMEOUT";
 
+const char *VecSimCommonStrings::SHARED_SVS_THREADPOOL_MEMORY_STRING =
+    "SHARED_SVS_THREADPOOL_MEMORY";
+
 // Log levels
 const char *VecSimCommonStrings::LOG_DEBUG_STRING = "debug";
 const char *VecSimCommonStrings::LOG_VERBOSE_STRING = "verbose";

src/VecSim/utils/vec_utils.h

@@ -87,6 +87,9 @@ struct VecSimCommonStrings {
     static const char *TIERED_SVS_UPDATE_THRESHOLD_STRING;
     static const char *TIERED_SVS_THREADS_RESERVE_TIMEOUT_STRING;
 
+    // Memory allocated by the shared (global) SVS thread pool singleton.
+    static const char *SHARED_SVS_THREADPOOL_MEMORY_STRING;
+
     // Log levels
     static const char *LOG_DEBUG_STRING;
     static const char *LOG_VERBOSE_STRING;

src/VecSim/vec_sim.cpp

@@ -379,7 +379,17 @@ extern "C" VecSimIndexDebugInfo VecSimIndex_DebugInfo(VecSimIndex *index) {
 }
 
 extern "C" VecSimDebugInfoIterator *VecSimIndex_DebugInfoIterator(VecSimIndex *index) {
-    return index->debugInfoIterator();
+    auto *infoIterator = index->debugInfoIterator();
+    // Append the shared (global) SVS thread pool memory at the top level only when the
+    // pool has actually allocated memory (i.e., the singleton has been constructed).
+    size_t shared_pool_mem = VecSimSVSThreadPool::getSharedAllocationSize();
+    if (shared_pool_mem > 0) {
+        infoIterator->addInfoField(
+            VecSim_InfoField{.fieldName = VecSimCommonStrings::SHARED_SVS_THREADPOOL_MEMORY_STRING,
+                             .fieldType = INFOFIELD_UINT64,
+                             .fieldValue = {FieldValue{.uintegerValue = shared_pool_mem}}});
+    }
+    return infoIterator;
 }
 
 extern "C" VecSimIndexBasicInfo VecSimIndex_BasicInfo(VecSimIndex *index) {
@@ -390,6 +400,10 @@ extern "C" VecSimIndexStatsInfo VecSimIndex_StatsInfo(VecSimIndex *index) {
     return index->statisticInfo();
 }
 
+extern "C" size_t VecSim_GetGlobalMemory(void) {
+    return VecSimSVSThreadPool::getSharedAllocationSize();
+}
+
 extern "C" VecSimBatchIterator *VecSimBatchIterator_New(VecSimIndex *index, const void *queryBlob,
                                                         VecSimQueryParams *queryParams) {
     return index->newBatchIterator(queryBlob, queryParams);

src/VecSim/vec_sim.h

@@ -183,6 +183,18 @@ VecSimIndexBasicInfo VecSimIndex_BasicInfo(VecSimIndex *index);
  */
 VecSimIndexStatsInfo VecSimIndex_StatsInfo(VecSimIndex *index);
 
+/**
+ * @brief Return process-wide VecSim statistics that are not tied to any single index.
+ * Currently exposes the memory used by the shared SVS thread pool singleton.
+ * Safe to call without holding any index lock; does not force initialization of the
+ * shared SVS pool (returns 0 in fields whose backing singleton has not been touched).
+ *
+ * @return Total bytes currently allocated by VecSim outside any single index
+ *         (e.g. the shared SVS thread pool singleton). 0 if no such allocations
+ *         have been made.
+ */
+size_t VecSim_GetGlobalMemory(void);
+
 /**
  * @brief Returns an info iterator for generic reply purposes.
  *

src/VecSim/vec_sim_common.h

@@ -355,7 +355,10 @@ typedef struct {
  * production without worrying about performance
  */
 typedef struct {
-    size_t memory;
+    size_t memory;                // Memory tracked by the index's own allocator. Does NOT include
+                                  // process-wide allocations such as the shared SVS thread pool;
+                                  // those are reported via VecSim_GetGlobalMemory() so callers
+                                  // that aggregate across indexes don't double-count them.
     size_t numberOfMarkedDeleted; // The number of vectors that are marked as deleted (HNSW/tiered
                                   // only).
     size_t directHNSWInsertions;  // Count of vectors inserted directly into HNSW by main thread

tests/unit/test_svs_threadpool.cpp

@@ -44,13 +44,19 @@ class SVSThreadPoolTest : public ::testing::Test {
         // don't assert on nullptr log_ctx (we don't have an index context).
         saved_callback_ = VecSimIndexInterface::logCallback;
         VecSimIndexInterface::logCallback = nullptr;
+        // Reset the shared singleton pool to size 1 — earlier test suites may have
+        // resized it via VecSim_UpdateThreadPoolSize() and left it in that state.
+        VecSimSVSThreadPool::resize(1);
     }
     void TearDown() override {
         // Reset the shared singleton pool to size 1 so tests don't leak state.
         VecSimSVSThreadPool::resize(1);
         VecSimIndexInterface::logCallback = saved_callback_;
     }
 
+    // Allocator used by VecSimSVSThreadPool wrappers constructed in tests.
+    std::shared_ptr<VecSimAllocator> allocator_ = VecSimAllocator::newVecsimAllocator();
+
 private:
     logCallbackFunction saved_callback_ = nullptr;
 };
@@ -116,7 +122,7 @@ TEST_F(SVSThreadPoolTest, ShrinkWhileRented) {
     ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 5);
 
     // Wrapper A uses parallelism 3 → rents 2 workers (s0, s1).
-    VecSimSVSThreadPool wrapperA;
+    VecSimSVSThreadPool wrapperA{allocator_};
     wrapperA.setParallelism(3);
 
     std::latch hold(1);          // blocks rented workers
@@ -154,7 +160,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;
+    VecSimSVSThreadPool wrapperB{allocator_};
     // 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).
@@ -183,7 +189,7 @@ TEST_F(SVSThreadPoolTest, GrowWhileRented) {
     ASSERT_EQ(VecSimSVSThreadPool::poolSize(), 3);
 
     // Wrapper A uses parallelism 3 → rents 2 workers (s0, s1).
-    VecSimSVSThreadPool wrapperA;
+    VecSimSVSThreadPool wrapperA{allocator_};
     wrapperA.setParallelism(3);
 
     std::latch hold(1);          // blocks rented workers
@@ -219,7 +225,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;
+    VecSimSVSThreadPool wrapperB{allocator_};
     wrapperB.setParallelism(3);
     std::atomic_int resultB{0};
     wrapperB.parallel_for([&](size_t) { resultB++; }, 3);
@@ -250,7 +256,7 @@ TEST_F(SVSThreadPoolTest, GrowWhileRented) {
 TEST_F(SVSThreadPoolTest, ParallelismPropagationAcrossCopies) {
     VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool original;
+    VecSimSVSThreadPool original{allocator_};
     original.setParallelism(2);
     ASSERT_EQ(original.size(), 2);
 
@@ -282,8 +288,8 @@ TEST_F(SVSThreadPoolTest, ParallelismPropagationAcrossCopies) {
 TEST_F(SVSThreadPoolTest, TwoIndexesIndependentParallelism) {
     VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool wrapperA;
-    VecSimSVSThreadPool wrapperB;
+    VecSimSVSThreadPool wrapperA{allocator_};
+    VecSimSVSThreadPool wrapperB{allocator_};
 
     wrapperA.setParallelism(2);
     wrapperB.setParallelism(5);
@@ -366,9 +372,9 @@ TEST_F(SVSThreadPoolTest, ConcurrentRentalFromTwoIndexes) {
     // Pool size 8: wrappers A (4) and B (4) sum to exactly 8.
     VecSimSVSThreadPool::resize(8);
 
-    VecSimSVSThreadPool wrapperA;
+    VecSimSVSThreadPool wrapperA{allocator_};
     wrapperA.setParallelism(4);
-    VecSimSVSThreadPool wrapperB;
+    VecSimSVSThreadPool wrapperB{allocator_};
     wrapperB.setParallelism(4);
 
     std::atomic_int resultA{0};
@@ -445,7 +451,7 @@ TEST_F(SVSThreadPoolTest, AllThreadsOccupied) {
     // Pool size 4 (3 worker slots). Wrapper A rents all 3.
     VecSimSVSThreadPool::resize(4);
 
-    VecSimSVSThreadPool wrapperA;
+    VecSimSVSThreadPool wrapperA{allocator_};
     wrapperA.setParallelism(4);
 
     std::latch hold(1);
@@ -471,7 +477,7 @@ TEST_F(SVSThreadPoolTest, AllThreadsOccupied) {
         << resultA << ", pool_size=" << wrapperA.poolSize();
 
     // All 3 worker slots are occupied. Wrapper B tries to rent 1 worker.
-    VecSimSVSThreadPool wrapperB;
+    VecSimSVSThreadPool wrapperB{allocator_};
     wrapperB.setParallelism(2);
 
 #ifdef NDEBUG

tests/unit/test_svs_tiered.cpp

@@ -3931,7 +3931,8 @@ TEST(SVSTieredIndexTest, testThreadPool) {
     // Test VecSimSVSThreadPool with shared pool
     const size_t num_threads = 4;
     VecSimSVSThreadPool::resize(num_threads);
-    VecSimSVSThreadPool pool;
+    auto allocator = VecSimAllocator::newVecsimAllocator();
+    VecSimSVSThreadPool pool{allocator};
     ASSERT_EQ(pool.poolSize(), num_threads);
     ASSERT_EQ(pool.size(), 1); // parallelism starts at 1 (calling thread)
     ASSERT_EQ(pool.getParallelism(), 1);
@@ -3974,7 +3975,7 @@ TEST(SVSTieredIndexTest, testThreadPool) {
 
     // Test write-in-place mode (pool with size 1)
     VecSimSVSThreadPool::resize(1);
-    VecSimSVSThreadPool inplace_pool;
+    VecSimSVSThreadPool inplace_pool{allocator};
     inplace_pool.setParallelism(1);
     ASSERT_EQ(inplace_pool.size(), 1);
     ASSERT_EQ(inplace_pool.poolSize(), 1);
@@ -3984,7 +3985,7 @@ TEST(SVSTieredIndexTest, testThreadPool) {
 
     // parallel_for works immediately with default parallelism 1
     VecSimSVSThreadPool::resize(num_threads);
-    VecSimSVSThreadPool default_pool;
+    VecSimSVSThreadPool default_pool{allocator};
     counter = 0;
     default_pool.parallel_for(task, 1);
     ASSERT_EQ(counter, 1); // 0+1 = 1