diff --git a/cmake/svs.cmake b/cmake/svs.cmake index fc5a3c91a..9496c722d 100644 --- a/cmake/svs.cmake +++ b/cmake/svs.cmake @@ -64,7 +64,7 @@ if(USE_SVS) if (CMAKE_CXX_COMPILER_ID MATCHES "Clang") if(GLIBC_VERSION VERSION_GREATER_EQUAL "2.31" AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "21.0") if(GLIBCXX_VERSION VERSION_GREATER_EQUAL "11") - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced-clang21-gcc11.tar.gz" CACHE STRING "SVS URL") + set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-clang21-gcc11-pr330.tar.gz" CACHE STRING "SVS URL") else() message(STATUS "libstdc++ >= GCC 11 is required for Clang SVS binaries - disabling SVS_SHARED_LIB") set(SVS_SHARED_LIB OFF) @@ -76,14 +76,12 @@ if(USE_SVS) else() if(GLIBC_VERSION VERSION_GREATER_EQUAL "2.28") if(CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "14.0") - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced-gcc14.tar.gz" CACHE STRING "SVS URL") + set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-gcc14-pr330.tar.gz" CACHE STRING "SVS URL") else() - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.1/svs-shared-library-reduced.tar.gz" CACHE STRING "SVS URL") + set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-reduced-pr330.tar.gz" CACHE STRING "SVS URL") endif() - elseif(GLIBC_VERSION VERSION_GREATER_EQUAL "2.26") - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/v0.3.0/svs-shared-library-0.3.0-reduced-glibc2_26.tar.gz" CACHE STRING "SVS URL") else() - message(STATUS "GLIBC >= 2.26 is required for GCC SVS binaries - disabling SVS_SHARED_LIB") + message(STATUS "GLIBC >= 2.28 is required for GCC SVS binaries - disabling SVS_SHARED_LIB") set(SVS_SHARED_LIB OFF) endif() endif() diff --git a/deps/ScalableVectorSearch b/deps/ScalableVectorSearch index 7786d43b9..6338b8cd8 160000 --- a/deps/ScalableVectorSearch +++ b/deps/ScalableVectorSearch @@ -1 +1 @@ -Subproject commit 7786d43b98ac9769ad7668d0d4896143cfb2f167 +Subproject commit 6338b8cd888c5176e7efb075cbae1faa2b38c3f8 diff --git a/src/VecSim/algorithms/svs/svs.h b/src/VecSim/algorithms/svs/svs.h index 7dc15dc0d..51db74e82 100644 --- a/src/VecSim/algorithms/svs/svs.h +++ b/src/VecSim/algorithms/svs/svs.h @@ -57,6 +57,10 @@ struct SVSIndexBase virtual std::unique_ptr createImpl(const void *vectors_data, const labelType *labels, size_t n) = 0; virtual void setImpl(std::unique_ptr impl) = 0; + + virtual std::unique_ptr + makeAddVectorsChanges(const void *vectors_data, const labelType *labels, size_t n) = 0; + virtual int applyAddVectorsChanges(std::unique_ptr changes) = 0; #ifdef BUILD_TESTS virtual svs::logging::logger_ptr getLogger() const = 0; #endif @@ -252,6 +256,68 @@ class SVSIndex : public VecSimIndexAbstract, fl this->impl_ = std::move(svs_handler->impl); } + struct SVSChangesHandler : public ImplHandler { + using points_type = svs::data::SimpleDataView; + using ids_type = std::span; + using changes_type = decltype(std::declval().add_points_compute_changes( + std::declval(), std::declval())); + + template + SVSChangesHandler(MemoryUtils::unique_blob processed_blob, points_type points, ids_type ids, + const Impl &impl, int result_num) + : processed_blob_{std::move(processed_blob)}, points_{std::move(points)}, + ids_{std::move(ids)}, changes_{impl.add_points_compute_changes(points_, ids_)}, + result_num_{result_num} {} + virtual ~SVSChangesHandler() = default; + + MemoryUtils::unique_blob processed_blob_; + points_type points_; + ids_type ids_; + changes_type changes_; + int result_num_; + }; + + virtual std::unique_ptr + makeAddVectorsChanges(const void *vectors_data, const labelType *labels, size_t n) override { + assert(impl_ != nullptr); + if (n == 0) { + return nullptr; + } + + int deleted_num = 0; + std::span ids(labels, n); + if constexpr (!isMulti) { + // SVS index does not support overriding vectors with the same label + // and we cannot delete vectors now, so these had to be deleted in advance. + // We expect that the caller (SVSTiered) has already deleted any existing labels from + // SVS index before calling this function inside addVector(). so just use assert() to + // check that no existing labels are present in the input. + assert(!std::any_of(ids.begin(), ids.end(), + [this](labelType label) { return impl_->has_id(label); })); + } + + auto processed_blob = this->preprocessForBatchStorage(vectors_data, n); + auto typed_vectors_data = static_cast(processed_blob.get()); + // Wrap data into SVS SimpleDataView for SVS API + auto points = svs::data::SimpleDataView{typed_vectors_data, n, this->dim}; + return std::make_unique(std::move(processed_blob), points, ids, *impl_, + n - deleted_num); + } + + virtual int applyAddVectorsChanges(std::unique_ptr changes) override { + assert(impl_ != nullptr); + if (!changes) { + return 0; + } + + SVSChangesHandler *svs_changes_handler = dynamic_cast(changes.get()); + if (!svs_changes_handler) { + throw std::logic_error("Failed to cast to SVSChangesHandler"); + } + impl_->add_points_commit(svs_changes_handler->ids_, svs_changes_handler->changes_); + return svs_changes_handler->result_num_; + } + // Assuming parallelism was updated to reflect the number of available threads before this // function was called. // This function assumes that the caller has already set parallelism to the appropriate value @@ -587,29 +653,58 @@ class SVSIndex : public VecSimIndexAbstract, fl 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; - } + if constexpr (isMulti) { + auto query = svs::data::ConstSimpleDataView{ + static_cast(processed_query), 1, this->dim}; + auto result = svs::QueryResult{query.size(), k}; + 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))}); + } + } else { + // Use the single-query search path of MutableVamanaIndex to avoid the + // parallel batch-search overhead for a single query. + auto scratch = impl_->scratchspace(sp); + // Ensure the scratch buffer can hold at least `k` neighbors. + if (scratch.buffer.target_capacity() < k) { + scratch.buffer.change_maxsize(k); + } - for (size_t i = 0; i < n_neighbors; i++) { - rep->results.push_back( - VecSimQueryResult{result.index(0, i), toVecSimDistance(result.distance(0, i))}); + auto query = std::span{static_cast(processed_query), + this->dim}; + impl_->search(query, scratch, cancel); + if (cancel()) { + rep->code = VecSim_QueryReply_TimedOut; + return rep; + } + + const auto &buffer = scratch.buffer; + const auto n_neighbors = std::min(k, buffer.size()); + rep->results.reserve(n_neighbors); + + for (size_t i = 0; i < n_neighbors; i++) { + const auto neighbor = buffer[i]; + rep->results.push_back( + VecSimQueryResult{impl_->translate_internal_id(neighbor.id()), + toVecSimDistance(neighbor.distance())}); + } } // Workaround for VecSim merge_results() that expects results to be sorted // by score, then by id from both indices. diff --git a/src/VecSim/algorithms/svs/svs_tiered.h b/src/VecSim/algorithms/svs/svs_tiered.h index 535920365..c7ebf5457 100644 --- a/src/VecSim/algorithms/svs/svs_tiered.h +++ b/src/VecSim/algorithms/svs/svs_tiered.h @@ -711,13 +711,14 @@ class TieredSVSIndex : public VecSimTieredIndex { std::lock_guard lock(this->mainIndexGuard); svs_index->setImpl(std::move(impl)); } else { + svs_index->setParallelism(std::min(availableThreads, labels_to_move.size())); // Backend index is initialized - just add the vectors. + auto changes = svs_index->makeAddVectorsChanges( + vectors_to_move.data(), labels_to_move.data(), labels_to_move.size()); + // Upgrade to unique lock to add 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()); + svs_index->applyAddVectorsChanges(std::move(changes)); } } executeTracingCallback("UpdateJob::after_add_to_svs"); diff --git a/tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h b/tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h index 225758018..1fa8b046a 100644 --- a/tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h +++ b/tests/benchmark/bm_initialization/bm_basics_svs_initialize_fp32.h @@ -68,3 +68,19 @@ BENCHMARK_REGISTER_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopK)) ->Args({200, 100}) ->Args({500, 500}) ->ArgNames({"window_size", "k"}); + +// Parallel TopK searches running concurrently with a background update job. +// Uses constant window_size=200 and k=100. +// {update_threshold, n_parallel_searches, thread_count} +BENCHMARK_TEMPLATE_DEFINE_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopKSearchDuringUpdate), + DATA_TYPE_INDEX_T) +(benchmark::State &st) { TopKSearchDuringUpdate(st); } +BENCHMARK_REGISTER_F(BM_VecSimSVS, BM_FUNC_NAME(BM_TopKSearchDuringUpdate)) + ->Unit(benchmark::kMillisecond) + ->Iterations(1) + ->ArgsProduct({{static_cast(BM_VecSimGeneral::block_size), + static_cast(10 * BM_VecSimGeneral::block_size)}, + {10, 50}, + {2, 4}}) + ->ArgNames({"update_threshold", "n_parallel_searches", "thread_count"}) + ->UseRealTime(); diff --git a/tests/benchmark/bm_vecsim_svs.h b/tests/benchmark/bm_vecsim_svs.h index 852973a50..57c0410cf 100644 --- a/tests/benchmark/bm_vecsim_svs.h +++ b/tests/benchmark/bm_vecsim_svs.h @@ -58,6 +58,11 @@ class BM_VecSimSVS : public BM_VecSimGeneral { // registration. void TopK_SVS(benchmark::State &st); + // Runs a number of parallel TopK searches while a background update job (moving vectors from + // the flat buffer into the SVS backend index) is in progress. Measures the time to complete + // all searches concurrently with the update. + void TopKSearchDuringUpdate(benchmark::State &st); + private: static const char *svs_index_tar_file; static std::string base_path; @@ -492,6 +497,68 @@ void BM_VecSimSVS::TopK_SVS(benchmark::State &st) { VecSimIndex_Free(index); } +template +void BM_VecSimSVS::TopKSearchDuringUpdate(benchmark::State &st) { + // ensure mode is async + ASSERT_EQ(VecSimIndexInterface::asyncWriteMode, VecSim_WriteAsync); + + const size_t window_size = 200; + const size_t k = 100; + size_t update_threshold = st.range(0); + size_t n_parallel_searches = st.range(1); + int unsigned num_threads = st.range(2); + + if (num_threads > std::thread::hardware_concurrency()) { + GTEST_SKIP() << "Not enough threads available, skipping test..."; + } + + // Ensure we have enough vectors to fill the flat buffer and to run the searches. + ASSERT_GE(N_QUERIES, update_threshold); + + auto mock_thread_pool = tieredIndexMock(num_threads); + ASSERT_EQ(mock_thread_pool.thread_pool_size, num_threads); + auto *tiered_index = CreateTieredSVSIndexFromFile(mock_thread_pool, update_threshold); + + // Fill the flat buffer up to just below the update threshold, so a single AddVector will + // trigger the background update job. + for (size_t i = 0; i < update_threshold - 1; ++i) { + int ret = VecSimIndex_AddVector(tiered_index, test_vectors[i].data(), i + N_VECTORS); + ASSERT_EQ(ret, 1); + } + + mock_thread_pool.init_threads(); + + for (auto _ : st) { + // Trigger the background update job by reaching the update threshold. It runs on the + // index's own thread pool. + int ret = VecSimIndex_AddVector(tiered_index, test_vectors[update_threshold - 1].data(), + update_threshold - 1 + N_VECTORS); + ASSERT_EQ(ret, 1); + + // Run the TopK searches on independent threads, asynchronously to the index thread pool, + // so they execute concurrently with the ongoing background update. + std::vector search_threads; + search_threads.reserve(n_parallel_searches); + for (size_t i = 0; i < n_parallel_searches; ++i) { + search_threads.emplace_back([tiered_index, i, window_size, k]() { + SVSRuntimeParams svs_params = {.windowSize = window_size}; + VecSimQueryParams query_params = {.svsRuntimeParams = svs_params}; + auto results = VecSimIndex_TopKQuery( + tiered_index, test_vectors[i % N_QUERIES].data(), k, &query_params, BY_SCORE); + VecSimQueryReply_Free(results); + }); + } + for (auto &t : search_threads) { + t.join(); + } + + // Wait for the background update to complete. + mock_thread_pool.thread_pool_wait(); + } + + mock_thread_pool.thread_pool_join(); +} + #define UNIT_AND_ITERATIONS Unit(benchmark::kMillisecond)->Iterations(2) #if HAVE_SVS_LVQ