[MOD-12668] Implement asynchronous SVS GC execution using SVSMultiThreadJob (#840)

* Implement asynchronous SVS GC execution using SVSMultiThreadJob

* Add runGCParallel test

* [MOD-12668] Add micro benchmark for SVS GC with threads (#844)

* add micro benchmark for GC with threads

* fix assert

Co-authored-by: meiravgri <[email protected]>

---------

Co-authored-by: meiravgri <[email protected]>

* Fix micro benchmark for SVS GC with threads

---------

Co-authored-by: alon <[email protected]>
Co-authored-by: alonre24 <[email protected]>
Co-authored-by: meiravgri <[email protected]>

Author: 3 people

src/VecSim/algorithms/svs/svs_tiered.h

@@ -225,6 +225,8 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
     // Used to prevent scheduling multiple index update jobs at the same time.
     // As far as the update job does a batch update, job queue should have just 1 job at the moment.
     std::atomic_flag indexUpdateScheduled = ATOMIC_FLAG_INIT;
+    // Used to prevent scheduling multiple index GC jobs at the same time.
+    std::atomic_flag indexGCScheduled = ATOMIC_FLAG_INIT;
     // Used to prevent running multiple index update jobs in parallel.
     // Even if update jobs scheduled sequentially, they can be started in parallel.
     mutable std::mutex updateJobMutex;
@@ -549,6 +551,39 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
         index->updateSVSIndex(availableThreads);
     }
 
+    /**
+     * @brief Run SVS index GC in a thread-safe manner.
+     *
+     * This static wrapper function performs the following actions:
+     * - Acquires a lock on the index's mainIndexGuard to ensure thread safety during the GC
+     * - Configures the number of threads for the underlying SVS index update operation.
+     * - Calls the SVSIndex::runGC() method to perform the actual index update.
+     * - Clears the indexGCScheduled flag to allow future scheduling.
+     *
+     * @param idx Pointer to the VecSimIndex to be updated.
+     * @param availableThreads The number of threads available for the update operation. Current
+     * thread us used as well, so the minimal value is 1.
+     * @note no need to implement extra non-static method, as GC logic is simple enough to be done
+     * here.
+     */
+    static void SVSIndexGCWrapper(VecSimIndex *idx, size_t availableThreads) {
+        assert(availableThreads > 0);
+        auto index = static_cast<TieredSVSIndex<DataType> *>(idx);
+        assert(index);
+
+        std::lock_guard lock{index->mainIndexGuard};
+        // Release the scheduled flag to allow scheduling again
+        index->indexGCScheduled.clear();
+
+        // Do SVS index GC
+        index->backendIndex->log(VecSimCommonStrings::LOG_VERBOSE_STRING,
+                                 "running asynchronous GC for tiered SVS index");
+        auto svs_index = index->GetSVSIndex();
+        svs_index->setNumThreads(std::min(availableThreads, index->backendIndex->indexSize()));
+        // VecSimIndexAbstract::runGC() is protected
+        static_cast<VecSimIndexInterface *>(index->backendIndex)->runGC();
+    }
+
 #ifdef BUILD_TESTS
 public:
 #endif
@@ -565,6 +600,19 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
         this->submitJobs(jobs);
     }
 
+    void scheduleSVSIndexGC() {
+        // do not schedule if scheduled already
+        if (indexGCScheduled.test_and_set()) {
+            return;
+        }
+
+        auto total_threads = this->GetSVSIndex()->getThreadPoolCapacity();
+        auto jobs = SVSMultiThreadJob::createJobs(
+            this->allocator, SVS_GC_JOB, SVSIndexGCWrapper, this, total_threads,
+            std::chrono::microseconds(updateJobWaitTime), &uncompletedJobs);
+        this->submitJobs(jobs);
+    }
+
 private:
     static void applySwapsToLabelsArray(std::vector<size_t> &labels,
                                         const std::vector<swap_record> &swaps) {
@@ -962,10 +1010,8 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
 
     void runGC() override {
         TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING,
-                   "running asynchronous GC for tiered SVS index");
-        std::lock_guard lock{this->mainIndexGuard};
-        // VecSimIndexAbstract::runGC() is protected
-        static_cast<VecSimIndexInterface *>(this->backendIndex)->runGC();
+                   "scheduling asynchronous GC for tiered SVS index");
+        scheduleSVSIndexGC();
     }
 
     void acquireSharedLocks() override {

src/VecSim/vec_sim_common.h

@@ -243,6 +243,7 @@ typedef enum {
     HNSW_SEARCH_JOB,
     HNSW_SWAP_JOB,
     SVS_BATCH_UPDATE_JOB,
+    SVS_GC_JOB,
     INVALID_JOB // to indicate that finding a JobType >= INVALID_JOB is an error
 } JobType;
 

tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h

@@ -19,11 +19,8 @@ BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimSVS, BM_FUNC_NAME(BM_RunGC), DATA_TYPE_INDE
 BENCHMARK_REGISTER_F(BM_VecSimSVS, BM_FUNC_NAME(BM_RunGC))
     ->Unit(benchmark::kMillisecond)
     ->Iterations(1)
-    ->Arg(50)
-    ->Arg(100)
-    ->Arg(250)
-    ->Arg(500)
-    ->ArgName("num_deletions");
+    ->ArgsProduct({{50, 100, 500}, {1, 4}})
+    ->ArgNames({"num_deletions", "thread_count"});
 
 // AddLabel one by one
 BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimSVS, BM_FUNC_NAME(BM_AddLabelOneByOne), DATA_TYPE_INDEX_T)

tests/benchmark/bm_vecsim_svs.h

@@ -407,10 +407,13 @@ template <typename index_type_t>
 void BM_VecSimSVS<index_type_t>::RunGC(benchmark::State &st) {
 
     size_t num_deletions = st.range(0);
-    auto mock_thread_pool = tieredIndexMock(1);
-    ASSERT_EQ(mock_thread_pool.thread_pool_size, 1);
+    auto mock_thread_pool = tieredIndexMock(st.range(1));
+    ASSERT_EQ(mock_thread_pool.thread_pool_size, st.range(1));
     auto *tiered_index = CreateTieredSVSIndexFromFile(mock_thread_pool, 1);
 
+    // start threadpool
+    mock_thread_pool.init_threads();
+
     // Delete vectors, not yet triggering consolidation.
     for (size_t i = 0; i < num_deletions; ++i) {
         int ret = VecSimIndex_DeleteVector(tiered_index, i);
@@ -421,6 +424,7 @@ void BM_VecSimSVS<index_type_t>::RunGC(benchmark::State &st) {
     ASSERT_EQ(info.svsInfo.numberOfMarkedDeletedNodes, num_deletions);
     for (auto _ : st) {
         VecSimTieredIndex_GC(tiered_index);
+        mock_thread_pool.thread_pool_wait();
     };
     // num deleted should be 0
     info = VecSimIndex_DebugInfo(tiered_index);

tests/unit/test_svs_tiered.cpp

@@ -3126,8 +3126,19 @@ TYPED_TEST(SVSTieredIndexTestBasic, runGCAPI) {
     ASSERT_EQ(tiered_index->GetSVSIndex()->indexStorageSize(), n);
     auto size_before_gc = tiered_index->getAllocationSize();
 
+    auto jobs_before_gc = mock_thread_pool.jobQ.size();
     // Run the GC API call, expect that we will clean up the SVS index.
     VecSimTieredIndex_GC(tiered_index);
+    // Expected that GC jobs were added to the queue.
+    ASSERT_EQ(mock_thread_pool.jobQ.size(), jobs_before_gc + mock_thread_pool.thread_pool_size);
+    // Run GC twice.
+    VecSimTieredIndex_GC(tiered_index);
+    // Expected that no new GC jobs were added to the queue.
+    ASSERT_EQ(mock_thread_pool.jobQ.size(), jobs_before_gc + mock_thread_pool.thread_pool_size);
+    // Wait for any pending jobs to complete. As far as SVS GC is done via a job.
+    mock_thread_pool.init_threads();
+    mock_thread_pool.thread_pool_join();
+    // Validate sizes after GC.
     ASSERT_EQ(tiered_index->indexSize(), n - threshold);
     ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), n - threshold);
     ASSERT_EQ(tiered_index->GetSVSIndex()->indexStorageSize(), n - threshold);
@@ -3138,6 +3149,68 @@ TYPED_TEST(SVSTieredIndexTestBasic, runGCAPI) {
     EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 0);
 }
 
+TYPED_TEST(SVSTieredIndexTestBasic, runGCParallel) {
+    // Create TieredSVS index instance with a mock queue.
+    size_t dim = 4;
+    size_t threshold = 1024;
+    const size_t n = threshold * 4;
+    SVSParams params = {.type = TypeParam::get_index_type(), .dim = dim, .metric = VecSimMetric_L2};
+    VecSimParams svs_params = CreateParams(params);
+    auto mock_thread_pool = tieredIndexMock();
+
+    // Force trigger the first update job for 64 first vectors.
+    auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, 64);
+    ASSERT_INDEX(tiered_index);
+    auto allocator = tiered_index->getAllocator();
+
+    // Insert n vectors directly to SVS.
+    std::srand(10); // create pseudo random generator with any arbitrary seed.
+    for (size_t i = 0; i < n; i++) {
+        TEST_DATA_T vector[dim];
+        for (size_t j = 0; j < dim; j++) {
+            vector[j] = std::rand() / (TEST_DATA_T)RAND_MAX;
+        }
+        VecSimIndex_AddVector(tiered_index->GetBackendIndex(), vector, i);
+    }
+
+    ASSERT_EQ(tiered_index->indexSize(), n);
+    ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), n);
+
+    // Initialize the thread pool to start processing jobs.
+    mock_thread_pool.init_threads();
+
+    // Run the mess of add, delete, GC
+    for (size_t i = 0; i < threshold; i++) {
+        // Run GC for every 64 iterations.
+        if (i % 64 == 0) {
+            VecSimTieredIndex_GC(tiered_index);
+        }
+        // Add a new vector
+        TEST_DATA_T vector[dim];
+        for (size_t j = 0; j < dim; j++) {
+            vector[j] = std::rand() / (TEST_DATA_T)RAND_MAX;
+        }
+        VecSimIndex_AddVector(tiered_index, vector, n + i);
+        // Delete an existing vector
+        tiered_index->deleteVector(i + threshold);
+    }
+    // Final GC after all operations.
+    VecSimTieredIndex_GC(tiered_index);
+    // Wait for any pending jobs to complete. As far as SVS GC is done via a job.
+    mock_thread_pool.thread_pool_join();
+
+    // Validate sizes after GC.
+    auto tiered_size = tiered_index->indexSize();
+    auto flat_size = tiered_index->GetFlatIndex()->indexSize();
+    auto backend_size = tiered_index->GetBackendIndex()->indexSize();
+    ASSERT_EQ(tiered_size, n);
+    ASSERT_EQ(tiered_size, backend_size + flat_size);
+    // Expect that GC cleaned all the deleted vectors.
+    ASSERT_EQ(tiered_index->GetSVSIndex()->indexStorageSize(), backend_size);
+    ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0);
+    EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 0);
+}
+
 TYPED_TEST(SVSTieredIndexTestBasic, switchDeleteModes) {
     // Create TieredSVS index instance with a mock queue.
     size_t dim = 16;