diff --git a/src/VecSim/algorithms/svs/svs.h b/src/VecSim/algorithms/svs/svs.h index 7dc15dc0d..9cf7d494f 100644 --- a/src/VecSim/algorithms/svs/svs.h +++ b/src/VecSim/algorithms/svs/svs.h @@ -37,16 +37,22 @@ struct SVSIndexBase { SVSIndexBase() : num_marked_deleted{0} {}; virtual ~SVSIndexBase() = default; - + virtual int addVector(const void *vector_data, labelType label) = 0; virtual int addVectors(const void *vectors_data, const labelType *labels, size_t n) = 0; + virtual int deleteVector(labelType label) = 0; virtual int deleteVectors(const labelType *labels, size_t n) = 0; + virtual void consolidate(const std::vector& labels) = 0; virtual bool isLabelExists(labelType label) const = 0; virtual size_t indexStorageSize() const = 0; virtual size_t getParallelism() const = 0; virtual void setParallelism(size_t parallelism) = 0; virtual size_t getPoolSize() const = 0; virtual bool isCompressed() const = 0; - size_t getNumMarkedDeleted() const { return num_marked_deleted; } + virtual bool ready() const = 0; + + size_t getNumMarkedDeleted() const { + return num_marked_deleted.load(std::memory_order_relaxed); + } // Abstract handler to manage SVS implementation instance // declared to avoid unsafe unique_ptr usage @@ -63,7 +69,7 @@ struct SVSIndexBase protected: // Index marked deleted vectors counter to initiate reindexing if it exceeds threshold // markIndexUpdate() manages this counter - size_t num_marked_deleted; + std::atomic num_marked_deleted; }; /** Thread Management Strategy: @@ -87,10 +93,12 @@ class SVSIndex : public VecSimIndexAbstract, fl using graph_builder_t = SVSGraphBuilder; using graph_type = typename graph_builder_t::graph_type; + // The concurrent index mirrors the upstream template signature and member surface, so + // the only change needed here is the namespace. using impl_type = std::conditional_t< isMulti, - svs::index::vamana::MultiMutableVamanaIndex, - svs::index::vamana::MutableVamanaIndex>; + svs::concurrent::MultiMutableVamanaIndex, + svs::concurrent::MutableVamanaIndex>; bool forcePreprocessing; @@ -114,6 +122,17 @@ class SVSIndex : public VecSimIndexAbstract, fl svs::logging::logger_ptr logger_; // SVS Index implementation instance std::unique_ptr impl_; + mutable std::shared_mutex pimplGuard_; + + std::atomic impl_ready_{false}; + + void setReady() { + this->impl_ready_.store(true, std::memory_order_release); + } + + void setUnready() { + this->impl_ready_.store(false, std::memory_order_release); + } static double toVecSimDistance(float v) { return svs_details::toVecSimDistance(v); } @@ -241,7 +260,7 @@ class SVSIndex : public VecSimIndexAbstract, fl } void setImpl(std::unique_ptr handler) override { - if (impl_ != nullptr) { + if (ready()) { throw std::logic_error("SVSIndex::setImpl called on non-empty impl_"); } @@ -249,7 +268,11 @@ class SVSIndex : public VecSimIndexAbstract, fl if (!svs_handler) { throw std::logic_error("Failed to cast to SVSImplHandler"); } - this->impl_ = std::move(svs_handler->impl); + { + std::lock_guard lock(this->pimplGuard_); + this->impl_ = std::move(svs_handler->impl); + if (this->impl_) setReady(); + } } // Assuming parallelism was updated to reflect the number of available threads before this @@ -264,11 +287,6 @@ class SVSIndex : public VecSimIndexAbstract, fl } int deleted_num = 0; - if constexpr (!isMulti) { - // SVS index does not support overriding vectors with the same label - // so we have to delete them first if needed - deleted_num = deleteVectorsImpl(labels, n); - } std::span ids(labels, n); auto processed_blob = this->preprocessForBatchStorage(vectors_data, n); @@ -276,23 +294,54 @@ class SVSIndex : public VecSimIndexAbstract, fl // Wrap data into SVS SimpleDataView for SVS API auto points = svs::data::SimpleDataView{typed_vectors_data, n, this->dim}; - if (!impl_) { + if constexpr (!isMulti) { + if (ready()) { + // SVS index does not support overriding vectors with the same label + // so we have to delete them first if needed + deleted_num = deleteVectorsImpl(labels, n); + } + } + + this->pimplGuard_.lock_shared(); + if (!ready()) { + this->pimplGuard_.unlock_shared(); // SVS index instance cannot be empty, so we have to construct it at first rows - impl_ = initImpl(points, ids); + std::lock_guard lock(this->pimplGuard_); + if (!ready()) { + impl_ = initImpl(points, ids); + assert(this->impl_ != nullptr); + setReady(); + } else { + impl_->add_points(points, ids, /*reuse_empty*/ false); + } } else { // Add new points to existing SVS index - impl_->add_points(points, ids); + impl_->add_points(points, ids, /*reuse_empty*/ false); + this->pimplGuard_.unlock_shared(); } return n - deleted_num; } + void consolidate(const std::vector& labels) { + std::shared_lock lock(this->pimplGuard_); + if (!ready()) + return; + + size_t n_consolidated = impl_->consolidate(labels); + num_marked_deleted.fetch_sub(n_consolidated, std::memory_order_relaxed); + } + int deleteVectorImpl(const labelType label) { - if (indexLabelCount() == 0 || !impl_->has_id(label)) { + if (indexLabelCount() == 0) { return 0; } - const auto deleted_num = impl_->delete_entries(std::span{&label, 1}); + int deleted_num = 0; + { + std::shared_lock lock(this->pimplGuard_); + deleted_num = impl_->delete_entries(std::span{&label, 1}); + } this->markIndexUpdate(deleted_num); return deleted_num; @@ -303,38 +352,37 @@ class SVSIndex : public VecSimIndexAbstract, fl return 0; } - // SVS fails if we try to delete non-existing entries - std::vector entries_to_delete; - entries_to_delete.reserve(n); - for (size_t i = 0; i < n; i++) { - if (impl_->has_id(labels[i])) { - entries_to_delete.push_back(labels[i]); - } - } - - if (entries_to_delete.size() == 0) { - return 0; + int deleted_num = 0; + { + std::shared_lock lock(this->pimplGuard_); + deleted_num = impl_->delete_entries(std::span{labels, n}); } - const auto deleted_num = impl_->delete_entries(entries_to_delete); + if (deleted_num > 0) + this->markIndexUpdate(deleted_num); - this->markIndexUpdate(deleted_num); return deleted_num; } // Count deletions and consolidate index if needed void markIndexUpdate(size_t n = 1) { - if (!impl_) + if (!ready()) return; // SVS index instance should not be empty if (indexLabelCount() == 0) { - this->impl_.reset(); - num_marked_deleted = 0; - return; + std::lock_guard lock(this->pimplGuard_); + if (indexLabelCountUnsafe() == 0) { + { + setUnready(); + this->impl_.reset(); + } + num_marked_deleted.store(0, std::memory_order_relaxed); + return; + } } - num_marked_deleted += n; + num_marked_deleted.fetch_add(n, std::memory_order_relaxed); } bool isTwoLevelLVQ(const VecSimSvsQuantBits &qbits) { @@ -373,25 +421,68 @@ class SVSIndex : public VecSimIndexAbstract, fl ~SVSIndex() = default; + bool ready() const override { + return this->impl_ready_.load(std::memory_order_acquire); + } + size_t indexSize() const override { return indexStorageSize(); } - size_t indexStorageSize() const override { return impl_ ? impl_->view_data().size() : 0; } + size_t indexStorageSize() const override { + std::shared_lock lock(this->pimplGuard_); + if (ready()) { + return impl_->view_data().size(); + } else { + return 0; + } + } size_t indexCapacity() const override { - return impl_ ? storage_traits_t::storage_capacity(impl_->view_data()) : 0; + std::shared_lock lock(this->pimplGuard_); + if (ready()) { + return storage_traits_t::storage_capacity(impl_->view_data()); + } else { + return 0; + } + } + + size_t indexLabelCountUnsafe() const { + if constexpr (isMulti) { + if (ready()) { + return impl_->labelcount(); + } else { + return 0; + } + } else { + if (ready()) { + return impl_->size(); + } else { + return 0; + } + } } size_t indexLabelCount() const override { if constexpr (isMulti) { - return impl_ ? impl_->labelcount() : 0; + if (ready()) { + std::shared_lock lock(this->pimplGuard_); + return impl_->labelcount(); + } else { + return 0; + } } else { - return impl_ ? impl_->size() : 0; + if (ready()) { + std::shared_lock lock(this->pimplGuard_); + return impl_->size(); + } else { + return 0; + } } } vecsim_stl::set getLabelsSet() const override { vecsim_stl::set labels(this->allocator); - if (impl_) { + if (ready()) { + std::shared_lock lock(this->pimplGuard_); impl_->on_ids([&labels](size_t label) { labels.insert(label); }); } return labels; @@ -549,7 +640,12 @@ class SVSIndex : public VecSimIndexAbstract, fl } bool isLabelExists(labelType label) const override { - return impl_ ? impl_->has_id(label) : false; + if (ready()) { + std::shared_lock lock(this->pimplGuard_); + return impl_->has_id(label); + } else { + return false; + } } size_t getParallelism() const override { return threadpool_.getParallelism(); } @@ -564,13 +660,23 @@ class SVSIndex : public VecSimIndexAbstract, fl } double getDistanceFrom_Unsafe(labelType label, const void *vector_data) const override { - if (!impl_ || !impl_->has_id(label)) { + if (!ready()) { return std::numeric_limits::quiet_NaN(); - }; + } + { + std::shared_lock lock(this->pimplGuard_); + if (!impl_->has_id(label)) { + return std::numeric_limits::quiet_NaN(); + } + } + auto query_datum = std::span{static_cast(vector_data), this->dim}; - auto dist = impl_->get_distance(label, query_datum); - return toVecSimDistance(dist); + { + std::shared_lock lock(this->pimplGuard_); + auto dist = impl_->get_distance(label, query_datum); + return toVecSimDistance(dist); + } } VecSimQueryReply *topKQuery(const void *queryBlob, size_t k, @@ -580,36 +686,38 @@ class SVSIndex : public VecSimIndexAbstract, fl if (k == 0 || this->indexLabelCount() == 0) { return rep; } + { + // limit result size to index size + k = std::min(k, this->indexLabelCount()); + + auto processed_query_ptr = this->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + + auto query = svs::data::ConstSimpleDataView{ + static_cast(processed_query), 1, this->dim}; + auto result = svs::QueryResult{query.size(), k}; + std::shared_lock lock(this->pimplGuard_); + auto sp = svs_details::joinSearchParams(impl_->get_search_parameters(), queryParams, + is_two_level_lvq); + + auto timeoutCtx = queryParams ? queryParams->timeoutCtx : nullptr; + auto cancel = [timeoutCtx]() { return VECSIM_TIMEOUT(timeoutCtx); }; + + impl_->search(result.view(), query, sp, cancel); + if (cancel()) { + rep->code = VecSim_QueryReply_TimedOut; + return rep; + } - // limit result size to index size - k = std::min(k, this->indexLabelCount()); - - auto processed_query_ptr = this->preprocessQuery(queryBlob); - const void *processed_query = processed_query_ptr.get(); - - auto query = svs::data::ConstSimpleDataView{ - static_cast(processed_query), 1, this->dim}; - auto result = svs::QueryResult{query.size(), k}; - auto sp = svs_details::joinSearchParams(impl_->get_search_parameters(), queryParams, - is_two_level_lvq); - - auto timeoutCtx = queryParams ? queryParams->timeoutCtx : nullptr; - auto cancel = [timeoutCtx]() { return VECSIM_TIMEOUT(timeoutCtx); }; - - impl_->search(result.view(), query, sp, cancel); - if (cancel()) { - rep->code = VecSim_QueryReply_TimedOut; - return rep; - } - - assert(result.n_queries() == 1); + assert(result.n_queries() == 1); - const auto n_neighbors = result.n_neighbors(); - rep->results.reserve(n_neighbors); + const auto n_neighbors = result.n_neighbors(); + rep->results.reserve(n_neighbors); - for (size_t i = 0; i < n_neighbors; i++) { - rep->results.push_back( - VecSimQueryResult{result.index(0, i), toVecSimDistance(result.distance(0, i))}); + for (size_t i = 0; i < n_neighbors; i++) { + rep->results.push_back( + VecSimQueryResult{result.index(0, i), toVecSimDistance(result.distance(0, i))}); + } } // Workaround for VecSim merge_results() that expects results to be sorted // by score, then by id from both indices. @@ -625,58 +733,61 @@ class SVSIndex : public VecSimIndexAbstract, fl if (radius == 0 || this->indexLabelCount() == 0) { return rep; } + { + std::shared_lock lock(this->pimplGuard_); + + auto timeoutCtx = queryParams ? queryParams->timeoutCtx : nullptr; + auto cancel = [timeoutCtx]() { return VECSIM_TIMEOUT(timeoutCtx); }; + + // Prepare query blob for SVS + auto processed_query_ptr = this->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + std::span query{static_cast(processed_query), + this->dim}; + + // Base search parameters for the SVS iterator schedule. + auto sp = svs_details::joinSearchParams(impl_->get_search_parameters(), queryParams, + is_two_level_lvq); + // SVS BatchIterator handles the search in batches + // The batch size is set to the index search window size by default + const size_t batch_size = sp.buffer_config_.get_search_window_size(); + + // Create SVS BatchIterator for range search + // Search result is cached in the iterator and can be accessed by the user + auto svs_it = impl_->make_batch_iterator(query); + svs_it.next(batch_size, cancel); + if (cancel()) { + rep->code = VecSim_QueryReply_TimedOut; + return rep; + } - auto timeoutCtx = queryParams ? queryParams->timeoutCtx : nullptr; - auto cancel = [timeoutCtx]() { return VECSIM_TIMEOUT(timeoutCtx); }; - - // Prepare query blob for SVS - auto processed_query_ptr = this->preprocessQuery(queryBlob); - const void *processed_query = processed_query_ptr.get(); - std::span query{static_cast(processed_query), - this->dim}; - - // Base search parameters for the SVS iterator schedule. - auto sp = svs_details::joinSearchParams(impl_->get_search_parameters(), queryParams, - is_two_level_lvq); - // SVS BatchIterator handles the search in batches - // The batch size is set to the index search window size by default - const size_t batch_size = sp.buffer_config_.get_search_window_size(); - - // Create SVS BatchIterator for range search - // Search result is cached in the iterator and can be accessed by the user - auto svs_it = impl_->make_batch_iterator(query); - svs_it.next(batch_size, cancel); - if (cancel()) { - rep->code = VecSim_QueryReply_TimedOut; - return rep; - } - - // range search using epsilon - const auto epsilon = queryParams && queryParams->svsRuntimeParams.epsilon != 0 - ? queryParams->svsRuntimeParams.epsilon - : this->epsilon; - - const auto range_search_boundaries = radius * (1.0 + std::abs(epsilon)); - bool keep_searching = true; - - // Loop while iterator cache is not empty and search radius + epsilon is not exceeded - while (keep_searching && svs_it.size() > 0) { - // Iterate over the cached search results - for (auto &neighbor : svs_it) { - const auto dist = toVecSimDistance(neighbor.distance()); - if (dist <= radius) { - rep->results.push_back(VecSimQueryResult{neighbor.id(), dist}); - } else if (dist > range_search_boundaries) { - keep_searching = false; + // range search using epsilon + const auto epsilon = queryParams && queryParams->svsRuntimeParams.epsilon != 0 + ? queryParams->svsRuntimeParams.epsilon + : this->epsilon; + + const auto range_search_boundaries = radius * (1.0 + std::abs(epsilon)); + bool keep_searching = true; + + // Loop while iterator cache is not empty and search radius + epsilon is not exceeded + while (keep_searching && svs_it.size() > 0) { + // Iterate over the cached search results + for (auto &neighbor : svs_it) { + const auto dist = toVecSimDistance(neighbor.distance()); + if (dist <= radius) { + rep->results.push_back(VecSimQueryResult{neighbor.id(), dist}); + } else if (dist > range_search_boundaries) { + keep_searching = false; + } } - } - // If search radius + epsilon is not exceeded, request SVS BatchIterator for the next - // batch - if (keep_searching) { - svs_it.next(batch_size, cancel); - if (cancel()) { - rep->code = VecSim_QueryReply_TimedOut; - return rep; + // If search radius + epsilon is not exceeded, request SVS BatchIterator for the next + // batch + if (keep_searching) { + svs_it.next(batch_size, cancel); + if (cancel()) { + rep->code = VecSim_QueryReply_TimedOut; + return rep; + } } } } @@ -734,15 +845,16 @@ class SVSIndex : public VecSimIndexAbstract, fl } void runGC() override { - if (impl_) { + if (ready()) { + std::shared_lock lock(this->pimplGuard_); // There is documentation for consolidate(): // https://intel.github.io/ScalableVectorSearch/python/dynamic.html#svs.DynamicVamana.consolidate - impl_->consolidate(); + // impl_->consolidate(); // There is documentation for compact(): // https://intel.github.io/ScalableVectorSearch/python/dynamic.html#svs.DynamicVamana.compact impl_->compact(); } - num_marked_deleted = 0; + num_marked_deleted.store(0, std::memory_order_relaxed); } #ifdef BUILD_TESTS diff --git a/src/VecSim/algorithms/svs/svs_extensions.h b/src/VecSim/algorithms/svs/svs_extensions.h index 2e93a6a58..f1734c79f 100644 --- a/src/VecSim/algorithms/svs/svs_extensions.h +++ b/src/VecSim/algorithms/svs/svs_extensions.h @@ -10,10 +10,20 @@ #pragma once #include "VecSim/algorithms/svs/svs_utils.h" #include "svs/extensions/vamana/scalar.h" +// Tells SQDataset that the concurrent SegmentedBlocked tag is a blocked allocator, so +// that resize()/compact() are not constrained away. Without it a scalar-quantized +// concurrent index fails to compile the moment the index grows. +#include "svs/concurrent/extensions/scalar.h" #if HAVE_SVS_LVQ #include SVS_LVQ_HEADER #include SVS_LEANVEC_HEADER +// The same treatment for the LVQ and LeanVec datasets, each of which keeps its own private +// copy of the blocked-allocator trait. LeanVec additionally picks the allocator for its inner +// datasets through that trait, and its blocked branch hardcodes svs::data::Blocked -- so +// without this the inner storage would quietly lose the grow-stable guarantee. +#include "svs/concurrent/extensions/lvq.h" +#include "svs/concurrent/extensions/leanvec.h" #endif // HAVE_SVS_LVQ // Scalar Quantization traits for SVS @@ -21,7 +31,7 @@ template struct SVSStorageTraits { using element_type = std::int8_t; using allocator_type = svs_details::SVSAllocator; - using blocked_type = svs::data::Blocked>; + using blocked_type = svs::concurrent::SegmentedBlocked>; using index_storage_type = svs::quantization::scalar::SQDataset; @@ -32,7 +42,7 @@ struct SVSStorageTraits { // SVS block size is a power of two, so we can use it directly auto svs_bs = svs_details::SVSBlockSize(block_size, element_size(dim)); allocator_type data_allocator{std::move(allocator)}; - return svs::make_blocked_allocator_handle({svs_bs}, data_allocator); + return svs_details::make_segmented_blocked_allocator_handle({svs_bs}, data_allocator); } static constexpr VecSimSvsQuantBits get_compression_mode() { return VecSimSvsQuant_Scalar; } @@ -89,7 +99,7 @@ template struct SVSStorageTraits 1)>> { using allocator_type = svs_details::SVSAllocator; - using blocked_type = svs::data::Blocked>; + using blocked_type = svs::concurrent::SegmentedBlocked>; using strategy_type = typename svs_details::LVQSelector::strategy; using index_storage_type = svs::quantization::lvq::LVQDataset @@ -181,7 +191,7 @@ struct SVSStorageTraits struct SVSStorageTraits { using allocator_type = svs_details::SVSAllocator; - using blocked_type = svs::data::Blocked>; + using blocked_type = svs::concurrent::SegmentedBlocked>; using index_storage_type = svs::leanvec::LeanDataset, svs::leanvec::UsingLVQ, svs::Dynamic, svs::Dynamic, blocked_type>; @@ -212,7 +222,7 @@ struct SVSStorageTraits { auto elem_size = std::max(primary_element_size(dim), secondary_element_size(dim)); auto svs_bs = svs_details::SVSBlockSize(block_size, elem_size); allocator_type data_allocator{std::move(allocator)}; - return svs::make_blocked_allocator_handle({svs_bs}, data_allocator); + return svs_details::make_segmented_blocked_allocator_handle({svs_bs}, data_allocator); } template diff --git a/src/VecSim/algorithms/svs/svs_serializer_impl.h b/src/VecSim/algorithms/svs/svs_serializer_impl.h index 2780d3457..5e551ef81 100644 --- a/src/VecSim/algorithms/svs/svs_serializer_impl.h +++ b/src/VecSim/algorithms/svs/svs_serializer_impl.h @@ -93,17 +93,17 @@ void SVSIndex compareMetadataFile(folder_path + "/metadata"); if constexpr (isMulti) { - auto loaded = svs::index::vamana::auto_multi_dynamic_assemble( + auto loaded = svs::concurrent::auto_multi_dynamic_assemble( folder_path + "/config", SVS_LAZY(graph_builder_t::load(folder_path + "/graph", this->blockSize, this->buildParams, this->getAllocator())), SVS_LAZY(storage_traits_t::load(folder_path + "/data", this->blockSize, this->dim, this->getAllocator())), distance_f(), std::move(threadpool_handle), - svs::index::vamana::MultiMutableVamanaLoad::FROM_MULTI, logger_); + svs::concurrent::MultiMutableVamanaLoad::FROM_MULTI, logger_); impl_ = std::make_unique(std::move(loaded)); } else { - auto loaded = svs::index::vamana::auto_dynamic_assemble( + auto loaded = svs::concurrent::auto_dynamic_assemble( folder_path + "/config", SVS_LAZY(graph_builder_t::load(folder_path + "/graph", this->blockSize, this->buildParams, this->getAllocator())), @@ -112,6 +112,7 @@ void SVSIndex distance_f(), std::move(threadpool_handle), false, logger_); impl_ = std::make_unique(std::move(loaded)); } + setReady(); } template #include +/** + * Definition of a job that inserts a new vector from flat into SVS Index. + */ +struct SVSInsertJob : public AsyncJob { + labelType label; + idType id; + + SVSInsertJob(std::shared_ptr allocator, labelType label_, idType id_, + JobCallback insertCb, VecSimIndex *index_) + : AsyncJob(allocator, SVS_INSERT_VECTOR_JOB, insertCb, index_), label(label_), id(id_) {} +}; + +/** + * Definition of a job that launches partial consolidation on SVS Index. + */ +struct SVSConsolidateJob : public AsyncJob { + std::vector labels; + + SVSConsolidateJob(std::shared_ptr allocator, const std::vector& labels_, + JobCallback insertCb, VecSimIndex *index_) + : AsyncJob(allocator, SVS_CONSOLIDATE_JOB, insertCb, index_), labels(labels_) {} +}; + +/** + * Definition of a job that launches GC on SVS Index. + */ +struct SVSGCJob : public AsyncJob { + SVSGCJob(std::shared_ptr allocator, + JobCallback insertCb, VecSimIndex *index_) + : AsyncJob(allocator, SVS_GC_JOB, insertCb, index_) {} +}; + + /** * @class SVSMultiThreadJob * @brief Represents a multi-threaded asynchronous job for the SVS algorithm. @@ -219,10 +252,11 @@ class SVSMultiThreadJob : public AsyncJob { template class TieredSVSIndex : public VecSimTieredIndex { + using DistType = float; using Self = TieredSVSIndex; - using Base = VecSimTieredIndex; - using flat_index_t = BruteForceIndex; - using backend_index_t = VecSimIndexAbstract; + using Base = VecSimTieredIndex; + using flat_index_t = BruteForceIndex; + using backend_index_t = VecSimIndexAbstract; using svs_index_t = SVSIndexBase; // swaps_journal is used by updateSVSIndex() to track vectors swap operations that were done in @@ -237,7 +271,7 @@ class TieredSVSIndex : public VecSimTieredIndex { using swap_record = std::tuple; constexpr static size_t SKIP_LABEL = std::numeric_limits::max(); std::vector swaps_journal; - // deleted_labels_journal is used by updateSVSIndex() to track vectors that were deleted from + // deleted_labels_journal is used by initSVSIndex() to track vectors that were deleted from // Flat index during SVS index updating. The journal contains the deleted labels. These labels // are used to delete the same vectors from the SVS index at the end of the update. std::vector deleted_labels_journal; @@ -252,11 +286,22 @@ class TieredSVSIndex : public VecSimTieredIndex { 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; + mutable std::shared_mutex updateJobMutex; // The reason of following container just to properly destroy jobs which not executed yet SVSMultiThreadJob::JobsRegistry uncompletedJobs; + std::atomic backendInitSubmited{false}; + std::unordered_set ids_to_init_; + + vecsim_stl::unordered_map> labelToInsertJobs; + // A mapping to hold invalid jobs, so we can dispose them upon index deletion. + vecsim_stl::unordered_map invalidJobs; + idType currInvalidJobId; // A unique arbitrary identifier for accessing invalid jobs + std::mutex invalidJobsLookupGuard; + + size_t flat_buffer_bound; + /// //////////////////////////////////////////////////////////////////////////////////////////////////// // TieredSVS_BatchIterator // @@ -280,7 +325,6 @@ class TieredSVSIndex : public VecSimTieredIndex { VecSimBatchIterator *flat_iterator; VecSimBatchIterator *svs_iterator; - std::shared_lock svs_lock; // On single value indices, this set holds the IDs of the results that were returned from // the flat buffer. @@ -346,7 +390,6 @@ class TieredSVSIndex : public VecSimTieredIndex { void acquire_svs_iterator() { assert(svs_iterator == nullptr); - this->index->mainIndexGuard.lock_shared(); svs_iterator = index->backendIndex->newBatchIterator( this->flat_iterator->getQueryBlob(), queryParams); } @@ -355,7 +398,6 @@ class TieredSVSIndex : public VecSimTieredIndex { if (svs_iterator != nullptr && svs_iterator != depleted()) { delete svs_iterator; svs_iterator = nullptr; - this->index->mainIndexGuard.unlock_shared(); } } @@ -380,7 +422,7 @@ class TieredSVSIndex : public VecSimTieredIndex { std::move(allocator)), index(index), flat_results(this->allocator), svs_results(this->allocator), flat_iterator(index->frontendIndex->newBatchIterator(query_vector, queryParams)), - svs_iterator(nullptr), svs_lock(index->mainIndexGuard, std::defer_lock), + svs_iterator(nullptr), returned_results_set(this->allocator) { if (queryParams) { this->queryParams = @@ -550,28 +592,37 @@ class TieredSVSIndex : public VecSimTieredIndex { private: /** - * @brief Updates the SVS index in a thread-safe manner. + * @brief Init the SVS index in a thread-safe manner. * * This static wrapper function performs the following actions: * - Acquires a lock on the index's updateJobMutex to prevent concurrent updates. * - Clears the indexUpdateScheduled flag to allow future scheduling. * - Configures the number of threads for the underlying SVS index update operation. - * - Calls the updateSVSIndex method to perform the actual index update. + * - Calls the initSVSIndex method to perform the actual index update. * * @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. */ - static void updateSVSIndexWrapper(VecSimIndex *idx, size_t availableThreads) { + static void initSVSIndexWrapper(VecSimIndex *idx, size_t availableThreads) { assert(availableThreads > 0); auto index = static_cast *>(idx); assert(index); // prevent parallel updates - std::lock_guard lock(index->updateJobMutex); + std::lock_guard lock(index->updateJobMutex); // Release the scheduled flag to allow scheduling again index->indexUpdateScheduled.clear(); // Update the SVS index - index->updateSVSIndex(availableThreads); + index->initSVSIndex(availableThreads); + } + + static void executeInsertJobWrapper(AsyncJob *job) { + auto *insert_job = static_cast(job); + auto *job_index = static_cast *>(insert_job->index); + // prevent parallel execution with index initilizing job + std::shared_lock lock(job_index->updateJobMutex); + job_index->executeInsertJob(insert_job); + delete job; } /** @@ -589,40 +640,96 @@ class TieredSVSIndex : public VecSimTieredIndex { * @note no need to implement extra non-static method, as GC logic is simple enough to be done * here. */ + // static void SVSIndexGCWrapper(AsyncJob *job) { + // auto gc_job = static_cast(job); + // auto index = static_cast *>(gc_job->index); + + // std::shared_lock lock(index->updateJobMutex); + // // Do SVS index GC + // index->backendIndex->log(VecSimCommonStrings::LOG_VERBOSE_STRING, + // "running asynchronous GC for tiered SVS index"); + // auto svs_index = index->GetSVSIndex(); + // if (index->backendIndex->indexSize() == 0) { + // index->indexGCScheduled.clear(); + // delete job; + // // No need to run GC on an empty index. + // return; + // } + // // svs_index->setParallelism(std::min(availableThreads, index->backendIndex->indexSize())); + // svs_index->setParallelism(1); + // // VecSimIndexAbstract::runGC() is protected + // fprintf(stderr, "runGC\n"); + // static_cast(index->backendIndex)->runGC(); + // fprintf(stderr, "GC done\n"); + + // // Release the scheduled flag to allow scheduling again + // index->indexGCScheduled.clear(); + // delete job; + // } + static void SVSIndexGCWrapper(VecSimIndex *idx, size_t availableThreads) { assert(availableThreads > 0); auto index = static_cast *>(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(); if (index->backendIndex->indexSize() == 0) { // No need to run GC on an empty index. + index->indexGCScheduled.clear(); return; } index->executeTracingCallback("GCJob::before_run_gc"); + std::lock_guard lock(index->updateJobMutex); + svs_index->setParallelism(std::min(availableThreads, index->backendIndex->indexSize())); // VecSimIndexAbstract::runGC() is protected static_cast(index->backendIndex)->runGC(); + svs_index->setParallelism(1); + + // Release the scheduled flag to allow scheduling again + index->indexGCScheduled.clear(); + } + + static void SVSIndexConsolidateWrapper(AsyncJob *job) { + auto consolidate_job = static_cast(job); + auto index = static_cast *>(consolidate_job->index); + + std::shared_lock lock(index->updateJobMutex); + auto svs_index = index->GetSVSIndex(); + svs_index->setParallelism(1); + svs_index->consolidate(consolidate_job->labels); + delete job; } #ifdef BUILD_TESTS public: #endif - void scheduleSVSIndexUpdate() { + + void updateInsertJobInternalId(idType prev_id, idType new_id, labelType label) { + // Update the pending job id, due to a swap that was caused after the removal of new_id. + assert(new_id != INVALID_ID && prev_id != INVALID_ID); + auto it = this->labelToInsertJobs.find(label); + if (it != this->labelToInsertJobs.end()) { + // There is a pending job for the label of the swapped last id - update its id. + for (SVSInsertJob *job_it : it->second) { + if (job_it->id == prev_id) { + job_it->id = new_id; + } + } + } + } + + void scheduleSVSIndexInit() { // do not schedule if scheduled already if (indexUpdateScheduled.test_and_set()) { return; } auto jobs = SVSMultiThreadJob::createScheduledJobs( - this->allocator, SVS_BATCH_UPDATE_JOB, updateSVSIndexWrapper, this, + this->allocator, SVS_BATCH_UPDATE_JOB, initSVSIndexWrapper, this, std::chrono::microseconds(updateJobWaitTime), &uncompletedJobs); this->submitJobs(jobs); } @@ -637,6 +744,20 @@ class TieredSVSIndex : public VecSimTieredIndex { this->allocator, SVS_GC_JOB, SVSIndexGCWrapper, this, std::chrono::microseconds(updateJobWaitTime), &uncompletedJobs); this->submitJobs(jobs); + + // AsyncJob *new_GC_job = new (this->allocator) + // SVSGCJob(this->allocator, SVSIndexGCWrapper, this); + + // // Insert job to the queue. + // this->submitSingleJob(new_GC_job); + } + + void scheduleSVSIndexConsolidate(labelType label) { + AsyncJob *new_consolidate_job = new (this->allocator) + SVSConsolidateJob(this->allocator, {label}, SVSIndexConsolidateWrapper, this); + + // Insert job to the queue. + this->submitSingleJob(new_consolidate_job); } private: @@ -670,97 +791,143 @@ class TieredSVSIndex : public VecSimTieredIndex { } } - void updateSVSIndex(size_t availableThreads) { + idType setAndSaveInvalidJob(AsyncJob *job) { + this->invalidJobsLookupGuard.lock(); + job->isValid = false; + idType curInvalidId = currInvalidJobId++; + this->invalidJobs.insert({curInvalidId, job}); + this->invalidJobsLookupGuard.unlock(); + return curInvalidId; + } + + void executeInsertJob(SVSInsertJob *job) { + auto svs_index = GetSVSIndex(); + assert(svs_index->ready()); + + // Note that accessing the job fields should occur with flat index guard held (here and later). + this->flatIndexGuard.lock_shared(); + if (!job->isValid) { + this->flatIndexGuard.unlock_shared(); + // Job has been invalidated in the meantime - nothing to execute, and remove it from the + // lookup. + this->invalidJobsLookupGuard.lock(); + this->invalidJobs.erase(job->id); + this->invalidJobsLookupGuard.unlock(); + return; + } + + // Copy the vector blob out of the flat buffer while holding flatIndexGuard, so we + // can release the flat lock before indexing into the SVS backend + size_t data_size = this->frontendIndex->getStoredDataSize(); + auto blob_copy = this->getAllocator()->allocate_unique(data_size); + memcpy(blob_copy.get(), this->frontendIndex->getDataByInternalId(job->id), data_size); + this->flatIndexGuard.unlock_shared(); + + // svs_index->setParallelism(1); + { + std::shared_lock lock(updateJobMutex); + svs_index->addVector(blob_copy.get(), job->label); + } + + // Remove the vector and the insert job from the flat buffer. + this->flatIndexGuard.lock(); + // The job might have been invalidated due to overwrite in the meantime. In this case, + // it was already deleted and the job has been evicted. Otherwise, we need to do it now. + if (job->isValid) { + // Remove the job pointer from the labelToInsertJobs mapping. + auto &jobs = labelToInsertJobs.at(job->label); + for (size_t i = 0; i < jobs.size(); i++) { + if (jobs[i]->id == job->id) { + jobs.erase(jobs.begin() + (long)i); + break; + } + } + if (labelToInsertJobs.at(job->label).empty()) { + labelToInsertJobs.erase(job->label); + } + // Remove the vector from the flat buffer. + // The flat buffer stores data in a contiguous + // array, so deleting an element may move the last element into the freed slot to keep + // ids dense. Capture the last id's label beforehand (after deletion it is no longer in + // the lookup) so we can later fix up its insert job. + labelType last_vec_label = + this->frontendIndex->getVectorLabel(this->frontendIndex->indexSize() - 1); + int deleted = this->frontendIndex->deleteVectorById(job->label, job->id); + if (deleted && job->id != this->frontendIndex->indexSize()) { + // If the vector removal caused a swap with the last id, update the relevant insert job. + this->updateInsertJobInternalId(this->frontendIndex->indexSize(), job->id, + last_vec_label); + } + } else { + // Remove the current job from the invalid jobs' lookup, as we are about to delete it now. + this->invalidJobsLookupGuard.lock(); + this->invalidJobs.erase(job->id); + this->invalidJobsLookupGuard.unlock(); + } + this->flatIndexGuard.unlock(); + } + + void initSVSIndex(size_t availableThreads) { + std::vector ids_to_move; std::vector labels_to_move; std::vector vectors_to_move; - std::vector deleted_labels_during_update; + executeTracingCallback("UpdateJob::before_add_to_svs"); { // lock frontendIndex from modifications - std::shared_lock flat_lock{this->flatIndexGuard}; + // The whole initialization is done under flatIndexGuard + std::lock_guard flat_lock{this->flatIndexGuard}; auto flat_index = this->GetFlatIndex(); - const auto frontend_index_size = this->frontendIndex->indexSize(); + const auto init_batch_size = ids_to_init_.size(); const size_t dim = flat_index->getDim(); - labels_to_move.reserve(frontend_index_size); - vectors_to_move.reserve(frontend_index_size * dim); - - for (idType i = 0; i < frontend_index_size; i++) { - labels_to_move.push_back(flat_index->getVectorLabel(i)); - auto data = flat_index->getDataByInternalId(i); + ids_to_move.reserve(init_batch_size); + labels_to_move.reserve(init_batch_size); + vectors_to_move.reserve(init_batch_size * dim); + + for (idType id : ids_to_init_) { + ids_to_move.push_back(id); + labels_to_move.push_back(flat_index->getVectorLabel(id)); + auto data = flat_index->getDataByInternalId(id); vectors_to_move.insert(vectors_to_move.end(), data, data + dim); } - // reset journal to the current frontend index state - swaps_journal.clear(); - deleted_labels_journal.clear(); - } // release frontend index + ids_to_init_.clear(); - executeTracingCallback("UpdateJob::before_add_to_svs"); - if (!labels_to_move.empty()) { - // lock backend index for writing and add vectors there - std::shared_lock main_shared_lock(this->mainIndexGuard); - auto svs_index = GetSVSIndex(); - assert(labels_to_move.size() == vectors_to_move.size() / this->frontendIndex->getDim()); - if (this->backendIndex->indexSize() == 0) { - // If backend index is empty, we need to initialize it first. + // Nothing to initialize from an empty batch, and setParallelism(0) is not a + // valid request against the shared pool. + if (!labels_to_move.empty()) { + auto svs_index = GetSVSIndex(); svs_index->setParallelism(std::min(availableThreads, labels_to_move.size())); + assert(labels_to_move.size() == vectors_to_move.size() / this->frontendIndex->getDim()); auto impl = svs_index->createImpl(vectors_to_move.data(), labels_to_move.data(), labels_to_move.size()); - - // Upgrade to unique lock to set the new impl - main_shared_lock.unlock(); - std::lock_guard lock(this->mainIndexGuard); + svs_index->setParallelism(1); svs_index->setImpl(std::move(impl)); - } else { - // Backend index is initialized - just add the vectors. - main_shared_lock.unlock(); - std::lock_guard lock(this->mainIndexGuard); - // Upgrade to unique lock to add vectors - svs_index->setParallelism(std::min(availableThreads, labels_to_move.size())); - svs_index->addVectors(vectors_to_move.data(), labels_to_move.data(), - labels_to_move.size()); } - } - executeTracingCallback("UpdateJob::after_add_to_svs"); - // clean-up frontend index - { // lock frontend index for writing and delete moved vectors - std::lock_guard lock(this->flatIndexGuard); - // swap deleted labels journal with the local variable to track deleted labels during - // update - std::swap(deleted_labels_during_update, deleted_labels_journal); - - // Apply swaps from journal to labels_to_move to reflect changes made in meanwhile. - applySwapsToLabelsArray(labels_to_move, this->swaps_journal); - - // delete vectors from the frontend index in reverse order - // it increases the chance of avoiding swaps in the frontend index and performance - // improvement - int deleted = 0; - idType id = labels_to_move.size(); - while (id-- > 0) { - auto label = labels_to_move[id]; + std::sort(ids_to_move.begin(), ids_to_move.end()); + int total_deleted = 0; + for (auto it = ids_to_move.rbegin(); it != ids_to_move.rend(); ++it) { + idType id = *it; + auto label = this->frontendIndex->getVectorLabel(id); // Delete the vector from the frontend index if not in-place updated. if (label != SKIP_LABEL) { - deleted += this->frontendIndex->deleteVectorById(label, id); + labelType last_vec_label = + this->frontendIndex->getVectorLabel(this->frontendIndex->indexSize() - 1); + int deleted = this->frontendIndex->deleteVectorById(label, id); + if (deleted && id != this->frontendIndex->indexSize()) { + // If the vector removal caused a swap with the last id, update the relevant insert job. + this->updateInsertJobInternalId(this->frontendIndex->indexSize(), id, + last_vec_label); + } + total_deleted += deleted; } } - assert(deleted == std::count_if(labels_to_move.begin(), labels_to_move.end(), - [](labelType label) { return label != SKIP_LABEL; }) && + + assert(total_deleted == std::count_if(labels_to_move.begin(), labels_to_move.end(), + [](labelType label) { return label != SKIP_LABEL; }) && "Deleted vectors count does not match the number of labels to delete"); - } - // delete vectors from backend index that were deleted from the frontend index during - // the update process. - { - std::lock_guard main_lock(this->mainIndexGuard); - - std::sort(deleted_labels_during_update.begin(), deleted_labels_during_update.end()); - auto it = std::unique(deleted_labels_during_update.begin(), - deleted_labels_during_update.end()); - deleted_labels_during_update.erase(it, deleted_labels_during_update.end()); - auto svs_index = GetSVSIndex(); - svs_index->deleteVectors(deleted_labels_during_update.data(), - deleted_labels_during_update.size()); - } + } // release frontend index + executeTracingCallback("UpdateJob::after_add_to_svs"); } public: @@ -768,11 +935,14 @@ class TieredSVSIndex : public VecSimTieredIndex { const TieredIndexParams &tiered_index_params, std::shared_ptr allocator) : Base(svs_index, bf_index, tiered_index_params, allocator), - uncompletedJobs(this->allocator) { + uncompletedJobs(this->allocator), + labelToInsertJobs(this->allocator), + invalidJobs(this->allocator), + currInvalidJobId(0) { const auto &tiered_svs_params = tiered_index_params.specificParams.tieredSVSParams; // If flatBufferLimit is not initialized (0), use the default update threshold. - const size_t flat_buffer_bound = tiered_index_params.flatBufferLimit == 0 + flat_buffer_bound = tiered_index_params.flatBufferLimit == 0 ? SVS_VAMANA_DEFAULT_UPDATE_THRESHOLD : tiered_index_params.flatBufferLimit; @@ -802,108 +972,162 @@ class TieredSVSIndex : public VecSimTieredIndex { int addVector(const void *blob, labelType label) override { int ret = 0; auto svs_index = GetSVSIndex(); - size_t update_threshold = 0; size_t frontend_index_size = 0; // In-Place mode - add vector syncronously to the backend index. if (this->getWriteMode() == VecSim_WriteInPlace) { // It is ok to lock everything at once for in-place mode, - // but we will have to unlock averything before calling updateSVSIndexWrapper() + // but we will have to unlock averything before calling initSVSIndexWrapper() // so make the minimal needed lock here. - std::shared_lock backend_shared_lock(this->mainIndexGuard); // Backend index initialization data have to be buffered for proper // compression/training. - if (this->backendIndex->indexSize() == 0) { + if ((!svs_index->ready()) && (!this->backendInitSubmited.load(std::memory_order_acquire))) { // If backend index size is 0, first collect vectors in frontend index // lock in scope to ensure that these will be released before - // updateSVSIndexWrapper() is called. + // initSVSIndexWrapper() is called. { std::lock_guard lock(this->flatIndexGuard); + if (this->frontendIndex->isLabelExists(label)) { + deleteAndUpdateInitIds(label); + } + ids_to_init_.insert(this->frontendIndex->indexSize()); ret = this->frontendIndex->addVector(blob, label); // If frontend size exceeds the update job threshold, ... frontend_index_size = this->frontendIndex->indexSize(); } // ... move vectors to the backend index. if (frontend_index_size >= this->trainingTriggerThreshold) { - // updateSVSIndexWrapper() accures it's own locks - backend_shared_lock.unlock(); + // initSVSIndexWrapper() accures it's own locks // initialize the SVS index synchonously using current thread only - updateSVSIndexWrapper(this, 1); + initSVSIndexWrapper(this, 1); } return ret; } else { // backend index is initialized - we can add the vector directly - backend_shared_lock.unlock(); auto storage_blob = this->frontendIndex->preprocessForStorage(blob); // prevent update job from running in parallel and lock any access to the backend // index - std::scoped_lock lock(this->updateJobMutex, this->mainIndexGuard); + // Only updateJobMutex is needed here, not mainIndexGuard: the concurrent + // backend index serializes writes against concurrent readers itself. + std::lock_guard lock(this->updateJobMutex); // Defensive: ensure single-threaded operation for write-in-place mode. // parallelism_ defaults to 1, so this is a no-op in the normal case. svs_index->setParallelism(1); - return this->backendIndex->addVector(storage_blob.get(), label); + int deleted = 0; + if (!this->backendIndex->isMultiValue()) { + deleted = svs_index->deleteVector(label); + if (deleted > 0) + svs_index->consolidate({label}); + } + return this->backendIndex->addVector(storage_blob.get(), label) - deleted; } } assert(this->getWriteMode() != VecSim_WriteInPlace && "InPlace mode returns early"); // Async mode - add vector to the frontend index and schedule an update job if needed. - if (!this->backendIndex->isMultiValue()) { - { - std::shared_lock lock(this->flatIndexGuard); - // If the label already exists in the frontend index, we should count it - // to prevent the case when existing vector is moved meanwhile by the update job. - if (this->frontendIndex->isLabelExists(label)) { - ret = -1; + if ((!svs_index->ready()) && (!this->backendInitSubmited.load(std::memory_order_acquire))) { + // Add vector to the frontend index. + std::lock_guard lock(this->flatIndexGuard); + if ((!svs_index->ready()) && (!this->backendInitSubmited.load(std::memory_order_acquire))) { + if (!this->frontendIndex->isMultiValue() && this->frontendIndex->isLabelExists(label)) { + deleteAndUpdateInitIds(label); } - } - // Remove vector from the backend index if it exists in case of non-MULTI. - auto label_exists = [&]() { - std::shared_lock lock(this->mainIndexGuard); - return svs_index->isLabelExists(label); - }(); - - if (label_exists) { - std::lock_guard lock(this->mainIndexGuard); - ret -= this->backendIndex->deleteVector(label); + ids_to_init_.insert(this->frontendIndex->indexSize()); + const auto ft_ret = this->frontendIndex->addVector(blob, label); + ret = std::max(ret + ft_ret, 0); + + if (this->frontendIndex->indexSize() >= this->trainingTriggerThreshold) { + this->backendInitSubmited.store(true, std::memory_order_release); + scheduleSVSIndexInit(); + } + return ret; } } - { // Add vector to the frontend index. - std::lock_guard lock(this->flatIndexGuard); - const auto ft_ret = this->frontendIndex->addVector(blob, label); - if (ft_ret == 0) { // Vector was overriden - add 'skiping' swap to the journal. - assert(!this->backendIndex->isMultiValue() && - "addVector() may return 0 for single value indices only"); - for (auto id : this->frontendIndex->getElementIds(label)) { - this->swaps_journal.emplace_back(SKIP_LABEL, id, id); + this->flatIndexGuard.lock_shared(); + if (svs_index->ready() && (this->frontendIndex->indexSize() >= flat_buffer_bound)) { + this->flatIndexGuard.unlock_shared(); + auto storage_blob = this->frontendIndex->preprocessForStorage(blob); + + if (!this->backendIndex->isMultiValue()) { + auto deleted = svs_index->deleteVector(label); + if (deleted > 0) + scheduleSVSIndexConsolidate(label); + } + + std::shared_lock lock(updateJobMutex); + return ret = svs_index->addVector(storage_blob.get(), label); + } else { + this->flatIndexGuard.unlock_shared(); + this->flatIndexGuard.lock(); + idType new_flat_id = this->frontendIndex->indexSize(); + if (this->frontendIndex->isLabelExists(label) && !this->frontendIndex->isMultiValue()) { + if (this->labelToInsertJobs.count(label) == 0) { + // No pending insert job. + // Just replace vector in the frontend + // If this label already exists, this will do overwrite. + ret = this->frontendIndex->addVector(blob, label); + this->flatIndexGuard.unlock(); + return ret; + } + + // Overwrite the vector and invalidate its only pending job (since we are not in MULTI). + // Label exists, but job doesn't. It means the label is used for initialization. + // Just create new job, this job will remove duplicate from the backend. + auto *old_job = this->labelToInsertJobs.at(label).at(0); + old_job->id = this->setAndSaveInvalidJob(old_job); + this->labelToInsertJobs.erase(label); + ret = 0; + // We are going to update the internal id that currently holds the vector associated with + // the given label. + new_flat_id = + dynamic_cast *>(this->frontendIndex) + ->getIdOfLabel(label); + // If we are adding a new element (rather than updating an exiting one) we may need to + // increase index capacity. + } + // If this label already exists, this will do overwrite. + ret += this->frontendIndex->addVector(blob, label); + + AsyncJob *new_insert_job = new (this->allocator) + SVSInsertJob(this->allocator, label, new_flat_id, executeInsertJobWrapper, this); + + // Save a pointer to the job, so that if the vector is overwritten, we'll have an indication. + if (this->labelToInsertJobs.find(label) != this->labelToInsertJobs.end()) { + // There's already a pending insert job for this label, add another one (without overwrite, + // only possible in multi index) + assert(this->backendIndex->isMultiValue()); + this->labelToInsertJobs.at(label).push_back((SVSInsertJob *)new_insert_job); + } else { + vecsim_stl::vector new_jobs_vec(1, (SVSInsertJob *)new_insert_job, + this->allocator); + this->labelToInsertJobs.insert({label, new_jobs_vec}); + } + this->flatIndexGuard.unlock(); + + // Here, a worker might ingest the previous vector that was stored under "label" + // (in case of override in non-MULTI index) - so if it's there, we remove it + // we submit the insert job. + if (!this->backendIndex->isMultiValue()) { + if (svs_index->ready()) { + // If we removed the previous vector from both svs and flat in the overwrite process, + // we still return 0 (not -1). + auto deleted = svs_index->deleteVector(label); + if (deleted > 0) + scheduleSVSIndexConsolidate(label); + ret = std::max(ret - deleted, 0); } - deleted_labels_journal.push_back(label); } - ret = std::max(ret + ft_ret, 0); - // Check frontend index size to determine if an update job schedule is needed. - frontend_index_size = this->frontendIndex->indexSize(); - } - { - // If main index is empty then update_threshold is trainingTriggerThreshold, - // overwise it is updateTriggerThreshold. - std::shared_lock lock(this->mainIndexGuard); - update_threshold = this->backendIndex->indexSize() == 0 ? this->trainingTriggerThreshold - : this->updateTriggerThreshold; - } - if (frontend_index_size >= update_threshold) { - scheduleSVSIndexUpdate(); - } - return ret; + // Insert job to the queue and signal the workers' updater. + this->submitSingleJob(new_insert_job); + return ret; + } } - int deleteAndRecordSwaps_Unsafe(labelType label) { + void deleteAndUpdateInitIds(labelType label) { auto deleting_ids = this->frontendIndex->getElementIds(label); - // assert if all elements of deleting_ids are unique - assert(std::set(deleting_ids.begin(), deleting_ids.end()).size() == deleting_ids.size() && - "deleting_ids should contain unique ids"); - // Sort deleting_ids by id descending order std::sort(deleting_ids.begin(), deleting_ids.end(), [](const auto &a, const auto &b) { return a > b; }); @@ -918,65 +1142,85 @@ class TieredSVSIndex : public VecSimTieredIndex { }) && "updated_ids should be a subset of deleting_ids"); - // Record swaps in the journal. - for (auto id : deleting_ids) { - auto it = updated_ids.find(id); - if (it != updated_ids.end()) { - assert(id == it->first && "id in updated_ids should match the id in deleting_ids"); - auto newId = id; - auto oldId = it->second.first; - auto oldLabel = it->second.second; - this->swaps_journal.emplace_back(oldLabel, oldId, newId); - } else { - // No swap, just delete is marked by oldId == newId == deleted id - this->swaps_journal.emplace_back(SKIP_LABEL, id, id); + std::vector new_ids_to_init; + for (auto &it : updated_ids) { + idType prev_id = it.second.first; + idType new_id = it.first; + + if (ids_to_init_.count(prev_id) > 0) { + ids_to_init_.erase(prev_id); + new_ids_to_init.push_back(new_id); } + labelType updated_vec_label = it.second.second; + this->updateInsertJobInternalId(prev_id, new_id, updated_vec_label); } - deleted_labels_journal.push_back(label); - return deleting_ids.size(); + for (idType new_id : new_ids_to_init) { + ids_to_init_.insert(new_id); + } } int deleteVector(labelType label) override { int ret = 0; - auto svs_index = GetSVSIndex(); - // Backend index deletions to be synchronized with the frontend index, - // elsewhere there is the risk of labels duplication in both indices which can lead to wrong - // results of topK queries. In such case we should behave as if InPlace mode is always set. - bool label_exists = [&]() { - std::shared_lock lock(this->flatIndexGuard); - return this->frontendIndex->isLabelExists(label); - }(); - - if (label_exists) { - std::lock_guard lock(this->flatIndexGuard); - ret = this->deleteAndRecordSwaps_Unsafe(label); - } - label_exists = [&]() { - std::shared_lock lock(this->mainIndexGuard); - return svs_index->isLabelExists(label); - }(); + this->flatIndexGuard.lock_shared(); + if (this->frontendIndex->isLabelExists(label)) { + this->flatIndexGuard.unlock_shared(); + std::lock_guard flat_lock{this->flatIndexGuard}; + // Check again if the label exists, as it may have been removed while we released the lock. + if (this->frontendIndex->isLabelExists(label)) { + auto deleting_ids = this->frontendIndex->getElementIds(label); + if (deleting_ids.size() == 0) return 0; + + // assert if all elements of deleting_ids are unique + assert(std::set(deleting_ids.begin(), deleting_ids.end()).size() == deleting_ids.size() && + "deleting_ids should contain unique ids"); + + // If id is deleted, don't use it for initialization + for (idType id : deleting_ids) { + ids_to_init_.erase(id); + } + + if (this->labelToInsertJobs.count(label) > 0) { + // Invalidate the pending insert job(s) into SVS associated with this label + auto &insert_jobs = this->labelToInsertJobs.at(label); + for (auto *job : insert_jobs) { + job->id = this->setAndSaveInvalidJob(job); + } + // Remove the pending insert job(s) from the labelToInsertJobs mapping. + this->labelToInsertJobs.erase(label); + } - if (label_exists) { - std::lock_guard lock(this->mainIndexGuard); - ret += this->backendIndex->deleteVector(label); + deleteAndUpdateInitIds(label); + ret += deleting_ids.size(); + } + } else { + this->flatIndexGuard.unlock_shared(); } + + auto deleted = this->backendIndex->deleteVector(label); + if (deleted > 0) { + if (this->getWriteMode() == VecSim_WriteInPlace) { + GetSVSIndex()->consolidate({label}); + } else { + scheduleSVSIndexConsolidate(label); + } + } + ret += deleted; return ret; } + size_t getNumMarkedDeleted() const override { return this->GetSVSIndex()->getNumMarkedDeleted(); } size_t indexSize() const override { std::shared_lock flat_lock(this->flatIndexGuard); - std::shared_lock main_lock(this->mainIndexGuard); return this->frontendIndex->indexSize() + this->backendIndex->indexSize(); } size_t indexCapacity() const override { std::shared_lock flat_lock(this->flatIndexGuard); - std::shared_lock main_lock(this->mainIndexGuard); return this->frontendIndex->indexCapacity() + this->backendIndex->indexCapacity(); } @@ -1010,23 +1254,24 @@ class TieredSVSIndex : public VecSimTieredIndex { }; { // Use try_lock to avoid blocking the main thread during long-running - // training operations. updateSVSIndexWrapper holds updateJobMutex for + // training operations. initSVSIndexWrapper holds updateJobMutex for // the entire training duration (which can take 40-85s on slow machines). // If the mutex is held, training is actively running, so we report // indexUpdateScheduled = true (BACKGROUND_INDEXING = 1). - std::unique_lock lock(this->updateJobMutex, std::try_to_lock); + std::unique_lock lock(this->updateJobMutex, std::try_to_lock); if (lock.owns_lock()) { svsTieredInfo.indexUpdateScheduled = this->indexUpdateScheduled.test(); } else { - // Mutex is held by updateSVSIndexWrapper — training is in progress. + // Mutex is held by initSVSIndexWrapper — training is in progress. svsTieredInfo.indexUpdateScheduled = true; } } info.tieredInfo.specificTieredBackendInfo.svsTieredInfo = svsTieredInfo; + // Background indexing is in progress whenever the flat buffer is non-empty: every vector + // buffered there has either a pending per-vector insert job or a batch-init job that will + // drain it into the backend. info.tieredInfo.backgroundIndexing = - svsTieredInfo.indexUpdateScheduled && info.tieredInfo.frontendCommonInfo.indexSize > 0 - ? VecSimBool_TRUE - : VecSimBool_FALSE; + info.tieredInfo.frontendCommonInfo.indexSize > 0 ? VecSimBool_TRUE : VecSimBool_FALSE; return info; } @@ -1068,17 +1313,81 @@ class TieredSVSIndex : public VecSimTieredIndex { VecSimQueryReply *topKQuery(const void *queryBlob, size_t k, VecSimQueryParams *queryParams) const override { - // SVS implements it's own distance computation functions which may cause sligthly different - // distance values than VecSim Flat Index does, so we always have to merge results with set. - return this->template topKQueryImp(queryBlob, k, queryParams); + // SVS handles its own internal locking for concurrent search + modification, + // so we don't need mainIndexGuard for backend queries. + this->flatIndexGuard.lock_shared(); + + if (this->frontendIndex->indexSize() == 0) { + this->flatIndexGuard.unlock_shared(); + + auto processed_query_ptr = this->frontendIndex->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + return this->backendIndex->topKQuery(processed_query, k, queryParams); + } else { + auto flat_results = this->frontendIndex->topKQuery(queryBlob, k, queryParams); + this->flatIndexGuard.unlock_shared(); + + if (flat_results->code != VecSim_QueryReply_OK) { + assert(flat_results->results.empty()); + return flat_results; + } + + auto processed_query_ptr = this->frontendIndex->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + auto main_results = this->backendIndex->topKQuery(processed_query, k, queryParams); + + if (main_results->code != VecSim_QueryReply_OK) { + VecSimQueryReply_Free(flat_results); + assert(main_results->results.empty()); + return main_results; + } + + return merge_result_lists(main_results, flat_results, k); + } } VecSimQueryReply *rangeQuery(const void *queryBlob, double radius, VecSimQueryParams *queryParams, VecSimQueryReply_Order order) const override { - // SVS implements it's own distance computation functions which may cause sligthly different - // distance values than VecSim Flat Index does, so we always have to merge results with set. - return this->template rangeQueryImp(queryBlob, radius, queryParams, order); + // SVS handles its own internal locking for concurrent search + modification, + // so we don't need mainIndexGuard for backend queries. + this->flatIndexGuard.lock_shared(); + + if (this->frontendIndex->indexSize() == 0) { + this->flatIndexGuard.unlock_shared(); + + auto processed_query_ptr = this->frontendIndex->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + auto res = this->backendIndex->rangeQuery(processed_query, radius, queryParams); + sort_results(res, order); + return res; + } else { + auto flat_results = this->frontendIndex->rangeQuery(queryBlob, radius, queryParams); + this->flatIndexGuard.unlock_shared(); + + if (flat_results->code != VecSim_QueryReply_OK) { + return flat_results; + } + + auto processed_query_ptr = this->frontendIndex->preprocessQuery(queryBlob); + const void *processed_query = processed_query_ptr.get(); + auto main_results = + this->backendIndex->rangeQuery(processed_query, radius, queryParams); + + if (BY_SCORE == order) { + sort_results_by_score_then_id(main_results); + sort_results_by_score_then_id(flat_results); + + auto code = main_results->code; + VecSimQueryReply *ret = merge_result_lists(main_results, flat_results, -1); + ret->code = code; + return ret; + } else { // BY_ID + concat_results(main_results, flat_results); + filter_results_by_id(main_results); + return main_results; + } + } } VecSimBatchIterator *newBatchIterator(const void *queryBlob, @@ -1097,7 +1406,6 @@ 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. - std::lock_guard lock{this->mainIndexGuard}; if (this->backendIndex->indexSize() == 0) { // No need to run GC on an empty index. return; @@ -1115,11 +1423,9 @@ class TieredSVSIndex : public VecSimTieredIndex { void acquireSharedLocks() override { this->flatIndexGuard.lock_shared(); - this->mainIndexGuard.lock_shared(); } void releaseSharedLocks() override { - this->mainIndexGuard.unlock_shared(); this->flatIndexGuard.unlock_shared(); } }; diff --git a/src/VecSim/algorithms/svs/svs_utils.h b/src/VecSim/algorithms/svs/svs_utils.h index 7d90f97c9..33c1135a9 100644 --- a/src/VecSim/algorithms/svs/svs_utils.h +++ b/src/VecSim/algorithms/svs/svs_utils.h @@ -16,6 +16,10 @@ #include "svs/core/distance.h" #include "svs/lib/float16.h" #include "svs/index/vamana/dynamic_index.h" +// Pulls in `svs::index::vamana::concurrent` and the `svs::concurrent` alias. The concurrent +// namespace redeclares only the entities it has to replace; every other name (build/search +// parameters, extension points) still resolves to the enclosing `svs::index::vamana`. +#include "svs/concurrent/concurrent.h" #if HAVE_SVS_LVQ #include "svs/cpuid.h" @@ -204,6 +208,21 @@ inline std::pair isSVSQuantBitsSupported(VecSimSvsQuan // unreachable code, but to avoid compiler warning return std::make_pair(VecSimSvsQuant_NONE, false); } + +// The `svs::concurrent` counterpart of `svs::make_blocked_allocator_handle`. +// +// The upstream helper cannot be reused here: it hardcodes `svs::data::Blocked` in its return +// type, so there is no way to ask it for the grow-stable tag. The type-erasing +// `AllocatorHandle` wrapper is what makes the compressed datasets (which take an allocator +// handle rather than a concrete allocator) work with VecSim's own allocator. +template +svs::concurrent::SegmentedBlocked> +make_segmented_blocked_allocator_handle(const svs::data::BlockingParameters ¶meters, + Alloc alloc) { + using handle_type = svs::AllocatorHandle; + return svs::concurrent::SegmentedBlocked{ + parameters, svs::make_allocator_handle(std::move(alloc))}; +} } // namespace svs_details template ; // Used in creating storage + // `SegmentedBlocked` is a drop-in replacement for `svs::data::Blocked` (it derives from + // it and carries the same blocking parameters) that additionally keeps already-published + // elements at a stable address when the dataset grows, which is what allows a concurrent + // reader to keep a pointer into the data while a writer appends. + using blocked_type = svs::concurrent::SegmentedBlocked; // svs::Dynamic means runtime dimensionality in opposite to compile-time dimensionality - using index_storage_type = svs::data::BlockedData; + using index_storage_type = + svs::concurrent::SegmentedBlockedData; static constexpr bool is_compressed() { return false; } @@ -278,9 +302,13 @@ struct SVSStorageTraits { template struct SVSGraphBuilder { using allocator_type = svs_details::SVSAllocator; - using blocked_type = svs::data::Blocked; - using graph_data_type = svs::data::BlockedData; - using graph_type = svs::graphs::SimpleGraph; + using blocked_type = svs::concurrent::SegmentedBlocked; + using graph_data_type = + svs::concurrent::SegmentedBlockedData; + // The concurrent `SimpleGraph` has the same shape as the upstream one but guards each + // adjacency list with a sequence lock, so a reader can traverse an edge list that a + // writer is concurrently rewriting. + using graph_type = svs::concurrent::graphs::SimpleGraph; static blocked_type make_blocked_allocator(size_t block_size, size_t graph_max_degree, std::shared_ptr allocator) { @@ -310,7 +338,7 @@ struct SVSGraphBuilder { // based on the data types, which we found to perform better through heuristic analysis. auto prefetch_parameters = svs::index::vamana::extensions::estimate_prefetch_parameters(data); - auto builder = svs::index::vamana::VamanaBuilder( + auto builder = svs::concurrent::VamanaBuilder( graph, data, std::move(distance), parameters, threadpool, prefetch_parameters, logger); // Specific to the Vamana algorithm: diff --git a/src/VecSim/vec_sim_common.h b/src/VecSim/vec_sim_common.h index 3ec945a01..b3ded66c0 100644 --- a/src/VecSim/vec_sim_common.h +++ b/src/VecSim/vec_sim_common.h @@ -50,6 +50,7 @@ extern "C" { #define SVS_VAMANA_DEFAULT_TRAINING_THRESHOLD (10 * DEFAULT_BLOCK_SIZE) // 10 * 1024 vectors // Default batch update threshold for SVS index. #define SVS_VAMANA_DEFAULT_UPDATE_THRESHOLD (1 * DEFAULT_BLOCK_SIZE) // 1 * 1024 vectors +#define SVS_VAMANA_DEFAULT_CONSOLIDATE_THRESHOLD (1 * DEFAULT_BLOCK_SIZE) // 1 * 1024 vectors #define SVS_VAMANA_DEFAULT_SEARCH_WINDOW_SIZE 10 // NOTE: No need to have SVS_VAMANA_DEFAULT_SEARCH_BUFFER_CAPACITY // as the default is determined by the search_window_size @@ -298,7 +299,9 @@ typedef enum { HNSW_SWAP_JOB, HNSW_DISK_JOB, SVS_BATCH_UPDATE_JOB, + SVS_INSERT_VECTOR_JOB, SVS_GC_JOB, + SVS_CONSOLIDATE_JOB, INVALID_JOB // to indicate that finding a JobType >= INVALID_JOB is an error } JobType; diff --git a/tests/unit/test_svs_fp16.cpp b/tests/unit/test_svs_fp16.cpp index fcf2c7dd7..895e8f8a5 100644 --- a/tests/unit/test_svs_fp16.cpp +++ b/tests/unit/test_svs_fp16.cpp @@ -2328,7 +2328,7 @@ class FP16SVSTieredIndexTest : public FP16SVSTest { } } // Submit the index update job. - tiered_index->scheduleSVSIndexUpdate(); + tiered_index->scheduleSVSIndexInit(); ASSERT_EQ(mock_thread_pool.jobQ.size(), mock_thread_pool.thread_pool_size); // Execute the job from the queue and validate that the index was updated properly. @@ -2848,8 +2848,9 @@ TYPED_TEST(FP16SVSTieredIndexTest, deleteVector) { ASSERT_EQ(tiered_index->indexSize(), 1); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 1); - // Move the vector to SVS by executing the insert job. - mock_thread_pool.thread_iteration(); + // Move the vector to SVS by executing the pending jobs. + while (mock_thread_pool.jobQ.size() > 0) + mock_thread_pool.thread_iteration(); ASSERT_EQ(tiered_index->indexLabelCount(), 1); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 1); // Scalar quantization accuracy is insufficient for this check. @@ -2896,7 +2897,9 @@ TYPED_TEST(FP16SVSTieredIndexTest, deleteVectorMulti) { ASSERT_EQ(tiered_index->deleteVector(vec_label), 2); ASSERT_EQ(tiered_index->indexSize(), 0); ASSERT_EQ(tiered_index->indexLabelCount(), 0); - mock_thread_pool.thread_iteration(); + + while (mock_thread_pool.jobQ.size() > 0) + mock_thread_pool.thread_iteration(); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Test deleting a label for which both of its vector's is in the flat index. @@ -2910,13 +2913,21 @@ TYPED_TEST(FP16SVSTieredIndexTest, deleteVectorMulti) { ASSERT_EQ(tiered_index->deleteVector(vec_label), 2); ASSERT_EQ(tiered_index->indexLabelCount(), 0); + + ASSERT_EQ(mock_thread_pool.jobQ.size(), 2); + mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 1); mock_thread_pool.thread_iteration(); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Test deleting a label for which both of its vector's is in SVS index. this->GenerateAndAddVector(tiered_index, dim, vec_label, vec_label); this->GenerateAndAddVector(tiered_index, dim, vec_label, other_vec_val); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 2); mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 1); + mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); ASSERT_EQ(tiered_index->indexLabelCount(), 1); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 2); diff --git a/tests/unit/test_svs_tiered.cpp b/tests/unit/test_svs_tiered.cpp index 9542e7cd6..e7142f455 100644 --- a/tests/unit/test_svs_tiered.cpp +++ b/tests/unit/test_svs_tiered.cpp @@ -476,7 +476,7 @@ TYPED_TEST(SVSTieredIndexTest, CreateIndexInstance) { ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 0); // Submit the index update job. - tiered_index->scheduleSVSIndexUpdate(); + tiered_index->scheduleSVSIndexInit(); ASSERT_EQ(mock_thread_pool.jobQ.size(), mock_thread_pool.thread_pool_size); // Execute the job from the queue and validate that the index was updated properly. @@ -671,14 +671,16 @@ TYPED_TEST(SVSTieredIndexTest, addVector) { ASSERT_LE(expected_mem, tiered_index->getAllocationSize()); if constexpr (TypeParam::isMulti()) { - // Add another vector under the same label + // Add another vector under the same label. The backend init was already submitted + // by the first vector (thread_pool_size batch-init jobs), so this second vector takes + // the per-vector async path and schedules one additional SVSInsertJob of its own. VecSimIndex_AddVector(tiered_index, vector, vec_label); ASSERT_EQ(tiered_index->indexSize(), 2); ASSERT_EQ(tiered_index->indexLabelCount(), 1); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 0); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 2); - // Validate that there still 1 update jobs set - ASSERT_EQ(mock_thread_pool.jobQ.size(), mock_thread_pool.thread_pool_size); + // The batch-init jobs plus the extra per-vector insert job. + ASSERT_EQ(mock_thread_pool.jobQ.size(), mock_thread_pool.thread_pool_size + 1); } } @@ -728,6 +730,12 @@ TYPED_TEST(SVSTieredIndexTest, background_indexing_check) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } + // The backgroundIndexing flag tracks the batch update job only; individual async + // SVS_INSERT_VECTOR_JOBs may still be in flight, each transiently holding its vector in + // both the flat and backend indices. Drain the queue so the total size is stable before + // asserting the exact count. + mock_thread_pool.thread_pool_join(); + ASSERT_GT(tiered_index->GetBackendIndex()->indexSize(), training_th + second_batch / update_th); ASSERT_LT(tiered_index->GetFlatIndex()->indexSize(), update_th); ASSERT_EQ(tiered_index->indexSize(), second_batch + training_th); @@ -1194,8 +1202,9 @@ TYPED_TEST(SVSTieredIndexTest, deleteVector) { ASSERT_EQ(tiered_index->indexSize(), 1); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 1); - // Move the vector to SVS by executing the insert job. - mock_thread_pool.thread_iteration(); + // Move the vector to SVS by executing the pending jobs. + while (mock_thread_pool.jobQ.size() > 0) + mock_thread_pool.thread_iteration(); ASSERT_EQ(tiered_index->indexLabelCount(), 1); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 1); // Scalar quantization accuracy is insufficient for this check. @@ -1236,7 +1245,7 @@ TYPED_TEST(SVSTieredIndexTestBasic, markedDeleted) { ASSERT_EQ(tiered_index->getNumMarkedDeleted(), 0); // Move vectors to the backend - mock_thread_pool.thread_iteration(); + while (mock_thread_pool.jobQ.size() > 0) mock_thread_pool.thread_iteration(); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), n); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); ASSERT_EQ(tiered_index->indexSize(), n); @@ -1318,6 +1327,8 @@ TYPED_TEST(SVSTieredIndexTestBasic, deleteVectorMulti) { ASSERT_EQ(tiered_index->deleteVector(vec_label), 2); ASSERT_EQ(tiered_index->indexSize(), 0); ASSERT_EQ(tiered_index->indexLabelCount(), 0); + + mock_thread_pool.thread_iteration(); mock_thread_pool.thread_iteration(); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); @@ -1330,13 +1341,20 @@ TYPED_TEST(SVSTieredIndexTestBasic, deleteVectorMulti) { ASSERT_EQ(tiered_index->indexSize(), 2); ASSERT_EQ(tiered_index->deleteVector(vec_label), 2); ASSERT_EQ(tiered_index->indexLabelCount(), 0); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 2); + mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 1); mock_thread_pool.thread_iteration(); ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); // Test deleting a label for which both of its vector's is in SVS index. GenerateAndAddVector(tiered_index, dim, vec_label, vec_label); GenerateAndAddVector(tiered_index, dim, vec_label, other_vec_val); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 2); mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 1); + mock_thread_pool.thread_iteration(); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); ASSERT_EQ(tiered_index->indexLabelCount(), 1); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 2); @@ -1730,7 +1748,7 @@ TYPED_TEST(SVSTieredIndexTest, parallelInsertAdHoc) { tiered_index->submitSingleJob(search_job); } - tiered_index->scheduleSVSIndexUpdate(); + tiered_index->scheduleSVSIndexInit(); mock_thread_pool.thread_pool_join(); EXPECT_EQ(successful_searches, n); @@ -2752,7 +2770,8 @@ TYPED_TEST(SVSTieredIndexTestBasic, overwriteVectorBasic) { ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(0, overwritten_vec), 0); // Ingest the updated vector to SVS. - mock_thread_pool.thread_iteration(); + while (mock_thread_pool.jobQ.size() > 0) + mock_thread_pool.thread_iteration(); ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 1); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); ASSERT_EQ(tiered_index->indexLabelCount(), 1); @@ -3031,12 +3050,13 @@ TYPED_TEST(SVSTieredIndexTest, writeInPlaceMode) { ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 2); ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); - // Overwrite inplace - only in single-value mode + // In write-in-place mode every mutation runs synchronously, including consolidation + // of soft-deleted slots. size_t expected_marked_deleted = 0; + // Overwrite inplace - only in single-value mode if (!TypeParam::isMulti()) { TEST_DATA_T overwritten_vec[] = {1, 1, 1, 1}; tiered_index->addVector(overwritten_vec, vec_label); - expected_marked_deleted++; ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 3); ASSERT_EQ(tiered_index->indexSize(), 3); ASSERT_EQ(tiered_index->indexLabelCount(), 2); @@ -3044,9 +3064,9 @@ TYPED_TEST(SVSTieredIndexTest, writeInPlaceMode) { ASSERT_EQ(tiered_index->getDistanceFrom_Unsafe(vec_label, overwritten_vec), 0); ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), expected_marked_deleted); } - // Validate that the vector is marked as deleted. + // Validate that the vector is deleted and consolidated synchronously (in-place mode), + // so no marked-deleted entry remains. tiered_index->deleteVector(vec_label); - expected_marked_deleted++; ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), TypeParam::isMulti() ? 2 : 3); ASSERT_EQ(tiered_index->indexLabelCount(), 1); @@ -3242,11 +3262,12 @@ TYPED_TEST(SVSTieredIndexTestBasic, runGCAPI) { 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. + // Expected that a single GC job was 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. + // Expected that no new GC jobs were added to the queue (indexGCScheduled is still set until the + // pending jobs run). 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(); @@ -3515,7 +3536,7 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalSingle) { // update job paused, we have vectors 0-(n-1) in the index, let's do index modifications // Remove vector label=n-2, it is copied to backend index. - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 2), 2); + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 2), 1); // Update vector label=1. EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, 1, 10), 0); // Add a new vector @@ -3523,9 +3544,9 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalSingle) { // Add another one EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, n + 1, n + 1), 1); // Remove vector label=0, it is copied to backend index. - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 0), 2); + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 0), 1); // Remove the last vector copied to backend index - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 1), 2); + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 1), 1); // Update vector label=2. EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, 2, 20), 0); // Remove vector label=2. @@ -3547,16 +3568,15 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalSingle) { // 0:deleted, 1: 10, 2: deleted, 3:3, ..., n-2:deleted n-1: 10(n-1), n+1: n+1; // total: n-2 vectors and labels ASSERT_EQ(tiered_index->indexLabelCount(), n - 2); - EXPECT_EQ(tiered_index->GetBackendIndex()->indexLabelCount(), n - 5); + // Nothing remains in flat. + EXPECT_EQ(tiered_index->GetBackendIndex()->indexLabelCount(), n - 2); - // We added 3 vectors to the flat index and removed 5 vectors from the backend index. - // Backend index: 0:deleted, 1:deleted, 2:deleted, 3:3, ..., n-2:deleted, n-1:deleted; - // total: n-5 - EXPECT_EQ(tiered_index->GetBackendIndex()->indexSize(), n); - ASSERT_EQ(tiered_index->getNumMarkedDeleted(), 5); - // Frontend index: 1:10, n-1:10(n-1), n+1:n+1 - ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 3); - ASSERT_EQ(tiered_index->indexSize(), n + tiered_index->GetFlatIndex()->indexSize()); + EXPECT_EQ(tiered_index->GetBackendIndex()->indexSize(), n - 2); + ASSERT_EQ(tiered_index->getNumMarkedDeleted(), 0); + // Flat buffer fully drained by the insert jobs. + ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); + ASSERT_EQ(tiered_index->indexSize(), tiered_index->GetBackendIndex()->indexSize() + + tiered_index->GetFlatIndex()->indexSize()); double abs_err = 1e-2; // Allow a larger relative error for quantization. TEST_DATA_T expected_vector[dim]; @@ -3569,7 +3589,7 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalSingle) { GenerateVector(expected_vector, dim, 0); ASSERT_TRUE(std::isnan(tiered_index->getDistanceFrom_Unsafe(n - 2, expected_vector))); - // Vector label=1, with value 10 should be in the flat index. + // Vector label=1 was updated to value 10; its latest value is retrievable from the index. GenerateVector(expected_vector, dim, 10); ASSERT_NEAR(tiered_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); @@ -3648,22 +3668,24 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalMulti) { // update job paused, we have vectors 0-(n-1) in the index, let's do index modifications - // Remove vector label=n-2, it is copied to backend index. - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 2), 2); + // Remove vector label=n-2. At the pause it lives only in the backend + //, so a single vector is removed. + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 2), 1); // Add one more vector label=1. EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, 1, 10), 1); // Add a new vector EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, n, n), 1); // Add another one EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, n + 1, n + 1), 1); - // Remove vector label=0, it is copied to backend index. - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 0), 2); - // Remove the last vector copied to backend index - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 1), 2); + // Remove vector label=0, only in the backend at the pause. + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 0), 1); + // Remove the last vector, only in the backend at the pause. + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n - 1), 1); // Add one more vector label=2. EXPECT_EQ(GenerateAndAddVector(tiered_index, dim, 2, 20), 1); - // Remove vector label=2: for multi: old is copied to backend , old + new are in flat - EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 2), 3); + // Remove vector label=2: for multi both the original (2) and the new (20) copies sit in + // the flat buffer at this point, so 2 vectors are removed. + EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, 2), 2); // Remove vector label=n - in flat only EXPECT_EQ(VecSimIndex_DeleteVector(tiered_index, n), 1); // Add vector (n-1) again @@ -3681,15 +3703,16 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalMulti) { // 0: deleted, 1: (1,10), 2: deleted, 3:3, ..., n-2: deleted n-1: 10(n-1), n+1: n+1; // total: n-2 labels, n-1 vectors ASSERT_EQ(tiered_index->indexLabelCount(), n - 2); - EXPECT_EQ(tiered_index->GetBackendIndex()->indexLabelCount(), n - 4); + // In the async-insert design the per-vector insert jobs drain every surviving vector from + // the flat buffer into the backend, so nothing remains in flat. + EXPECT_EQ(tiered_index->GetBackendIndex()->indexLabelCount(), n - 2); - // We added 3 vectors to the flat index and removed 4 vectors from the backend index. - // Backend index: 0:deleted, 1:1, 2:deleted, 3:3, ..., n-2:deleted, n-1:deleted; total: n-4 - EXPECT_EQ(tiered_index->GetBackendIndex()->indexSize(), n); - ASSERT_EQ(tiered_index->getNumMarkedDeleted(), 4); - // Frontend index: 1:10, n-1:10(n-1), n+1:n+1 - ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 3); - ASSERT_EQ(tiered_index->indexSize(), n + tiered_index->GetFlatIndex()->indexSize()); + EXPECT_EQ(tiered_index->GetBackendIndex()->indexSize(), n - 1); + ASSERT_EQ(tiered_index->getNumMarkedDeleted(), 0); + // Flat buffer fully drained by the insert jobs. + ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); + ASSERT_EQ(tiered_index->indexSize(), tiered_index->GetBackendIndex()->indexSize() + + tiered_index->GetFlatIndex()->indexSize()); double abs_err = 1e-2; // Allow a larger relative error for quantization. TEST_DATA_T expected_vector[dim]; @@ -3702,11 +3725,11 @@ TYPED_TEST(SVSTieredIndexTestBasic, testSwapJournalMulti) { GenerateVector(expected_vector, dim, 0); ASSERT_TRUE(std::isnan(tiered_index->getDistanceFrom_Unsafe(n - 2, expected_vector))); - // There are 2 vectors labeled "1" with values 1 in backend and 10 in flat. - // We expect the minimal distance for the query 10 to be taken from flat index. + // Label "1" is multi-valued and keeps both of its vectors (values 1 and 10). + // The minimal distance for the query 10 should match the second value. GenerateVector(expected_vector, dim, 10); ASSERT_NEAR(tiered_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); - // And the minimal distance for the query 1.0 to be taken from backend + // And the minimal distance for the query 1.0 should match the first value. GenerateVector(expected_vector, dim, 1); ASSERT_NEAR(tiered_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); @@ -3806,8 +3829,9 @@ TYPED_TEST(SVSTieredIndexTestBasic, testDeletedJournalSingle) { mock_thread_pool.thread_pool_join(); - // Verify that vectors labels: {0, 1, 2, n-1} are marked as deleted in the SVS index. - ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 4); + // The update job is paused before it transfers vectors to the backend, so at the pause the + // backend is still empty. + ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0); // Verify that the deleted vectors are not accessible. double abs_err = 1e-2; // Allow a larger relative error for quantization. @@ -3824,9 +3848,10 @@ TYPED_TEST(SVSTieredIndexTestBasic, testDeletedJournalSingle) { ASSERT_TRUE(std::isnan(tiered_index->getDistanceFrom_Unsafe(label, expected_vector))); } - // label 1 - updated to 10 but deleted in the SVS index - ASSERT_TRUE(flat_index->isLabelExists(1)); - ASSERT_FALSE(svs_index->isLabelExists(1)); + // label 1 - updated to 10 during the pause. The insert job drains it from the flat buffer + // into the backend, so it now lives in the SVS index (not flat) with its updated value. + ASSERT_FALSE(flat_index->isLabelExists(1)); + ASSERT_TRUE(svs_index->isLabelExists(1)); GenerateVector(expected_vector, dim, 10); ASSERT_NEAR(tiered_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); } @@ -3908,8 +3933,9 @@ TYPED_TEST(SVSTieredIndexTestBasic, testDeletedJournalMulti) { mock_thread_pool.thread_pool_join(); - // Verify that vectors labels: {0, 2, n-1} are marked as deleted in the SVS index. - ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 3); + // The update job is paused before it transfers vectors to the backend, so at the pause the + // backend is still empty. + ASSERT_EQ(tiered_index->GetSVSIndex()->getNumMarkedDeleted(), 0); // Verify that the deleted vectors are not accessible. double abs_err = 1e-2; // Allow a larger relative error for quantization. @@ -3926,13 +3952,14 @@ TYPED_TEST(SVSTieredIndexTestBasic, testDeletedJournalMulti) { ASSERT_TRUE(std::isnan(tiered_index->getDistanceFrom_Unsafe(label, expected_vector))); } - // label 1 - multi-value 1 (in SVS) and 10 (in flat) - ASSERT_TRUE(flat_index->isLabelExists(1)); + // label 1 - multi-value with both vectors (1 and 10). The insert jobs drain the flat buffer + // into the backend, so both values now live in the SVS index and none remain in flat. + ASSERT_FALSE(flat_index->isLabelExists(1)); ASSERT_TRUE(svs_index->isLabelExists(1)); GenerateVector(expected_vector, dim, 1); ASSERT_NEAR(backend_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); GenerateVector(expected_vector, dim, 10); - ASSERT_NEAR(flat_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); + ASSERT_NEAR(backend_index->getDistanceFrom_Unsafe(1, expected_vector), 0, abs_err); } TEST(SVSTieredIndexTest, testThreadPool) {