From 3211f8218b548f4943633cd2de0cc11af9fa14d2 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Sun, 4 Oct 2026 11:14:59 +0300 Subject: [PATCH 1/5] [MOD-19284] Reuse ids on async multi-value updateVectors An async updateVectors on a multi-value label deleted every old id and appended n fresh ones, so the index grew until GC ran. The label's old ids are now reserved, one per new vector, for the new vectors' insert jobs, which are held back until their id's repairs are done and then store into it. Surplus old ids (the label shrinks) go to normal compaction, and surplus new vectors (it grows) append. deleteLabelFromHNSW takes a span of reuse jobs and nulls the entries whose submission it delegated to the repair flow. Falls back to delete-then-add when the flat buffer can't hold all n vectors. Plain addVector on a multi-value label is unchanged. Gated on reuseIdOnUpdate. Co-Authored-By: Claude Sonnet 5.5 --- src/VecSim/algorithms/hnsw/hnsw_tiered.h | 105 +++++++-- .../hnsw/hnsw_tiered_tests_friends.h | 5 + tests/unit/test_hnsw_tiered.cpp | 217 ++++++++++++++++-- 3 files changed, 293 insertions(+), 34 deletions(-) diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index ad5d8cff7..aaac32d4c 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -136,7 +136,8 @@ class TieredHNSWIndex : public VecSimTieredIndex { // Whether addVector/updateVectors overwrite an existing label's vector(s) reusing their // internal id(s) - in place in write-in-place mode, once the old id is repaired in async mode - // (single-value only) - instead of deleting them and appending fresh ones. + // (for a multi-value label, only through updateVectors) - instead of deleting them and + // appending fresh ones. // unprotected because it is only ever read or written from the main thread. bool reuseIdOnUpdate{true}; @@ -221,13 +222,15 @@ class TieredHNSWIndex : public VecSimTieredIndex { // Helper function for performing in place mark delete of vector(s) associated with a label // and creating the appropriate repair jobs for the effected connections. This should be called // while *HNSW shared lock is held* (shared locked). - // If `reuse_insert_job` is given (a single-value overwrite, not yet submitted), the deleted id - // is reserved for it and that insert job stores its vector there instead of appending. If the - // id has repairs pending, the job is then run once they're done instead of being submitted. - // Returns the number of deleted vectors and whether submission of the insert job was delegated - // to the repair flow (in which case the caller must neither submit nor touch it anymore). - std::tuple deleteLabelFromHNSW(labelType label, - HNSWInsertJob *reuse_insert_job = nullptr); + // If `reuse_insert_jobs` is given (the not yet submitted insert jobs of a label's replacement + // vectors), the label's deleted ids are reserved for them, one id per job in order, and each + // such job stores its vector there instead of appending. Ids beyond the number of jobs go + // through the ordinary compaction, and jobs beyond the number of ids append as usual. If a + // reserved id has repairs pending, its job is then run once they're done instead of being + // submitted. Returns the number of deleted vectors. Submission of every job whose id has + // repairs pending is delegated to the repair flow: its entry in `reuse_insert_jobs` is set to + // nullptr, and the caller must neither submit nor touch it anymore. + int deleteLabelFromHNSW(labelType label, std::span reuse_insert_jobs = {}); // Account for a swap job whose last repair is done. Called with `idToRepairJobsGuard` held. void onSwapJobRepaired(HNSWSwapJob *swap_job); @@ -283,6 +286,14 @@ class TieredHNSWIndex : public VecSimTieredIndex { // and all); if it's growing, the blobs beyond the label's current count are freshly appended. void updateMultiValueInPlace(labelType label, const void *new_blobs, size_t n); + // Async replacement of a multi-value label's vectors, buffering the n new blobs in the flat + // buffer as `addVector` would, and reserving the label's existing ids for their insert jobs + // (see `deleteLabelFromHNSW`) so they're reused instead of left for compaction while n fresh + // ids are appended. Returns false, having changed nothing but dropping the label's buffered + // vectors, if the flat buffer has no room for all n blobs - the caller falls back to deleting + // the label and adding the blobs one by one. + bool updateMultiValueAsync(labelType label, const void *new_blobs, size_t n); + #ifdef BUILD_TESTS #include "VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h" #endif @@ -563,23 +574,21 @@ void TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToR } template -std::tuple -TieredHNSWIndex::deleteLabelFromHNSW(labelType label, - HNSWInsertJob *reuse_insert_job) { +int TieredHNSWIndex::deleteLabelFromHNSW( + labelType label, std::span reuse_insert_jobs) { auto *hnsw_index = getHNSWIndex(); this->mainIndexGuard.lock_shared(); // Get the required data about the relevant ids to delete. // Internally, this will hold the index data lock. auto internal_ids = hnsw_index->markDelete(label); - assert((!reuse_insert_job || internal_ids.size() <= 1) && - "id reuse pairs a single deleted id with a single insert job"); - bool submission_delegated = false; for (size_t i = 0; i < internal_ids.size(); i++) { idType id = internal_ids[i]; vecsim_stl::vector repair_jobs(this->allocator); auto *swap_job = new (this->allocator) HNSWSwapJob(this->allocator, id); + HNSWInsertJob *reuse_insert_job = + i < reuse_insert_jobs.size() ? reuse_insert_jobs[i] : nullptr; if (reuse_insert_job) { // Not yet visible to any other thread: the swap job isn't published until its repair // jobs are, and the insert job isn't run before its repairs are done. @@ -628,7 +637,9 @@ TieredHNSWIndex::deleteLabelFromHNSW(labelType label, } // Capture this before publishing the repair jobs: a synchronous submit callback may run // the last repair and then the reuse insert before `deleteLabelFromHNSW` returns. - submission_delegated = reuse_insert_job && !incomingEdges.empty(); + if (reuse_insert_job && !incomingEdges.empty()) { + reuse_insert_jobs[i] = nullptr; + } this->idToRepairJobsGuard.unlock(); if (incomingEdges.size() == 0) { @@ -649,9 +660,9 @@ TieredHNSWIndex::deleteLabelFromHNSW(labelType label, } } this->mainIndexGuard.unlock_shared(); - // With no repairs to wait for, the id is ready already and the caller submits the job as + // A job whose id has no repairs to wait for is ready already, and the caller submits it as // usual. Otherwise, whichever path accounts for the last repair submits it. - return {static_cast(internal_ids.size()), submission_delegated}; + return internal_ids.size(); } template @@ -1322,9 +1333,8 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l // we still return 0 (not -1). With reuse on, the new vector takes over the old one's id // instead of appending a fresh one. HNSWInsertJob *reuse_job = this->reuseIdOnUpdate ? new_insert_job : nullptr; - auto [deleted_count, delegated] = this->deleteLabelFromHNSW(label, reuse_job); - ret = std::max(ret - deleted_count, 0); - submission_delegated = delegated; + ret = std::max(ret - this->deleteLabelFromHNSW(label, {&reuse_job, 1}), 0); + submission_delegated = this->reuseIdOnUpdate && !reuse_job; } // Apply ready swap jobs if number of deleted vectors reached the threshold (under exclusive // lock of the main index guard). @@ -1393,7 +1403,7 @@ int TieredHNSWIndex::deleteVector(labelType label) { // writeMode is not protected since it is assumed to be called only from the "main thread" // (that is the thread that is exclusively calling add/delete vector). if (this->getWriteMode() == VecSim_WriteAsync) { - num_deleted_vectors += std::get<0>(this->deleteLabelFromHNSW(label)); + num_deleted_vectors += this->deleteLabelFromHNSW(label); // Apply ready swap jobs if number of deleted vectors reached the threshold // (under exclusive lock of the main index guard). if (readySwapJobs >= this->pendingSwapJobsThreshold) { @@ -1463,6 +1473,11 @@ VecSimUpdateCode TieredHNSWIndex::updateVectors(labelType la return VecSimUpdate_OK; } + if (this->getWriteMode() == VecSim_WriteAsync && this->reuseIdOnUpdate && + !sqAccumulationState && n > 0 && this->updateMultiValueAsync(label, new_blobs, n)) { + return VecSimUpdate_OK; + } + this->deleteVector(label); const char *blob = static_cast(new_blobs); for (size_t i = 0; i < n; i++) { @@ -1471,6 +1486,54 @@ VecSimUpdateCode TieredHNSWIndex::updateVectors(labelType la return VecSimUpdate_OK; } +template +bool TieredHNSWIndex::updateMultiValueAsync(labelType label, + const void *new_blobs, size_t n) { + // Drop the label's buffered copies first, which also makes room for the new ones. + this->deleteFromFlatAndInsertJobs(label); + if (this->frontendIndex->indexSize() + n > this->flatBufferLimit) { + return false; + } + + const char *blob = static_cast(new_blobs); + vecsim_stl::vector insert_jobs(this->allocator); + insert_jobs.reserve(n); + this->flatIndexGuard.lock(); + for (size_t i = 0; i < n; i++) { + idType flat_id = this->frontendIndex->indexSize(); + this->frontendIndex->addVector(blob + i * this->frontendIndex->getInputBlobSize(), label); + auto *job = new (this->allocator) + HNSWInsertJob(this->allocator, label, flat_id, executeInsertJobWrapper, this); + insert_jobs.push_back(job); + auto [it, inserted] = this->labelToInsertJobs.try_emplace(label, 1, job, this->allocator); + if (!inserted) { + it->second.push_back(job); + } + } + this->flatIndexGuard.unlock(); + + // Mark the label's indexed vectors deleted *before* any of the new ones can be ingested, or + // they'd be marked deleted as well. The jobs whose id has repairs pending are set to nullptr + // here, as they're submitted later by the repair flow. + auto reuse_jobs = std::span(insert_jobs.data(), insert_jobs.size()); + this->deleteLabelFromHNSW(label, reuse_jobs); + if (readySwapJobs >= this->pendingSwapJobsThreshold) { + this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + } + + vecsim_stl::vector jobs_to_submit(this->allocator); + for (auto *job : insert_jobs) { + if (job) { + jobs_to_submit.push_back(job); + } + } + if (!jobs_to_submit.empty()) { + this->submitJobs(jobs_to_submit); + } + this->submitDeferredReuseInsertJobs(); + return true; +} + template void TieredHNSWIndex::updateMultiValueInPlace(labelType label, const void *new_blobs, size_t n) { diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h index 8ecddb376..ab27b24c9 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h @@ -106,5 +106,10 @@ INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_asyncOverwriteReusesIdRelocated INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_asyncOverwriteWithReuseDisabledAppends_Test) INDEX_TEST_FRIEND_CLASS( HNSWTieredIndexTestBasic_asyncOverwriteReusesSlotRelocatedAfterClearing_Test) +INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_asyncUpdateVectorsMultiReusesIds_Test) +INDEX_TEST_FRIEND_CLASS( + HNSWTieredIndexTestBasic_asyncUpdateVectorsMultiReleasesIdsWhenUpdatedAgain_Test) +INDEX_TEST_FRIEND_CLASS( + HNSWTieredIndexTestBasic_asyncUpdateVectorsMultiWithReuseDisabledAppends_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_addRelabelAndUpdateTogetherUnderLoad_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_queryWhileUpdating_Test) diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index c333371fe..da7e77b4b 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -90,6 +90,31 @@ class HNSWTieredIndexTest : public ::testing::Test { EXPECT_EQ(CastToHNSW(tiered_index)->indexSize(), n); return tiered_index; } + + // A multi-value index in async mode holding labels 0..n_labels-1 with `per_label` vectors + // each (label l's j-th vector is generated from l * 10 + j), all ingested into HNSW, with no + // queued jobs left. + TieredHNSWIndex * + CreateIngestedAsyncMultiIndex(tieredIndexMock &mock_thread_pool, size_t n_labels, + size_t per_label, size_t dim) { + HNSWParams params = {.type = index_type_t::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = true}; + VecSimParams hnsw_params = CreateParams(params); + auto *tiered_index = CreateTieredHNSWIndex(hnsw_params, mock_thread_pool); + VecSim_SetWriteMode(VecSim_WriteAsync); + for (size_t label = 0; label < n_labels; label++) { + for (size_t j = 0; j < per_label; j++) { + GenerateAndAddVector(tiered_index, dim, label, label * 10 + j); + } + } + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + EXPECT_EQ(CastToHNSW(tiered_index)->indexSize(), n_labels * per_label); + return tiered_index; + } }; TYPED_TEST_SUITE(HNSWTieredIndexTest, DataTypeSetExtended); @@ -1205,24 +1230,24 @@ TYPED_TEST(HNSWTieredIndexTest, deleteFromHNSWBasic) { auto allocator = tiered_index->getAllocator(); // Delete a non existing label. - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(0)), 0); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(0)), 0); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Insert one vector to HNSW and then delete it (it should have no neighbors to repair). GenerateAndAddVector(tiered_index->backendIndex, dim, 0); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(0)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(0)), 1); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Add another vector and remove it. Since the other vector in the index has marked deleted, // this vector should have no neighbors, and again, no neighbors to repair. GenerateAndAddVector(tiered_index->backendIndex, dim, 1, 1); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(1)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(1)), 1); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Add two vectors and delete one, expect that at backendIndex one repair job will be created. GenerateAndAddVector(tiered_index->backendIndex, dim, 2, 2); GenerateAndAddVector(tiered_index->backendIndex, dim, 3, 3); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(3)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(3)), 1); // The first job should be a repair job of the first inserted non-deleted node id (2) // in level 0. @@ -1257,7 +1282,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteFromHNSWMulti) { // Add two vectors and delete one, expect that at least one repair job will be created. GenerateAndAddVector(tiered_index->backendIndex, dim, 0, 0); GenerateAndAddVector(tiered_index->backendIndex, dim, 1, 1); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(0)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(0)), 1); ASSERT_EQ(tiered_index->idToRepairJobs.size(), 1); ASSERT_EQ(tiered_index->idToRepairJobs.at(1).size(), 1); ASSERT_EQ(tiered_index->idToRepairJobs.at(1)[0]->associatedSwapJobs.size(), 1); @@ -1274,7 +1299,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteFromHNSWMulti) { // also an outgoing edge to his other (deleted) neighbor (0), so there will be no new // repair job created for 1, since the previous repair job is expected to have both 0 and 2 in // its associated swap jobs. Also, there is an edge 0->1 whose going to be repaired as well. - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(1)), 2); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(1)), 2); ASSERT_EQ(mock_thread_pool.jobQ.size(), 2); ASSERT_EQ(((HNSWRepairJob *)(mock_thread_pool.jobQ.front().job))->node_id, 0); ASSERT_EQ(((HNSWRepairJob *)(mock_thread_pool.jobQ.front().job))->level, 0); @@ -1324,7 +1349,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteFromHNSWMultiLevels) { } while (num_elements_with_multiple_levels < 2); // Delete the last inserted vector, which is in level 1. - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(vec_id)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(vec_id)), 1); ASSERT_EQ(tiered_index->getHNSWIndex()->getGraphDataByInternalId(vec_id)->toplevel, 1); // This should be an array of length 1. auto &level_one = tiered_index->getHNSWIndex()->getElementLevelData(vec_id, 1); @@ -1371,7 +1396,7 @@ TYPED_TEST(HNSWTieredIndexTest, deleteFromHNSWWithRepairJobExec) { idType ep = tiered_index->getHNSWIndex()->safeGetEntryPointState().first; auto incoming_neighbors = tiered_index->getHNSWIndex()->safeCollectAllNodeIncomingNeighbors(ep); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(ep)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(ep)), 1); ASSERT_EQ(mock_thread_pool.jobQ.size(), incoming_neighbors.size()); ASSERT_EQ(tiered_index->getHNSWIndex()->checkIntegrity().connections_to_repair, mock_thread_pool.jobQ.size()); @@ -1437,7 +1462,7 @@ TYPED_TEST(HNSWTieredIndexTest, manageIndexOwnershipWithPendingJobs) { // Add two vectors directly to HNSW, and remove one vector to create a repair job. GenerateAndAddVector(tiered_index->backendIndex, dim, 0, 0); GenerateAndAddVector(tiered_index->backendIndex, dim, 1, 1); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(0)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(0)), 1); ASSERT_EQ(tiered_index->idToRepairJobs.size(), 1); // Delete the index before the job was executed (this would delete the pending job as well). @@ -4546,7 +4571,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteBothAsyncAndInplace) { GenerateAndAddVector(tiered_index->backendIndex, dim, 0); // Add another vector and remove it. Expect that at HNSW index one repair job will be created. GenerateAndAddVector(tiered_index->backendIndex, dim, 1, 1); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(1)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(1)), 1); ASSERT_EQ(mock_thread_pool.jobQ.size(), 1); // The first job should be a repair job of the first inserted node id (0) in level 0. @@ -4563,7 +4588,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteBothAsyncAndInplace) { // Add one more vector and remove it, expect that the same repair job for 0 would be created // for repairing 0->2. GenerateAndAddVector(tiered_index->backendIndex, dim, 2, 2); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(2)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(2)), 1); ASSERT_TRUE(tiered_index->idToSwapJob.contains(2)); ASSERT_EQ(tiered_index->idToRepairJobs.size(), 1); ASSERT_EQ(tiered_index->idToRepairJobs.at(0)[0]->associatedSwapJobs.size(), 2); @@ -4604,7 +4629,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteBothAsyncAndInplaceMulti) { GenerateAndAddVector(tiered_index->backendIndex, dim, 0); // Add another vector and remove it. Expect that at HNSW index one repair job will be created. GenerateAndAddVector(tiered_index->backendIndex, dim, 1, 1); - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(1)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(1)), 1); // Add one more vector (id=2) under label 0. GenerateAndAddVector(tiered_index->backendIndex, dim, 0, 2); @@ -4617,7 +4642,7 @@ TYPED_TEST(HNSWTieredIndexTestBasic, deleteBothAsyncAndInplaceMulti) { // Remove vector with label=3, expect that the same repair job for 0 // would be created for repairing 0->3, and a new repair jobs for 2 and 4 to repair 2->3 and // 4->3. - ASSERT_EQ(std::get<0>(tiered_index->deleteLabelFromHNSW(3)), 1); + ASSERT_EQ((tiered_index->deleteLabelFromHNSW(3)), 1); ASSERT_TRUE(tiered_index->idToSwapJob.contains(3)); ASSERT_EQ(tiered_index->idToRepairJobs.size(), 3); ASSERT_EQ(tiered_index->idToRepairJobs.at(0)[0]->associatedSwapJobs.size(), 2); @@ -6135,6 +6160,172 @@ TYPED_TEST(HNSWTieredIndexTestBasic, asyncOverwriteWithReuseDisabledAppends) { ASSERT_TRUE(hnsw_index->checkIntegrity().valid_state); } +// An async update of a multi-value label hands its existing ids over to the new vectors, one for +// one, instead of appending fresh ids and compacting the old ones. Ids beyond the new count are +// left for compaction, and vectors beyond the old count are appended. As with a single-value +// overwrite, each insert job is held until the id it reuses is repaired. +TYPED_TEST(HNSWTieredIndexTestBasic, asyncUpdateVectorsMultiReusesIds) { + size_t dim = 4; + size_t n_labels = 10; + size_t per_label = 3; + // The new vector count: same as, fewer than, and more than the label's current one. + for (size_t n_new : {per_label, size_t{1}, per_label + 2}) { + SCOPED_TRACE("n_new = " + std::to_string(n_new)); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateIngestedAsyncMultiIndex(mock_thread_pool, n_labels, per_label, dim); + auto *hnsw_index = this->CastToHNSW(tiered_index); + const size_t total_before = n_labels * per_label; + + labelType label = n_labels / 2; + auto old_ids = hnsw_index->getElementIds(label); + ASSERT_EQ(old_ids.size(), per_label); + std::sort(old_ids.begin(), old_ids.end()); + const size_t n_reused = std::min(n_new, per_label); + + std::vector new_vecs(n_new * dim); + for (size_t j = 0; j < n_new; j++) { + GenerateVector(new_vecs.data() + j * dim, dim, 1000 + j); + } + ASSERT_EQ(tiered_index->updateVectors(label, new_vecs.data(), n_new), VecSimUpdate_OK); + + // The first `n_reused` of the old ids are held for the new vectors, so only the rest are + // counted towards compaction. The insert jobs of the held ids are held back as well. + size_t n_reserved = 0; + for (idType id : old_ids) { + n_reserved += tiered_index->idToSwapJob.at(id)->reservedForReuse; + } + ASSERT_EQ(n_reserved, n_reused); + ASSERT_FALSE(tiered_index->idToRepairJobs.empty()); + ASSERT_LE(CountQueuedInsertJobs(mock_thread_pool), n_new - n_reused); + + RunAllJobs(mock_thread_pool); + auto new_ids = hnsw_index->getElementIds(label); + ASSERT_EQ(new_ids.size(), n_new); + std::sort(new_ids.begin(), new_ids.end()); + // Every reused id is one of the label's old ids, and every other old id is still deleted. + size_t n_old_ids_kept = 0; + for (idType id : new_ids) { + n_old_ids_kept += std::binary_search(old_ids.begin(), old_ids.end(), id); + } + ASSERT_EQ(n_old_ids_kept, n_reused); + ASSERT_EQ(hnsw_index->indexSize(), total_before + (n_new - n_reused)); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), per_label - n_reused); + ASSERT_EQ(tiered_index->readySwapJobs, per_label - n_reused); + ASSERT_EQ(tiered_index->frontendIndex->indexSize(), 0); + ASSERT_EQ(tiered_index->labelToInsertJobs.size(), 0); + + tiered_index->runGC(); + ASSERT_EQ(hnsw_index->indexSize(), total_before - per_label + n_new); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), 0); + ASSERT_EQ(tiered_index->idToSwapJob.size(), 0); + ASSERT_EQ(tiered_index->invalidJobs.size(), 0); + ASSERT_TRUE(hnsw_index->checkIntegrity().valid_state); + for (size_t j = 0; j < n_new; j++) { + ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(label, new_vecs.data() + j * dim), 0) + << "missing " << j; + } + // Every other label's vectors are untouched. + for (size_t other = 0; other < n_labels; other++) { + if (other == label) { + continue; + } + ASSERT_EQ(hnsw_index->getElementIds(other).size(), per_label) << "label " << other; + for (size_t j = 0; j < per_label; j++) { + TEST_DATA_T untouched[dim]; + GenerateVector(untouched, dim, other * 10 + j); + ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(other, untouched), 0) + << "label " << other << " vector " << j; + } + } + } +} + +// Updating the label again before the new vectors are ingested invalidates the insert jobs the +// old ids were held for. Every id must then be released for compaction rather than held forever. +TYPED_TEST(HNSWTieredIndexTestBasic, asyncUpdateVectorsMultiReleasesIdsWhenUpdatedAgain) { + size_t dim = 4; + size_t n_labels = 10; + size_t per_label = 3; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateIngestedAsyncMultiIndex(mock_thread_pool, n_labels, per_label, dim); + auto *hnsw_index = this->CastToHNSW(tiered_index); + + labelType label = n_labels / 2; + auto old_ids = hnsw_index->getElementIds(label); + TEST_DATA_T first[per_label * dim]; + TEST_DATA_T second[2 * dim]; + for (size_t j = 0; j < per_label; j++) { + GenerateVector(first + j * dim, dim, 1000 + j); + } + for (size_t j = 0; j < 2; j++) { + GenerateVector(second + j * dim, dim, 2000 + j); + } + ASSERT_EQ(tiered_index->updateVectors(label, first, per_label), VecSimUpdate_OK); + // The label is no longer in HNSW, so the second update has no ids to reuse. + ASSERT_EQ(tiered_index->updateVectors(label, second, 2), VecSimUpdate_OK); + for (idType id : old_ids) { + ASSERT_TRUE(tiered_index->idToSwapJob.at(id)->reservedForReuse); + } + + RunAllJobs(mock_thread_pool); + for (idType id : old_ids) { + // The invalid first insert jobs released the ids, and the second update's vectors appended. + auto it = tiered_index->idToSwapJob.find(id); + ASSERT_TRUE(it == tiered_index->idToSwapJob.end() || !it->second->reservedForReuse); + } + ASSERT_EQ(hnsw_index->indexSize(), n_labels * per_label + 2); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), per_label); + + tiered_index->runGC(); + ASSERT_EQ(hnsw_index->indexSize(), n_labels * per_label - per_label + 2); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), 0); + ASSERT_EQ(tiered_index->idToSwapJob.size(), 0); + ASSERT_EQ(tiered_index->invalidJobs.size(), 0); + ASSERT_EQ(hnsw_index->getElementIds(label).size(), 2); + for (size_t j = 0; j < 2; j++) { + ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(label, second + j * dim), 0); + } + ASSERT_TRUE(hnsw_index->checkIntegrity().valid_state); +} + +// With reuse disabled, an async update of a multi-value label reserves nothing: the new vectors +// get fresh ids and the old ones are compacted as before. +TYPED_TEST(HNSWTieredIndexTestBasic, asyncUpdateVectorsMultiWithReuseDisabledAppends) { + size_t dim = 4; + size_t n_labels = 10; + size_t per_label = 3; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateIngestedAsyncMultiIndex(mock_thread_pool, n_labels, per_label, dim); + auto *hnsw_index = this->CastToHNSW(tiered_index); + tiered_index->setReuseOnUpdate(false); + + labelType label = n_labels / 2; + auto old_ids = hnsw_index->getElementIds(label); + TEST_DATA_T new_vecs[per_label * dim]; + for (size_t j = 0; j < per_label; j++) { + GenerateVector(new_vecs + j * dim, dim, 1000 + j); + } + ASSERT_EQ(tiered_index->updateVectors(label, new_vecs, per_label), VecSimUpdate_OK); + for (idType id : old_ids) { + ASSERT_FALSE(tiered_index->idToSwapJob.at(id)->reservedForReuse); + } + + RunAllJobs(mock_thread_pool); + ASSERT_EQ(hnsw_index->indexSize(), n_labels * per_label + per_label); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), per_label); + + tiered_index->runGC(); + ASSERT_EQ(hnsw_index->indexSize(), n_labels * per_label); + ASSERT_EQ(hnsw_index->getNumMarkedDeleted(), 0); + for (size_t j = 0; j < per_label; j++) { + ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(label, new_vecs + j * dim), 0); + } + ASSERT_TRUE(hnsw_index->checkIntegrity().valid_state); +} + // updateMultiValueInPlace must preprocess each new blob the same way any other direct-to-backend // write does: the backend assumes already-normalized input for cosine (it doesn't normalize // itself), so writing a raw caller blob straight into it - skipping the frontend's From 51eefd9f4d4234ad4366d8f4b25851a40d2aeb2a Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Mon, 5 Oct 2026 09:35:22 +0300 Subject: [PATCH 2/5] [MOD-19284] Count the cost of defrag in the tiered HNSW debug info Route every executeReadySwapJobs call through defrag() and record, per index, the number of defrag runs, the swap jobs they executed and the wall time they took (including the wait for the main index guard). The values are reported in the tiered HNSW debug info as TIERED_HNSW_DEFRAG_RUNS, _SWAP_JOBS and _TIME_NS. executeReadySwapJobs now returns how many jobs it ran. Co-Authored-By: Claude Sonnet 5.5 --- src/VecSim/algorithms/hnsw/hnsw_tiered.h | 53 ++++++++++++++++++++---- src/VecSim/utils/vec_utils.cpp | 4 ++ src/VecSim/utils/vec_utils.h | 3 ++ src/VecSim/vec_sim_common.h | 3 ++ tests/unit/test_hnsw_tiered.cpp | 11 +++++ tests/unit/unit_test_utils.cpp | 23 +++++++++- 6 files changed, 88 insertions(+), 9 deletions(-) diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index aaac32d4c..93e4e0a64 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -10,6 +10,7 @@ #pragma once #include +#include #include #include #include @@ -114,6 +115,11 @@ class TieredHNSWIndex : public VecSimTieredIndex { size_t pendingSwapJobsThreshold; size_t readySwapJobs; + // Cost of defrag(), for `debugInfo()`. Statistics only, so relaxed ordering is enough. + std::atomic defragRuns{0}; + std::atomic defragSwapJobs{0}; + std::atomic defragTimeNs{0}; + // Protect the both idToRepairJobs lookup and the pending_repair_jobs_counter for the // associated swap jobs. std::mutex idToRepairJobsGuard; @@ -203,7 +209,8 @@ class TieredHNSWIndex : public VecSimTieredIndex { void invalidateRepairJobs(idType deleted_id); // Execute the ready swap jobs, run no more than 'maxSwapsToRun' jobs (run all of them for -1). - void executeReadySwapJobs(size_t maxSwapsToRun = -1); + // Returns the number of swap jobs executed. + size_t executeReadySwapJobs(size_t maxSwapsToRun = -1); // Wrappers static functions to be sent as callbacks upon creating the jobs (since members // functions cannot serve as callback, this serve as the "gateway" to the appropriate index). @@ -408,11 +415,22 @@ class TieredHNSWIndex : public VecSimTieredIndex { inline void setLastSearchMode(VecSearchMode mode) override { this->backendIndex->setLastSearchMode(mode); } + // swap latest elements deleted slots + void defrag() { + const auto start = std::chrono::steady_clock::now(); + const size_t executed = this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + // Includes the wait for the main index guard, which is part of what defrag costs a caller. + const auto elapsed = std::chrono::steady_clock::now() - start; + defragRuns.fetch_add(1, std::memory_order_relaxed); + defragSwapJobs.fetch_add(executed, std::memory_order_relaxed); + defragTimeNs.fetch_add( + std::chrono::duration_cast(elapsed).count(), + std::memory_order_relaxed); + } void runGC() override { - // Run no more than pendingSwapJobsThreshold value jobs. TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, "running asynchronous GC for tiered HNSW index"); - this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + this->defrag(); // Disposing of deleted elements can complete a reserved id's repairs. this->submitDeferredReuseInsertJobs(); } @@ -543,7 +561,7 @@ void TieredHNSWIndex::executeSwapJob(vecsim_stl::vector -void TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToRun) { +size_t TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToRun) { // Execute swap jobs - acquire hnsw write lock. const auto main_index_lock = this->acquireMainIndexGuard(); @@ -571,6 +589,7 @@ void TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsToR readySwapJobs -= idsToRemove.size(); TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, "Tiered HNSW index GC: done executing %zu swap jobs", idsToRemove.size()); + return idsToRemove.size(); } template @@ -1342,7 +1361,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l // batch of jobs for which all of its pending repair jobs were executed (otherwise finish and // return). if (readySwapJobs >= this->pendingSwapJobsThreshold) { - this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + this->defrag(); } // Insert job to the queue and signal the workers' updater - unless it's held until the id it @@ -1407,7 +1426,7 @@ int TieredHNSWIndex::deleteVector(labelType label) { // Apply ready swap jobs if number of deleted vectors reached the threshold // (under exclusive lock of the main index guard). if (readySwapJobs >= this->pendingSwapJobsThreshold) { - this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + this->defrag(); } } else { // delete in place. @@ -1518,7 +1537,7 @@ bool TieredHNSWIndex::updateMultiValueAsync(labelType label, auto reuse_jobs = std::span(insert_jobs.data(), insert_jobs.size()); this->deleteLabelFromHNSW(label, reuse_jobs); if (readySwapJobs >= this->pendingSwapJobsThreshold) { - this->executeReadySwapJobs(this->pendingSwapJobsThreshold); + this->defrag(); } vecsim_stl::vector jobs_to_submit(this->allocator); @@ -1989,7 +2008,11 @@ template VecSimIndexDebugInfo TieredHNSWIndex::debugInfo() const { auto info = VecSimTieredIndex::debugInfo(); - HnswTieredInfo hnswTieredInfo = {.pendingSwapJobsThreshold = this->pendingSwapJobsThreshold}; + HnswTieredInfo hnswTieredInfo = {.pendingSwapJobsThreshold = this->pendingSwapJobsThreshold, + .defragRuns = defragRuns.load(std::memory_order_relaxed), + .defragSwapJobs = + defragSwapJobs.load(std::memory_order_relaxed), + .defragTimeNs = defragTimeNs.load(std::memory_order_relaxed)}; info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo = hnswTieredInfo; info.tieredInfo.backgroundIndexing = @@ -2011,6 +2034,20 @@ VecSimDebugInfoIterator *TieredHNSWIndex::debugInfoIterator( .fieldValue = {FieldValue{.uintegerValue = info.tieredInfo.specificTieredBackendInfo .hnswTieredInfo.pendingSwapJobsThreshold}}}); + const auto &tieredInfo = info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo; + infoIterator->addInfoField( + VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING, + .fieldType = INFOFIELD_UINT64, + .fieldValue = {FieldValue{.uintegerValue = tieredInfo.defragRuns}}}); + infoIterator->addInfoField( + VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING, + .fieldType = INFOFIELD_UINT64, + .fieldValue = {FieldValue{.uintegerValue = tieredInfo.defragSwapJobs}}}); + infoIterator->addInfoField( + VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING, + .fieldType = INFOFIELD_UINT64, + .fieldValue = {FieldValue{.uintegerValue = tieredInfo.defragTimeNs}}}); + return infoIterator; } diff --git a/src/VecSim/utils/vec_utils.cpp b/src/VecSim/utils/vec_utils.cpp index 3459d3d4f..a79367015 100644 --- a/src/VecSim/utils/vec_utils.cpp +++ b/src/VecSim/utils/vec_utils.cpp @@ -82,6 +82,10 @@ const char *VecSimCommonStrings::BACKEND_INDEX_STRING = "BACKEND_INDEX"; // Tiered HNSW specific const char *VecSimCommonStrings::TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING = "TIERED_HNSW_SWAP_JOBS_THRESHOLD"; +const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING = "TIERED_HNSW_DEFRAG_RUNS"; +const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING = + "TIERED_HNSW_DEFRAG_SWAP_JOBS"; +const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING = "TIERED_HNSW_DEFRAG_TIME_NS"; // Tiered SVS specific const char *VecSimCommonStrings::TIERED_SVS_TRAINING_THRESHOLD_STRING = "TIERED_SVS_TRAINING_THRESHOLD"; diff --git a/src/VecSim/utils/vec_utils.h b/src/VecSim/utils/vec_utils.h index 8bcc7093a..36d64e96a 100644 --- a/src/VecSim/utils/vec_utils.h +++ b/src/VecSim/utils/vec_utils.h @@ -82,6 +82,9 @@ struct VecSimCommonStrings { static const char *BACKEND_INDEX_STRING; // tiered HNSW specific static const char *TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING; + static const char *TIERED_HNSW_DEFRAG_RUNS_STRING; + static const char *TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING; + static const char *TIERED_HNSW_DEFRAG_TIME_NS_STRING; // tiered SVS specific static const char *TIERED_SVS_TRAINING_THRESHOLD_STRING; static const char *TIERED_SVS_UPDATE_THRESHOLD_STRING; diff --git a/src/VecSim/vec_sim_common.h b/src/VecSim/vec_sim_common.h index f1ae56e2e..de2b46cbb 100644 --- a/src/VecSim/vec_sim_common.h +++ b/src/VecSim/vec_sim_common.h @@ -453,6 +453,9 @@ typedef struct { typedef struct HnswTieredInfo { size_t pendingSwapJobsThreshold; + size_t defragRuns; // Number of defrag() calls so far. + size_t defragSwapJobs; // Swap jobs those calls executed. + size_t defragTimeNs; // Wall time spent in defrag(), including the main index guard wait. } HnswTieredInfo; typedef struct SvsTieredInfo { diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index da7e77b4b..a3ddba5b1 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -2302,6 +2302,17 @@ TYPED_TEST(HNSWTieredIndexTest, invalidRepairJobOnSwap) { auto state = tiered_index->getHNSWIndex()->checkIntegrity(); EXPECT_EQ(state.valid_state, true); EXPECT_EQ(state.connections_to_repair, 0); + + // Both deleted elements were disposed of by defrag(), which the info reports. Other callers + // of defrag() may have run too, so only the swap jobs, which are exactly the two ids, are + // compared exactly. + EXPECT_GE(tiered_index->defragRuns.load(), 2); + EXPECT_EQ(tiered_index->defragSwapJobs.load(), 2); + EXPECT_GT(tiered_index->defragTimeNs.load(), 0); + auto info = tiered_index->debugInfo().tieredInfo.specificTieredBackendInfo.hnswTieredInfo; + EXPECT_EQ(info.defragRuns, tiered_index->defragRuns.load()); + EXPECT_EQ(info.defragSwapJobs, 2); + EXPECT_EQ(info.defragTimeNs, tiered_index->defragTimeNs.load()); } // A set of lambdas that determine whether a vector should be inserted to the diff --git a/tests/unit/unit_test_utils.cpp b/tests/unit/unit_test_utils.cpp index e33e83299..296e9063b 100644 --- a/tests/unit/unit_test_utils.cpp +++ b/tests/unit/unit_test_utils.cpp @@ -26,7 +26,7 @@ namespace DebugInfoIteratorFieldCount { constexpr size_t FLAT = 11; constexpr size_t HNSW = 18; constexpr size_t SVS = 25; -constexpr size_t TIERED_HNSW = 16; +constexpr size_t TIERED_HNSW = 19; constexpr size_t TIERED_SVS = 18; } // namespace DebugInfoIteratorFieldCount @@ -492,6 +492,24 @@ void compareTieredIndexInfoToIterator(VecSimIndexDebugInfo info, ASSERT_EQ( infoField->fieldValue.uintegerValue, info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.pendingSwapJobsThreshold); + } else if (!strcmp(infoField->fieldName, + VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING)) { + ASSERT_EQ(backendAlgo, VecSimAlgo_HNSWLIB); + ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); + ASSERT_EQ(infoField->fieldValue.uintegerValue, + info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.defragRuns); + } else if (!strcmp(infoField->fieldName, + VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING)) { + ASSERT_EQ(backendAlgo, VecSimAlgo_HNSWLIB); + ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); + ASSERT_EQ(infoField->fieldValue.uintegerValue, + info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.defragSwapJobs); + } else if (!strcmp(infoField->fieldName, + VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING)) { + ASSERT_EQ(backendAlgo, VecSimAlgo_HNSWLIB); + ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); + ASSERT_EQ(infoField->fieldValue.uintegerValue, + info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.defragTimeNs); // SVS specific fields } else if (!strcmp(infoField->fieldName, VecSimCommonStrings::TIERED_SVS_TRAINING_THRESHOLD_STRING)) { @@ -846,6 +864,9 @@ std::vector getTieredHNSWFields() { fields.push_back( VecSimCommonStrings:: TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING); // 15. TIERED_HNSW_SWAP_JOBS_THRESHOLD + fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING); // 16. + fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING); // 17. + fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING); // 18. fields.push_back(VecSimCommonStrings::SHARED_MEMORY_STRING); return fields; } From e7455055f09b5dce8203e386888aebb6e2a97361 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Mon, 5 Oct 2026 10:54:14 +0300 Subject: [PATCH 3/5] [MOD-19284] Count defrag runs and time for tiered SVS; log HNSW swap jobs Drop the swap-jobs counter from the tiered HNSW info and log the number of swap jobs each defrag executes at debug level instead. Count backend GC runs and their time in the tiered SVS index as well, reported as TIERED_SVS_DEFRAG_RUNS and TIERED_SVS_DEFRAG_TIME_NS. The time includes the wait for the main index guard, as for HNSW. Co-Authored-By: Claude Sonnet 5.5 --- src/VecSim/algorithms/hnsw/hnsw_tiered.h | 30 ++++++++++----------- src/VecSim/algorithms/svs/svs_tiered.h | 34 ++++++++++++++++++++++++ src/VecSim/utils/vec_utils.cpp | 4 +-- src/VecSim/utils/vec_utils.h | 3 ++- src/VecSim/vec_sim_common.h | 7 ++--- tests/unit/test_hnsw_tiered.cpp | 7 ++--- tests/unit/test_svs_tiered.cpp | 11 ++++++++ tests/unit/unit_test_utils.cpp | 29 ++++++++++++-------- 8 files changed, 88 insertions(+), 37 deletions(-) diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index 93e4e0a64..91429ccb4 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -117,7 +117,6 @@ class TieredHNSWIndex : public VecSimTieredIndex { // Cost of defrag(), for `debugInfo()`. Statistics only, so relaxed ordering is enough. std::atomic defragRuns{0}; - std::atomic defragSwapJobs{0}; std::atomic defragTimeNs{0}; // Protect the both idToRepairJobs lookup and the pending_repair_jobs_counter for the @@ -399,6 +398,7 @@ class TieredHNSWIndex : public VecSimTieredIndex { size_t indexCapacity() const override; int updateVectorInPlace(labelType label, HNSWIndex *hnsw_index, const MemoryUtils::unique_blob &storage_blob); + auto shouldDefrag(); double getDistanceFrom_Unsafe(labelType label, const void *blob) const override; // Do nothing here, each tier (flat buffer and HNSW) should increase capacity for itself when // needed. @@ -420,12 +420,14 @@ class TieredHNSWIndex : public VecSimTieredIndex { const auto start = std::chrono::steady_clock::now(); const size_t executed = this->executeReadySwapJobs(this->pendingSwapJobsThreshold); // Includes the wait for the main index guard, which is part of what defrag costs a caller. - const auto elapsed = std::chrono::steady_clock::now() - start; + const auto elapsedNs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); defragRuns.fetch_add(1, std::memory_order_relaxed); - defragSwapJobs.fetch_add(executed, std::memory_order_relaxed); - defragTimeNs.fetch_add( - std::chrono::duration_cast(elapsed).count(), - std::memory_order_relaxed); + defragTimeNs.fetch_add(elapsedNs, std::memory_order_relaxed); + TIERED_LOG(VecSimCommonStrings::LOG_DEBUG_STRING, + "Tiered HNSW index defrag: executed %zu swap jobs in %lld ns", executed, + static_cast(elapsedNs)); } void runGC() override { TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, @@ -1256,6 +1258,10 @@ int TieredHNSWIndex::updateVectorInPlace( int ret = (flat_deleted == 0) ? backend_ret : 0; return ret; } +template +auto TieredHNSWIndex::shouldDefrag() { + return readySwapJobs >= this->pendingSwapJobsThreshold; +} // In the tiered index, we assume that the blobs are processed by the flat buffer // before being transferred to the HNSW index. // When inserting vectors directly into the HNSW index—such as in VecSim_WriteInPlace mode— or when @@ -1360,7 +1366,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l // If swapJobs size is equal or larger than a threshold, go over the swap jobs and execute a // batch of jobs for which all of its pending repair jobs were executed (otherwise finish and // return). - if (readySwapJobs >= this->pendingSwapJobsThreshold) { + if (shouldDefrag()) { this->defrag(); } @@ -1425,7 +1431,7 @@ int TieredHNSWIndex::deleteVector(labelType label) { num_deleted_vectors += this->deleteLabelFromHNSW(label); // Apply ready swap jobs if number of deleted vectors reached the threshold // (under exclusive lock of the main index guard). - if (readySwapJobs >= this->pendingSwapJobsThreshold) { + if (shouldDefrag()) { this->defrag(); } } else { @@ -1536,7 +1542,7 @@ bool TieredHNSWIndex::updateMultiValueAsync(labelType label, // here, as they're submitted later by the repair flow. auto reuse_jobs = std::span(insert_jobs.data(), insert_jobs.size()); this->deleteLabelFromHNSW(label, reuse_jobs); - if (readySwapJobs >= this->pendingSwapJobsThreshold) { + if (shouldDefrag()) { this->defrag(); } @@ -2010,8 +2016,6 @@ VecSimIndexDebugInfo TieredHNSWIndex::debugInfo() const { HnswTieredInfo hnswTieredInfo = {.pendingSwapJobsThreshold = this->pendingSwapJobsThreshold, .defragRuns = defragRuns.load(std::memory_order_relaxed), - .defragSwapJobs = - defragSwapJobs.load(std::memory_order_relaxed), .defragTimeNs = defragTimeNs.load(std::memory_order_relaxed)}; info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo = hnswTieredInfo; @@ -2039,10 +2043,6 @@ VecSimDebugInfoIterator *TieredHNSWIndex::debugInfoIterator( VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING, .fieldType = INFOFIELD_UINT64, .fieldValue = {FieldValue{.uintegerValue = tieredInfo.defragRuns}}}); - infoIterator->addInfoField( - VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING, - .fieldType = INFOFIELD_UINT64, - .fieldValue = {FieldValue{.uintegerValue = tieredInfo.defragSwapJobs}}}); infoIterator->addInfoField( VecSim_InfoField{.fieldName = VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING, .fieldType = INFOFIELD_UINT64, diff --git a/src/VecSim/algorithms/svs/svs_tiered.h b/src/VecSim/algorithms/svs/svs_tiered.h index 32220358f..e83bccf60 100644 --- a/src/VecSim/algorithms/svs/svs_tiered.h +++ b/src/VecSim/algorithms/svs/svs_tiered.h @@ -250,6 +250,9 @@ class TieredSVSIndex : public VecSimTieredIndex { 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; + // Cost of the backend GC, for `debugInfo()`. Statistics only, so relaxed ordering is enough. + std::atomic defragRuns{0}; + std::atomic defragTimeNs{0}; // 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; @@ -594,6 +597,7 @@ class TieredSVSIndex : public VecSimTieredIndex { auto index = static_cast *>(idx); assert(index); + const auto start = std::chrono::steady_clock::now(); std::lock_guard lock{index->mainIndexGuard}; // Release the scheduled flag to allow scheduling again index->indexGCScheduled.clear(); @@ -610,6 +614,20 @@ class TieredSVSIndex : public VecSimTieredIndex { svs_index->setParallelism(std::min(availableThreads, index->backendIndex->indexSize())); // VecSimIndexAbstract::runGC() is protected static_cast(index->backendIndex)->runGC(); + index->recordDefrag(start); + } + + // Account for one backend GC run that started at `start`. The time includes the wait for the + // main index guard, which is part of what GC costs, as in the tiered HNSW defrag(). + void recordDefrag(std::chrono::steady_clock::time_point start) { + const auto elapsedNs = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + defragRuns.fetch_add(1, std::memory_order_relaxed); + defragTimeNs.fetch_add(elapsedNs, std::memory_order_relaxed); + this->backendIndex->log(VecSimCommonStrings::LOG_DEBUG_STRING, + "Tiered SVS index defrag took %lld ns", + static_cast(elapsedNs)); } #ifdef BUILD_TESTS @@ -1116,6 +1134,8 @@ class TieredSVSIndex : public VecSimTieredIndex { .trainingTriggerThreshold = this->trainingTriggerThreshold, .updateTriggerThreshold = this->updateTriggerThreshold, .updateJobWaitTime = this->updateJobWaitTime, + .defragRuns = defragRuns.load(std::memory_order_relaxed), + .defragTimeNs = defragTimeNs.load(std::memory_order_relaxed), }; { // Use try_lock to avoid blocking the main thread during long-running @@ -1172,6 +1192,18 @@ class TieredSVSIndex : public VecSimTieredIndex { .fieldValue = {FieldValue{ .uintegerValue = info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.updateJobWaitTime}}}); + infoIterator->addInfoField(VecSim_InfoField{ + .fieldName = VecSimCommonStrings::TIERED_SVS_DEFRAG_RUNS_STRING, + .fieldType = INFOFIELD_UINT64, + .fieldValue = {FieldValue{ + .uintegerValue = + info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.defragRuns}}}); + infoIterator->addInfoField(VecSim_InfoField{ + .fieldName = VecSimCommonStrings::TIERED_SVS_DEFRAG_TIME_NS_STRING, + .fieldType = INFOFIELD_UINT64, + .fieldValue = {FieldValue{ + .uintegerValue = + info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.defragTimeNs}}}); return infoIterator; } @@ -1191,6 +1223,7 @@ class TieredSVSIndex : public VecSimTieredIndex { TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, "running synchronous GC for tiered SVS index in write-in-place mode"); // In write-in-place mode, we run GC synchronously. + const auto start = std::chrono::steady_clock::now(); std::lock_guard lock{this->mainIndexGuard}; if (this->backendIndex->indexSize() == 0) { // No need to run GC on an empty index. @@ -1200,6 +1233,7 @@ class TieredSVSIndex : public VecSimTieredIndex { this->GetSVSIndex()->setParallelism(1); // VecSimIndexAbstract::runGC() is protected static_cast(this->backendIndex)->runGC(); + this->recordDefrag(start); return; } TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, diff --git a/src/VecSim/utils/vec_utils.cpp b/src/VecSim/utils/vec_utils.cpp index a79367015..bb3e23cde 100644 --- a/src/VecSim/utils/vec_utils.cpp +++ b/src/VecSim/utils/vec_utils.cpp @@ -83,13 +83,13 @@ const char *VecSimCommonStrings::BACKEND_INDEX_STRING = "BACKEND_INDEX"; const char *VecSimCommonStrings::TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING = "TIERED_HNSW_SWAP_JOBS_THRESHOLD"; const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING = "TIERED_HNSW_DEFRAG_RUNS"; -const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING = - "TIERED_HNSW_DEFRAG_SWAP_JOBS"; const char *VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING = "TIERED_HNSW_DEFRAG_TIME_NS"; // Tiered SVS specific const char *VecSimCommonStrings::TIERED_SVS_TRAINING_THRESHOLD_STRING = "TIERED_SVS_TRAINING_THRESHOLD"; const char *VecSimCommonStrings::TIERED_SVS_UPDATE_THRESHOLD_STRING = "TIERED_SVS_UPDATE_THRESHOLD"; +const char *VecSimCommonStrings::TIERED_SVS_DEFRAG_RUNS_STRING = "TIERED_SVS_DEFRAG_RUNS"; +const char *VecSimCommonStrings::TIERED_SVS_DEFRAG_TIME_NS_STRING = "TIERED_SVS_DEFRAG_TIME_NS"; const char *VecSimCommonStrings::TIERED_SVS_THREADS_RESERVE_TIMEOUT_STRING = "TIERED_SVS_THREADS_RESERVE_TIMEOUT"; diff --git a/src/VecSim/utils/vec_utils.h b/src/VecSim/utils/vec_utils.h index 36d64e96a..25a35208e 100644 --- a/src/VecSim/utils/vec_utils.h +++ b/src/VecSim/utils/vec_utils.h @@ -83,10 +83,11 @@ struct VecSimCommonStrings { // tiered HNSW specific static const char *TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING; static const char *TIERED_HNSW_DEFRAG_RUNS_STRING; - static const char *TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING; static const char *TIERED_HNSW_DEFRAG_TIME_NS_STRING; // tiered SVS specific static const char *TIERED_SVS_TRAINING_THRESHOLD_STRING; + static const char *TIERED_SVS_DEFRAG_RUNS_STRING; + static const char *TIERED_SVS_DEFRAG_TIME_NS_STRING; static const char *TIERED_SVS_UPDATE_THRESHOLD_STRING; static const char *TIERED_SVS_THREADS_RESERVE_TIMEOUT_STRING; diff --git a/src/VecSim/vec_sim_common.h b/src/VecSim/vec_sim_common.h index de2b46cbb..7d0676fb8 100644 --- a/src/VecSim/vec_sim_common.h +++ b/src/VecSim/vec_sim_common.h @@ -453,9 +453,8 @@ typedef struct { typedef struct HnswTieredInfo { size_t pendingSwapJobsThreshold; - size_t defragRuns; // Number of defrag() calls so far. - size_t defragSwapJobs; // Swap jobs those calls executed. - size_t defragTimeNs; // Wall time spent in defrag(), including the main index guard wait. + size_t defragRuns; // Number of defrag() calls so far. + size_t defragTimeNs; // Wall time spent in defrag(), including the main index guard wait. } HnswTieredInfo; typedef struct SvsTieredInfo { @@ -464,6 +463,8 @@ typedef struct SvsTieredInfo { size_t updateJobWaitTime; // The time (microseconds) to wait for Redis threads reservation // before executing the scheduled SVS Index update job. bool indexUpdateScheduled; + size_t defragRuns; // Number of backend GC runs so far. + size_t defragTimeNs; // Wall time spent in them, including the main index guard wait. } SvsTieredInfo; typedef struct { diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index a3ddba5b1..539f239d2 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -2303,15 +2303,12 @@ TYPED_TEST(HNSWTieredIndexTest, invalidRepairJobOnSwap) { EXPECT_EQ(state.valid_state, true); EXPECT_EQ(state.connections_to_repair, 0); - // Both deleted elements were disposed of by defrag(), which the info reports. Other callers - // of defrag() may have run too, so only the swap jobs, which are exactly the two ids, are - // compared exactly. + // Both GC rounds went through defrag(), which the info reports. Other callers of defrag() may + // have run too, hence the lower bound on the count. EXPECT_GE(tiered_index->defragRuns.load(), 2); - EXPECT_EQ(tiered_index->defragSwapJobs.load(), 2); EXPECT_GT(tiered_index->defragTimeNs.load(), 0); auto info = tiered_index->debugInfo().tieredInfo.specificTieredBackendInfo.hnswTieredInfo; EXPECT_EQ(info.defragRuns, tiered_index->defragRuns.load()); - EXPECT_EQ(info.defragSwapJobs, 2); EXPECT_EQ(info.defragTimeNs, tiered_index->defragTimeNs.load()); } diff --git a/tests/unit/test_svs_tiered.cpp b/tests/unit/test_svs_tiered.cpp index c4b2d712f..4f765453e 100644 --- a/tests/unit/test_svs_tiered.cpp +++ b/tests/unit/test_svs_tiered.cpp @@ -4031,6 +4031,11 @@ TYPED_TEST(SVSTieredIndexTestBasic, runGCAPI) { ASSERT_LT(size_after_gc, size_before_gc); ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0); EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 0); + + // The two GC API calls scheduled a single backend GC, which the info accounts for. + auto svs_info = tiered_index->debugInfo().tieredInfo.specificTieredBackendInfo.svsTieredInfo; + EXPECT_EQ(svs_info.defragRuns, 1); + EXPECT_GT(svs_info.defragTimeNs, 0); } TYPED_TEST(SVSTieredIndexTestBasic, runGCParallel) { @@ -4169,6 +4174,12 @@ TYPED_TEST(SVSTieredIndexTestBasic, runGCInPlaceMode) { ASSERT_EQ(tiered_index->GetSVSIndex()->indexStorageSize(), backend_size); ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0); EXPECT_EQ(tiered_index->statisticInfo().numberOfMarkedDeleted, 0); + + // One backend GC for every 64th iteration of the loop plus the final one. The GC on the empty + // index at the start returns early and is not counted. + auto svs_info = tiered_index->debugInfo().tieredInfo.specificTieredBackendInfo.svsTieredInfo; + EXPECT_EQ(svs_info.defragRuns, threshold / 64 + 1); + EXPECT_GT(svs_info.defragTimeNs, 0); } TYPED_TEST(SVSTieredIndexTestBasic, switchDeleteModes) { diff --git a/tests/unit/unit_test_utils.cpp b/tests/unit/unit_test_utils.cpp index 296e9063b..87085dabc 100644 --- a/tests/unit/unit_test_utils.cpp +++ b/tests/unit/unit_test_utils.cpp @@ -26,8 +26,8 @@ namespace DebugInfoIteratorFieldCount { constexpr size_t FLAT = 11; constexpr size_t HNSW = 18; constexpr size_t SVS = 25; -constexpr size_t TIERED_HNSW = 19; -constexpr size_t TIERED_SVS = 18; +constexpr size_t TIERED_HNSW = 18; +constexpr size_t TIERED_SVS = 20; } // namespace DebugInfoIteratorFieldCount static void chooseCompareIndexInfoToIterator(VecSimIndexDebugInfo info, @@ -498,12 +498,6 @@ void compareTieredIndexInfoToIterator(VecSimIndexDebugInfo info, ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); ASSERT_EQ(infoField->fieldValue.uintegerValue, info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.defragRuns); - } else if (!strcmp(infoField->fieldName, - VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING)) { - ASSERT_EQ(backendAlgo, VecSimAlgo_HNSWLIB); - ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); - ASSERT_EQ(infoField->fieldValue.uintegerValue, - info.tieredInfo.specificTieredBackendInfo.hnswTieredInfo.defragSwapJobs); } else if (!strcmp(infoField->fieldName, VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING)) { ASSERT_EQ(backendAlgo, VecSimAlgo_HNSWLIB); @@ -531,6 +525,18 @@ void compareTieredIndexInfoToIterator(VecSimIndexDebugInfo info, ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); ASSERT_EQ(infoField->fieldValue.uintegerValue, info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.updateJobWaitTime); + } else if (!strcmp(infoField->fieldName, + VecSimCommonStrings::TIERED_SVS_DEFRAG_RUNS_STRING)) { + ASSERT_EQ(backendAlgo, VecSimAlgo_SVS); + ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); + ASSERT_EQ(infoField->fieldValue.uintegerValue, + info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.defragRuns); + } else if (!strcmp(infoField->fieldName, + VecSimCommonStrings::TIERED_SVS_DEFRAG_TIME_NS_STRING)) { + ASSERT_EQ(backendAlgo, VecSimAlgo_SVS); + ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); + ASSERT_EQ(infoField->fieldValue.uintegerValue, + info.tieredInfo.specificTieredBackendInfo.svsTieredInfo.defragTimeNs); } else if (!strcmp(infoField->fieldName, VecSimCommonStrings::SHARED_MEMORY_STRING)) { // Process-wide shared allocation appended by VecSimIndex_DebugInfoIterator. ASSERT_EQ(infoField->fieldType, INFOFIELD_UINT64); @@ -853,6 +859,8 @@ std::vector getTieredSVSFields() { fields.push_back(VecSimCommonStrings::TIERED_SVS_TRAINING_THRESHOLD_STRING); fields.push_back(VecSimCommonStrings::TIERED_SVS_UPDATE_THRESHOLD_STRING); fields.push_back(VecSimCommonStrings::TIERED_SVS_THREADS_RESERVE_TIMEOUT_STRING); + fields.push_back(VecSimCommonStrings::TIERED_SVS_DEFRAG_RUNS_STRING); + fields.push_back(VecSimCommonStrings::TIERED_SVS_DEFRAG_TIME_NS_STRING); fields.push_back(VecSimCommonStrings::SHARED_MEMORY_STRING); return fields; } @@ -864,9 +872,8 @@ std::vector getTieredHNSWFields() { fields.push_back( VecSimCommonStrings:: TIERED_HNSW_SWAP_JOBS_THRESHOLD_STRING); // 15. TIERED_HNSW_SWAP_JOBS_THRESHOLD - fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING); // 16. - fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_SWAP_JOBS_STRING); // 17. - fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING); // 18. + fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_RUNS_STRING); // 16. + fields.push_back(VecSimCommonStrings::TIERED_HNSW_DEFRAG_TIME_NS_STRING); // 17. fields.push_back(VecSimCommonStrings::SHARED_MEMORY_STRING); return fields; } From afe1d6de1b57bb8109b6380e8bf5004944ac3e20 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Mon, 5 Oct 2026 11:26:52 +0300 Subject: [PATCH 4/5] [MOD-19284] Expect the defrag debug log in testlogTieredIndex defrag() now logs a debug-level summary after each run, so the tiered index log in this test has one more entry. Its elapsed time varies, so only the fixed part of the message is compared. Co-Authored-By: Claude Sonnet 5.5 --- tests/unit/test_common.cpp | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_common.cpp b/tests/unit/test_common.cpp index 724c0f74f..8f93f1125 100644 --- a/tests/unit/test_common.cpp +++ b/tests/unit/test_common.cpp @@ -629,7 +629,7 @@ TEST(CommonAPITest, testlogTieredIndex) { } return buffer; }; - ASSERT_EQ(log.logBuffer.size(), 8) << buffer_as_string(); + ASSERT_EQ(log.logBuffer.size(), 9) << buffer_as_string(); size_t log_iter = 0; ASSERT_EQ(log.logBuffer[log_iter++], "verbose: " + log.prefix + "Resizing FLAT index from 0 to 1024") @@ -666,6 +666,14 @@ TEST(CommonAPITest, testlogTieredIndex) { "verbose: " + log.prefix + "Tiered HNSW index GC: done executing 1 swap jobs") << "failed at log index:" << log_iter - 1 << std::endl << "expected log: " << buffer_as_string(); + + // defrag() summarizes the run at debug level. The elapsed time varies, so only the fixed part + // of the message is compared. + const std::string defrag_summary = + "debug: " + log.prefix + "Tiered HNSW index defrag: executed 1 swap jobs in "; + ASSERT_EQ(log.logBuffer[log_iter].rfind(defrag_summary, 0), 0) + << "failed at log index:" << log_iter << std::endl + << "expected log: " << buffer_as_string(); } TEST(CommonAPITest, NormalizeBfloat16) { From b0c52fccd3e684b11cc720104c58c20b1eb91520 Mon Sep 17 00:00:00 2001 From: sharonrosenfeld Date: Tue, 6 Oct 2026 12:46:43 +0300 Subject: [PATCH 5/5] [MOD-19284] Address review: lock scope, try_emplace once, atomic readySwapJobs - updateMultiValueAsync: scoped lock_guard on flatIndexGuard that also covers the flat buffer capacity check; look up the label's insert-job vector once before the staging loop. - readySwapJobs is atomic: shouldDefrag() reads it on the writer thread while workers update it. Co-Authored-By: Claude Sonnet 5.5 --- src/VecSim/algorithms/hnsw/hnsw_tiered.h | 36 +++++++++++++----------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index 91429ccb4..21ebaa342 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -113,7 +113,7 @@ class TieredHNSWIndex : public VecSimTieredIndex { // vectors reached this limit, we apply swap jobs *only for vectors that has no more pending // repair jobs*, and are ready to be removed from the graph. size_t pendingSwapJobsThreshold; - size_t readySwapJobs; + std::atomic_size_t readySwapJobs; // Cost of defrag(), for `debugInfo()`. Statistics only, so relaxed ordering is enough. std::atomic defragRuns{0}; @@ -569,7 +569,7 @@ size_t TieredHNSWIndex::executeReadySwapJobs(size_t maxJobsT const auto main_index_lock = this->acquireMainIndexGuard(); TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, "Tiered HNSW index GC: there are %zu ready swap jobs. Start executing %zu swap jobs", - readySwapJobs, std::min(readySwapJobs, maxJobsToRun)); + readySwapJobs.load(), std::min(readySwapJobs.load(), maxJobsToRun)); vecsim_stl::vector idsToRemove(this->allocator); idsToRemove.reserve(idToSwapJob.size()); @@ -1260,7 +1260,7 @@ int TieredHNSWIndex::updateVectorInPlace( } template auto TieredHNSWIndex::shouldDefrag() { - return readySwapJobs >= this->pendingSwapJobsThreshold; + return readySwapJobs.load() >= this->pendingSwapJobsThreshold; } // In the tiered index, we assume that the blobs are processed by the flat buffer // before being transferred to the HNSW index. @@ -1516,26 +1516,28 @@ bool TieredHNSWIndex::updateMultiValueAsync(labelType label, const void *new_blobs, size_t n) { // Drop the label's buffered copies first, which also makes room for the new ones. this->deleteFromFlatAndInsertJobs(label); - if (this->frontendIndex->indexSize() + n > this->flatBufferLimit) { - return false; - } const char *blob = static_cast(new_blobs); vecsim_stl::vector insert_jobs(this->allocator); insert_jobs.reserve(n); - this->flatIndexGuard.lock(); - for (size_t i = 0; i < n; i++) { - idType flat_id = this->frontendIndex->indexSize(); - this->frontendIndex->addVector(blob + i * this->frontendIndex->getInputBlobSize(), label); - auto *job = new (this->allocator) - HNSWInsertJob(this->allocator, label, flat_id, executeInsertJobWrapper, this); - insert_jobs.push_back(job); - auto [it, inserted] = this->labelToInsertJobs.try_emplace(label, 1, job, this->allocator); - if (!inserted) { - it->second.push_back(job); + { + std::lock_guard flat_lock(this->flatIndexGuard); + if (this->frontendIndex->indexSize() + n > this->flatBufferLimit) { + return false; + } + auto &label_jobs = + this->labelToInsertJobs.try_emplace(label, this->allocator).first->second; + label_jobs.reserve(label_jobs.size() + n); + for (size_t i = 0; i < n; i++) { + idType flat_id = this->frontendIndex->indexSize(); + this->frontendIndex->addVector(blob + i * this->frontendIndex->getInputBlobSize(), + label); + auto *job = new (this->allocator) + HNSWInsertJob(this->allocator, label, flat_id, executeInsertJobWrapper, this); + insert_jobs.push_back(job); + label_jobs.push_back(job); } } - this->flatIndexGuard.unlock(); // Mark the label's indexed vectors deleted *before* any of the new ones can be ingested, or // they'd be marked deleted as well. The jobs whose id has repairs pending are set to nullptr