diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 88cff6e32..18fb8c61a 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -33,6 +33,8 @@ set(SVS_C_API_HEADERS set(SVS_C_API_SOURCES src/algorithm.hpp + src/dispatcher_concurrent_vamana.hpp + src/dispatcher_dynamic_vamana_copy.hpp src/error.hpp src/filtered_search.hpp src/index.hpp @@ -49,6 +51,7 @@ set(SVS_C_API_SOURCES src/svs_c.cpp src/dispatcher_vamana.cpp src/dispatcher_dynamic_vamana.cpp + src/dispatcher_concurrent_vamana.cpp src/leanvec_training_data.cpp src/data_builder.cpp src/index_builder.cpp diff --git a/bindings/c/docs/C_API_Design.md b/bindings/c/docs/C_API_Design.md index fcd14c5a2..aa4fdb276 100644 --- a/bindings/c/docs/C_API_Design.md +++ b/bindings/c/docs/C_API_Design.md @@ -277,12 +277,20 @@ Invalid parameters are rejected with `SVS_ERROR_INVALID_ARGUMENT`; unlike the le |-----------------|----------| | `SVS_SYNC_KIND_NONE` | No internal synchronization (default); the caller must serialize access | | `SVS_SYNC_KIND_GLOBAL` | Index-wide reader/writer lock: read-only operations (search, `has_id`, `get_distance`, reconstruct, `get_size`, `get_num_threads`, memory queries, use as a conversion source) run concurrently; mutating operations (add/delete points, consolidate, compact, save, `set_num_threads`) are exclusive | -| `SVS_SYNC_KIND_FINE_GRAIN` | Currently behaves as `SVS_SYNC_KIND_GLOBAL`; reserved for finer-grained locking in future releases | +| `SVS_SYNC_KIND_FINE_GRAIN` | Backed by the concurrent Vamana index, which synchronizes updates itself: read-only operations and add/delete points, consolidate, compact all take a shared lock and run in parallel (compact still waits inside the index). Save, `set_num_threads`, memory queries and use as a conversion source are exclusive | Synchronization limits: - `svs_index_free()` must not run concurrently with any other call on the same handle. - Each thread must use its own `svs_search_results_t` and `svs_error_h`. -- The lock gives no writer priority; sustained read load may starve writers. +- The lock gives no writer priority; sustained read load may starve operations that + need an exclusive lock. +- With `SVS_SYNC_KIND_FINE_GRAIN`, add/delete counts and the index size are a snapshot + and may be affected by concurrent updates. +- The concurrent index keeps a reverse-edge list per node; its memory is reported (and + estimated, when `sync_kind` is passed to + `svs_index_builder_estimate_memory_dynamic_ex`) as part of `graph_bytes`. +- Conversion (`svs_index_convert_dynamic_ex`) works between any sync kinds; deleted but + not yet consolidated vectors are not copied. - The sync kind is not persisted by `svs_index_save()`; pass it again on load. Query it with `svs_index_dynamic_get_sync_kind()`. @@ -325,7 +333,8 @@ plan capacity up front: for the block size; `svs_index_builder_get_default_blocksize_bytes` returns the default block size (`blocksize_bytes = 0` selects it). `svs_index_builder_estimate_memory_dynamic_ex` takes the block size from - `svs_dynamic_index_params_t` instead, including `blocksize_elements`. + `svs_dynamic_index_params_t` instead, including `blocksize_elements`, and accounts + for the reverse edges of the `SVS_SYNC_KIND_FINE_GRAIN` index. - `svs_index_builder_estimate_search_memory` / `svs_index_builder_estimate_search_memory_dynamic` / `svs_index_builder_estimate_search_memory_dynamic_ex` — estimate the scratch memory a diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h index 7380d173f..2d2b87559 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -588,23 +588,33 @@ struct svs_memory_breakdown { /// @brief Dynamic index synchronization kind. /// /// With synchronization enabled, the dynamic index handle guards itself with an internal -/// reader/writer lock: read-only operations (search, has_id, get_distance, reconstruct, -/// get_size, get_num_threads, memory queries, and use as a conversion source) take a -/// shared lock and may run concurrently, while mutating operations (add_points, -/// delete_points, consolidate, compact, save, set_num_threads) take an exclusive lock. +/// reader/writer lock. Read-only operations (search, has_id, get_distance, reconstruct, +/// get_size, get_num_threads) always take a shared lock and may run concurrently. +/// +/// With SVS_SYNC_KIND_GLOBAL, mutating operations (add_points, delete_points, +/// consolidate, compact, save, set_num_threads) take an exclusive lock. +/// +/// With SVS_SYNC_KIND_FINE_GRAIN, the handle is backed by a concurrent index that +/// synchronizes updates itself: add_points, delete_points, consolidate and compact take a +/// shared lock and run alongside searches and each other (compact still waits for +/// in-flight operations inside the index). Only save, set_num_threads, memory queries +/// and use as a conversion source take an exclusive lock. /// /// @remarks Synchronization covers operations on the index only: /// * svs_index_free() must not run concurrently with any other call on the same handle. /// * Each thread must use its own svs_search_results_t and svs_error_h. /// * The lock gives no writer priority: under sustained read load, readers may starve -/// writers. +/// operations that need an exclusive lock. +/// * With FINE_GRAIN, counts returned by add_points/delete_points and the index size +/// reflect a point in time and may be affected by concurrent updates. +/// * With FINE_GRAIN, search, add_points and delete_points accept at most one query, +/// vector or ID per call; larger batches fail with SVS_ERROR_NOT_IMPLEMENTED. /// * The sync kind is a runtime property: it is not persisted by svs_index_save(), so /// pass it again to svs_index_load_dynamic_ex(). enum svs_sync_kind { SVS_SYNC_KIND_NONE = 0, ///< No internal synchronization; caller must synchronize SVS_SYNC_KIND_GLOBAL = 1, ///< Index-wide reader/writer lock - SVS_SYNC_KIND_FINE_GRAIN = 2 ///< Currently the same as GLOBAL; reserved for - ///< finer-grained locking + SVS_SYNC_KIND_FINE_GRAIN = 2 ///< Concurrent index; updates run in parallel }; /// @brief Structure to hold dynamic index parameters. @@ -1165,7 +1175,8 @@ SVS_API bool svs_index_builder_estimate_memory_dynamic( /// @param builder The index builder handle /// @param num_vectors The number of vectors to be indexed /// @param params Pointer to the dynamic index parameters structure, or NULL for defaults; -/// only the block size fields are used +/// the block size fields and @p sync_kind are used (SVS_SYNC_KIND_FINE_GRAIN adds the +/// concurrent index's reverse edges to graph_bytes) /// @param out_breakdown Pointer to a structure to hold the memory breakdown /// @param out_err An optional error handle to capture errors /// @return true on success, false on failure @@ -1477,6 +1488,8 @@ SVS_API bool svs_index_get_size( /// @param id_filter The ID filter interface (can be NULL for no filtering) /// @param out_err An optional error handle to capture errors /// @return true on success, false on failure +/// @error SVS_ERROR_NOT_IMPLEMENTED if @p index is a dynamic index with +/// SVS_SYNC_KIND_FINE_GRAIN and @p num_queries is greater than 1 SVS_API bool svs_index_search_topk( svs_index_h index, const float* queries, @@ -1554,6 +1567,8 @@ SVS_API bool svs_index_save_stream( /// vectors /// @param out_err An optional error handle to capture errors /// @return true on success, false on failure +/// @error SVS_ERROR_NOT_IMPLEMENTED if @p index uses SVS_SYNC_KIND_FINE_GRAIN and +/// @p num_vectors is greater than 1 SVS_API bool svs_index_dynamic_add_points( svs_index_h index, const float* new_points, @@ -1571,6 +1586,8 @@ SVS_API bool svs_index_dynamic_add_points( /// vectors /// @param out_err An optional error handle to capture errors /// @return true on success, false on failure +/// @error SVS_ERROR_NOT_IMPLEMENTED if @p index uses SVS_SYNC_KIND_FINE_GRAIN and +/// @p num_ids is greater than 1 SVS_API bool svs_index_dynamic_delete_points( svs_index_h index, const size_t* ids, diff --git a/bindings/c/src/allocator.hpp b/bindings/c/src/allocator.hpp index 9b09c700d..d191ac135 100644 --- a/bindings/c/src/allocator.hpp +++ b/bindings/c/src/allocator.hpp @@ -19,6 +19,7 @@ #include "error.hpp" +#include #include #include #include @@ -28,9 +29,27 @@ namespace svs { namespace c_runtime { -template > -using MaybeBlockedAlloc = - std::conditional_t, Allocator>; +/// Storage layout of a dataset or graph. +enum class BlockKind { + None, ///< Single contiguous allocation (static index). + Blocked, ///< data::Blocked (regular dynamic index). + Segmented ///< concurrent::SegmentedBlocked (concurrent dynamic index). +}; + +namespace detail { +template struct BlockedAllocSelector { + using type = Allocator; +}; +template struct BlockedAllocSelector { + using type = svs::data::Blocked; +}; +template struct BlockedAllocSelector { + using type = svs::index::vamana::concurrent::SegmentedBlocked; +}; +} // namespace detail + +template > +using MaybeBlockedAlloc = typename detail::BlockedAllocSelector::type; class AllocatorBuilder { template class CustomAllocator { diff --git a/bindings/c/src/data_builder.cpp b/bindings/c/src/data_builder.cpp index 1cc1e0607..8eea99f98 100644 --- a/bindings/c/src/data_builder.cpp +++ b/bindings/c/src/data_builder.cpp @@ -25,7 +25,7 @@ #include namespace svs::c_runtime { -// namespace { +namespace { template size_t estimate_size(DataBuilder builder, size_t num_vectors, size_t dimension, svs::lib::Empty) { @@ -59,19 +59,13 @@ void register_data_size_specializations(Dispatcher& dispatcher) { dispatcher.register_target(&estimate_size); }; - for_simple_specializations(size_closure); - for_leanvec_specializations(size_closure); - for_lvq_specializations(size_closure); - for_sq_specializations(size_closure); + for_all_specializations(size_closure); auto blocked_size_closure = [&dispatcher]() { dispatcher.register_target(&estimate_blocked_size); }; - for_simple_specializations(blocked_size_closure); - for_leanvec_specializations(blocked_size_closure); - for_lvq_specializations(blocked_size_closure); - for_sq_specializations(blocked_size_closure); + for_all_specializations(blocked_size_closure); } using BlocksizeArg = std::variant; @@ -95,7 +89,7 @@ size_t dispatch_data_size_estimation( storage, num_vectors, dimension, std::move(block_params) ); } -//} // namespace +} // namespace size_t estimate_data_size(const Storage* storage, size_t num_vectors, size_t dimension) { if (storage == nullptr) { diff --git a/bindings/c/src/data_builder.hpp b/bindings/c/src/data_builder.hpp index d341352b0..cd8d5d27f 100644 --- a/bindings/c/src/data_builder.hpp +++ b/bindings/c/src/data_builder.hpp @@ -22,6 +22,14 @@ #include "storage.hpp" namespace svs::c_runtime { + +template void for_all_specializations(F&& f) { + for_simple_specializations(std::forward(f)); + for_leanvec_specializations(std::forward(f)); + for_lvq_specializations(std::forward(f)); + for_sq_specializations(std::forward(f)); +} + size_t estimate_data_size(const Storage* storage, size_t num_vectors, size_t dimension); size_t estimate_data_size_blocked( const Storage* storage, diff --git a/bindings/c/src/data_builder/leanvec.hpp b/bindings/c/src/data_builder/leanvec.hpp index 62bad86da..2550461be 100644 --- a/bindings/c/src/data_builder/leanvec.hpp +++ b/bindings/c/src/data_builder/leanvec.hpp @@ -15,10 +15,11 @@ */ #pragma once -#ifdef SVS_RUNTIME_ENABLE_LVQ_LEANVEC - #include "svs/c/svs_c.h" +#include "allocator.hpp" + +#ifdef SVS_RUNTIME_ENABLE_LVQ_LEANVEC #include "data_builder/lvq.hpp" #include "leanvec_training_data.hpp" #include "storage.hpp" @@ -41,6 +42,8 @@ #include #endif // SVS_LEANVEC_HEADER +#include + #include #include #include @@ -172,10 +175,9 @@ struct lib:: } }; -template void for_leanvec_specializations(F&& f) { - using byte_alloc = svs::c_runtime:: - MaybeBlockedAlloc>; - +template void for_leanvec_specializations(F&& f) { + using byte_alloc = + svs::c_runtime::MaybeBlockedAlloc>; #define X(P, S, D) f.template operator(), D>(); #define XX(P, S) X(P, S, DistanceL2) X(P, S, DistanceIP) X(P, S, DistanceCosineSimilarity) // Pattern: @@ -192,7 +194,7 @@ template void for_leanvec_specializations(F&& f) { #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled namespace svs { // Define empty stubs for LeanVec-related functions when LVQ/LeanVec support is disabled -template void for_leanvec_specializations(F&&) {} +template void for_leanvec_specializations(F&&) {} } // namespace svs #endif // SVS_RUNTIME_ENABLE_LVQ_LEANVEC diff --git a/bindings/c/src/data_builder/lvq.hpp b/bindings/c/src/data_builder/lvq.hpp index f063b451c..116760bf2 100644 --- a/bindings/c/src/data_builder/lvq.hpp +++ b/bindings/c/src/data_builder/lvq.hpp @@ -15,10 +15,11 @@ */ #pragma once -#ifdef SVS_RUNTIME_ENABLE_LVQ_LEANVEC - #include "svs/c/svs_c.h" +#include "allocator.hpp" + +#ifdef SVS_RUNTIME_ENABLE_LVQ_LEANVEC #include "storage.hpp" #include "types_support.hpp" @@ -39,6 +40,8 @@ #include #endif // SVS_LVQ_HEADER +#include + #include #include #include @@ -160,9 +163,9 @@ struct lib::DispatchConverter< static To convert(From SVS_UNUSED(from)) { return To{}; } }; -template void for_lvq_specializations(F&& f) { - using byte_alloc = svs::c_runtime:: - MaybeBlockedAlloc>; +template void for_lvq_specializations(F&& f) { + using byte_alloc = + svs::c_runtime::MaybeBlockedAlloc>; #define X(P, S, D) f.template operator(), D>(); #define XX(P, S) X(P, S, DistanceL2) X(P, S, DistanceIP) X(P, S, DistanceCosineSimilarity) // Pattern: @@ -180,7 +183,7 @@ template void for_lvq_specializations(F&& f) { #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled namespace svs { // Define empty stubs for LVQ-related functions when LVQ/LeanVec support is disabled -template void for_lvq_specializations(F&&) {} +template void for_lvq_specializations(F&&) {} } // namespace svs #endif // SVS_RUNTIME_ENABLE_LVQ_LEANVEC diff --git a/bindings/c/src/data_builder/simple.hpp b/bindings/c/src/data_builder/simple.hpp index 070da5512..2744e8ff7 100644 --- a/bindings/c/src/data_builder/simple.hpp +++ b/bindings/c/src/data_builder/simple.hpp @@ -92,11 +92,11 @@ struct lib::DispatchConverter void for_simple_specializations(F&& f) { +template void for_simple_specializations(F&& f) { using float_alloc = - svs::c_runtime::MaybeBlockedAlloc>; + svs::c_runtime::MaybeBlockedAlloc>; using float16_alloc = svs::c_runtime:: - MaybeBlockedAlloc>; + MaybeBlockedAlloc>; #define X(T, A, D) f.template operator(), D>(); #define XX(T, A) X(T, A, DistanceL2) X(T, A, DistanceIP) X(T, A, DistanceCosineSimilarity) XX(float, float_alloc) diff --git a/bindings/c/src/data_builder/sq.hpp b/bindings/c/src/data_builder/sq.hpp index 571b8cdb5..6d1957dd3 100644 --- a/bindings/c/src/data_builder/sq.hpp +++ b/bindings/c/src/data_builder/sq.hpp @@ -31,6 +31,8 @@ #include #include #include +// Must follow scalar.h, which does not include its own dependencies. +#include #include #include @@ -98,11 +100,11 @@ struct lib::DispatchConverter static To convert(From SVS_UNUSED(from)) { return To{}; } }; -template void for_sq_specializations(F&& f) { +template void for_sq_specializations(F&& f) { using int8_alloc = - svs::c_runtime::MaybeBlockedAlloc>; + svs::c_runtime::MaybeBlockedAlloc>; using uint8_alloc = - svs::c_runtime::MaybeBlockedAlloc>; + svs::c_runtime::MaybeBlockedAlloc>; #define X(T, A, D) f.template operator(), D>(); #define XX(T, A) X(T, A, DistanceL2) X(T, A, DistanceIP) X(T, A, DistanceCosineSimilarity) XX(uint8_t, uint8_alloc) diff --git a/bindings/c/src/dispatcher_concurrent_vamana.cpp b/bindings/c/src/dispatcher_concurrent_vamana.cpp new file mode 100644 index 000000000..dc2b4878a --- /dev/null +++ b/bindings/c/src/dispatcher_concurrent_vamana.cpp @@ -0,0 +1,345 @@ +/* + * Copyright 2026 Intel Corporation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "dispatcher_concurrent_vamana.hpp" + +#include "allocator.hpp" +#include "data_builder.hpp" +#include "dispatcher_dynamic_vamana.hpp" +#include "dispatcher_dynamic_vamana_copy.hpp" +#include "storage.hpp" +#include "types_support.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace svs::c_runtime { + +namespace { + +namespace cc = svs::index::vamana::concurrent; +using RegularSlot = svs::index::vamana::SlotMetadata; + +using ConcurrentGraph = cc::graphs::SimpleBlockedGraph>; +using ConcurrentGraphAllocator = cc::SegmentedBlocked>; + +RegularSlot to_regular_slot(cc::SlotMetadata s) { + switch (s) { + case cc::SlotMetadata::Empty: + return RegularSlot::Empty; + case cc::SlotMetadata::Valid: + return RegularSlot::Valid; + case cc::SlotMetadata::Deleted: + return RegularSlot::Deleted; + case cc::SlotMetadata::Pending: + break; + } + throw ANNEXCEPTION("Cannot convert an index with in-flight insertions"); +} + +cc::SlotMetadata to_concurrent_slot(RegularSlot s) { + switch (s) { + case RegularSlot::Empty: + return cc::SlotMetadata::Empty; + case RegularSlot::Valid: + return cc::SlotMetadata::Valid; + case RegularSlot::Deleted: + return cc::SlotMetadata::Deleted; + } + throw ANNEXCEPTION("Unknown slot state"); +} + +// Copy flavor for svs::index::vamana::concurrent::MutableVamanaIndex. +struct ConcurrentVamanaFlavor { + static constexpr BlockKind block_kind = BlockKind::Segmented; + + using graph_allocator_type = ConcurrentGraphAllocator; + + template + using impl_type = cc::MutableVamanaIndex; + + template static std::vector status(const Impl& impl) { + auto snapshot = impl.status_snapshot(); + std::vector result; + result.reserve(snapshot.size()); + for (auto s : snapshot) { + result.push_back(to_regular_slot(s)); + } + return result; + } + + template + static svs::DynamicVamana assemble( + const svs::index::vamana::VamanaIndexParameters& config, + Data data, + const SrcGraph& src_graph, + std::vector status, + std::span external_ids, + std::span internal_ids, + Distance distance, + svs::threads::ThreadPoolHandle pool, + graph_allocator_type graph_allocator, + svs::logging::logger_ptr logger + ) { + auto graph = ConcurrentGraph( + src_graph.n_nodes(), + config.build_parameters.graph_max_degree, + std::move(graph_allocator) + ); + svs::data::copy(src_graph.get_data(), graph.get_data()); + + auto translator = cc::IDTranslator{}; + translator.insert(external_ids, internal_ids); + auto concurrent_status = std::vector(); + concurrent_status.reserve(status.size()); + for (auto s : status) { + concurrent_status.push_back(to_concurrent_slot(s)); + } + + return svs::ConcurrentDynamicVamana( + svs::DynamicVamana::AssembleTag{}, + svs::lib::Types{}, + impl_type( + config, + std::move(data), + std::move(graph), + std::move(distance), + concurrent_status, + std::move(translator), + std::move(pool), + std::move(logger) + ) + ); + } +}; + +template +svs::DynamicVamana build_concurrent_vamana_index( + const svs::index::vamana::VamanaBuildParameters& build_params, + std::pair, std::span> src_data, + DataBuilder builder, + Distance D, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + using allocator_type = typename DataBuilder::allocator_type; + using value_type = typename allocator_type::value_type; + + auto data_allocator = + allocator_type{block_params, allocator_builder.build()}; + auto data = builder.build(std::move(src_data.first), pool, data_allocator); + + auto graph_allocator = ConcurrentGraphAllocator{ + block_params, allocator_builder.build_for_graph()}; + return svs::ConcurrentDynamicVamana::build( + build_params, + std::move(data), + src_data.second, + std::move(D), + std::move(pool), + graph_allocator, + std::move(logger) + ); +} + +template +svs::DynamicVamana load_concurrent_vamana_index( + const svs::index::vamana::VamanaBuildParameters& SVS_UNUSED(build_params), + const std::filesystem::path& directory, + DataLoader loader, + Distance D, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + using allocator_type = typename DataLoader::allocator_type; + using value_type = typename allocator_type::value_type; + auto allocator = allocator_type{block_params, allocator_builder.build()}; + auto data = loader.load(directory / "data", allocator); + + auto graph_allocator = ConcurrentGraphAllocator{ + block_params, allocator_builder.build_for_graph()}; + return svs::ConcurrentDynamicVamana::assemble( + directory / "config", + SVS_LAZY(ConcurrentGraph::load(directory / "graph", graph_allocator)), + std::move(data), + std::move(D), + std::move(pool), + /*debug_load_from_static=*/false, + std::move(logger) + ); +} + +using ConcurrentVamanaSource = std::variant< + std::pair, std::span>, + std::filesystem::path>; + +using BuildConcurrentIndexDispatcher = svs::lib::Dispatcher< + svs::DynamicVamana, + const svs::index::vamana::VamanaBuildParameters&, + ConcurrentVamanaSource, + const Storage*, + svs::DistanceType, + svs::threads::ThreadPoolHandle, + const AllocatorBuilder&, + const svs::data::BlockingParameters&, + svs::logging::logger_ptr>; + +const BuildConcurrentIndexDispatcher& build_concurrent_vamana_index_dispatcher() { + static const BuildConcurrentIndexDispatcher dispatcher = [] { + BuildConcurrentIndexDispatcher d{}; + auto build_closure = [&d]() { + d.register_target(&build_concurrent_vamana_index); + }; + auto load_closure = [&d]() { + d.register_target(&load_concurrent_vamana_index); + }; + + constexpr auto kind = ConcurrentVamanaFlavor::block_kind; + + for_all_specializations(build_closure); + for_all_specializations(load_closure); + return d; + }(); + return dispatcher; +} + +const detail::CopyDynamicIndexDispatcher& +copy_concurrent_index_dispatcher(bool src_concurrent, bool dst_concurrent) { + using detail::copy_dynamic_index_dispatcher; + using detail::RegularVamanaFlavor; + if (src_concurrent) { + return dst_concurrent ? copy_dynamic_index_dispatcher< + ConcurrentVamanaFlavor, + ConcurrentVamanaFlavor>() + : copy_dynamic_index_dispatcher< + ConcurrentVamanaFlavor, + RegularVamanaFlavor>(); + } + INVALID_ARGUMENT_IF(!dst_concurrent, "Neither index is concurrent"); + return copy_dynamic_index_dispatcher(); +} + +} // namespace + +svs::DynamicVamana dispatch_concurrent_vamana_index_build( + const svs::index::vamana::VamanaBuildParameters& build_params, + svs::data::ConstSimpleDataView data, + std::span ids, + const Storage* storage, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + return build_concurrent_vamana_index_dispatcher().invoke( + build_params, + ConcurrentVamanaSource{std::make_pair(data, ids)}, + storage, + distance_type, + std::move(pool), + allocator_builder, + block_params, + std::move(logger) + ); +} + +svs::DynamicVamana dispatch_concurrent_vamana_index_load( + const svs::index::vamana::VamanaBuildParameters& build_params, + const std::filesystem::path& directory, + const Storage* storage, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + return build_concurrent_vamana_index_dispatcher().invoke( + build_params, + ConcurrentVamanaSource{directory}, + storage, + distance_type, + std::move(pool), + allocator_builder, + block_params, + std::move(logger) + ); +} + +svs::DynamicVamana dispatch_concurrent_vamana_index_copy( + const svs::index::vamana::VamanaBuildParameters& build_params, + const svs::DynamicVamana& src_index, + bool src_concurrent, + const Storage* src_storage, + const Storage* dst_storage, + bool dst_concurrent, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + return copy_concurrent_index_dispatcher(src_concurrent, dst_concurrent) + .invoke( + build_params, + src_index, + src_storage, + dst_storage, + distance_type, + std::move(pool), + allocator_builder, + block_params, + std::move(logger) + ); +} + +svs::index::vamana::MemoryBreakdown dispatch_concurrent_vamana_memory_estimate( + const svs::index::vamana::VamanaBuildParameters& build_params, + size_t num_vectors, + size_t dimension, + const Storage* storage, + svs::DistanceType distance_type, + const svs::data::BlockingParameters& block_params +) { + auto breakdown = dispatch_dynamic_vamana_memory_estimate( + build_params, num_vectors, dimension, storage, distance_type, block_params + ); + // One in-neighbor list per node; each edge is recorded once in reverse. + // Add 25% slack for heap capacity rounding and allocator overhead. This is + // still an estimate, not a hard upper bound (allocator behavior may vary). + using index_type = uint32_t; + const auto reverse_edges_bytes = + num_vectors * (sizeof(std::vector) + sizeof(cc::SpinLock) + + build_params.graph_max_degree * sizeof(index_type)); + breakdown.reverse_edges_bytes = reverse_edges_bytes + reverse_edges_bytes / 4; + return breakdown; +} + +} // namespace svs::c_runtime diff --git a/bindings/c/src/dispatcher_concurrent_vamana.hpp b/bindings/c/src/dispatcher_concurrent_vamana.hpp new file mode 100644 index 000000000..0710e88bb --- /dev/null +++ b/bindings/c/src/dispatcher_concurrent_vamana.hpp @@ -0,0 +1,81 @@ +/* + * Copyright 2026 Intel Corporation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include "allocator.hpp" +#include "storage.hpp" + +#include +#include +#include +#include +#include + +#include +#include + +namespace svs::c_runtime { + +// The returned index wraps svs::ConcurrentDynamicVamana. +svs::DynamicVamana dispatch_concurrent_vamana_index_build( + const svs::index::vamana::VamanaBuildParameters& build_params, + svs::data::ConstSimpleDataView data, + std::span ids, + const Storage* storage, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +); + +svs::DynamicVamana dispatch_concurrent_vamana_index_load( + const svs::index::vamana::VamanaBuildParameters& build_params, + const std::filesystem::path& directory, + const Storage* storage, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +); + +// Copies between a concurrent and a regular or concurrent dynamic index; +// at least one of `src_concurrent` and `dst_concurrent` must be true. +svs::DynamicVamana dispatch_concurrent_vamana_index_copy( + const svs::index::vamana::VamanaBuildParameters& build_params, + const svs::DynamicVamana& src_index, + bool src_concurrent, + const Storage* src_storage, + const Storage* dst_storage, + bool dst_concurrent, + svs::DistanceType distance_type, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +); + +svs::index::vamana::MemoryBreakdown dispatch_concurrent_vamana_memory_estimate( + const svs::index::vamana::VamanaBuildParameters& build_params, + size_t num_vectors, + size_t dimension, + const Storage* storage, + svs::DistanceType distance_type, + const svs::data::BlockingParameters& block_params +); + +} // namespace svs::c_runtime diff --git a/bindings/c/src/dispatcher_dynamic_vamana.cpp b/bindings/c/src/dispatcher_dynamic_vamana.cpp index ff1653b21..7862b17b4 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.cpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.cpp @@ -15,20 +15,16 @@ */ #include "dispatcher_dynamic_vamana.hpp" -#include "algorithm.hpp" #include "allocator.hpp" #include "data_builder.hpp" +#include "dispatcher_dynamic_vamana_copy.hpp" #include "storage.hpp" -#include "threadpool.hpp" #include "types_support.hpp" -#include #include #include -#include #include #include -#include #include #include @@ -41,6 +37,7 @@ namespace svs::c_runtime { namespace { + template svs::DynamicVamana build_dynamic_vamana_index( const svs::index::vamana::VamanaBuildParameters& build_params, @@ -147,18 +144,10 @@ void register_dynamic_vamana_index_specializations(Dispatcher& dispatcher) { dispatcher.register_target(&load_stream_dynamic_vamana_index); }; - for_simple_specializations(build_closure); - for_simple_specializations(load_closure); - for_simple_specializations(load_stream_closure); - for_leanvec_specializations(build_closure); - for_leanvec_specializations(load_closure); - for_leanvec_specializations(load_stream_closure); - for_lvq_specializations(build_closure); - for_lvq_specializations(load_closure); - for_lvq_specializations(load_stream_closure); - for_sq_specializations(build_closure); - for_sq_specializations(load_closure); - for_sq_specializations(load_stream_closure); + constexpr auto kind = BlockKind::Blocked; + for_all_specializations(build_closure); + for_all_specializations(load_closure); + for_all_specializations(load_stream_closure); } // Stream load alternative, matched via generic variant DispatchConverter like the @@ -180,7 +169,7 @@ using BuildDynamicIndexDispatcher = svs::lib::Dispatcher< svs::logging::logger_ptr>; const BuildDynamicIndexDispatcher& build_dynamic_vamana_index_dispatcher() { - static BuildDynamicIndexDispatcher dispatcher = [] { + static const BuildDynamicIndexDispatcher dispatcher = [] { BuildDynamicIndexDispatcher d{}; register_dynamic_vamana_index_specializations(d); return d; @@ -188,149 +177,6 @@ const BuildDynamicIndexDispatcher& build_dynamic_vamana_index_dispatcher() { return dispatcher; } -using CopyDynamicIndexDispatcher = svs::lib::Dispatcher< - svs::DynamicVamana, - const svs::index::vamana::VamanaBuildParameters&, - const svs::DynamicVamana&, - const Storage*, // src - const Storage*, // dst - svs::DistanceType, - svs::threads::ThreadPoolHandle, - const AllocatorBuilder&, - const svs::data::BlockingParameters&, - svs::logging::logger_ptr>; - -template -svs::DynamicVamana copy_dynamic_vamana_index( - const svs::index::vamana::VamanaBuildParameters& build_params, - const svs::DynamicVamana& src_index, - SrcDataBuilder src_builder, - DstDataBuilder dst_builder, - Distance distance, - svs::threads::ThreadPoolHandle pool, - const AllocatorBuilder& allocator_builder, - const svs::data::BlockingParameters& block_params, - svs::logging::logger_ptr logger -) { - auto config = src_index.parameters(); - - // A defaulted alpha is resolved per-metric at build time, so it cannot be compared. - constexpr svs::index::vamana::VamanaBuildParameters default_build_params{}; - const bool alpha_is_default = build_params.alpha == default_build_params.alpha; - - // Validate build parameters match - if (config.build_parameters.graph_max_degree != build_params.graph_max_degree || - (!alpha_is_default && config.build_parameters.alpha != build_params.alpha)) { - throw not_implemented("Index build parameters mismatch"); - } - - // Other build parameters that are not explicitly checked above are updated here. - config.build_parameters.apply(build_params); - verify_and_set_default_index_parameters(config.build_parameters, distance); - - // Must match the graph type produced by the build and load paths above. - using GraphType = - svs::graphs::SimpleGraph>>; - - // Get the typed index implementation from the source index - using SrcDataType = typename SrcDataBuilder::data_type; - using IndexImplType = - svs::index::vamana::MutableVamanaIndex; - auto src_index_impl = - src_index.template get_typed_impl, IndexImplType>(); - if (!src_index_impl) { - throw std::runtime_error("Failed to get typed index implementation"); - } - - // Copy the graph structure from the source index to the new graph instance - const auto& src_graph = src_index_impl->view_graph(); - assert(src_graph.max_degree() == config.build_parameters.graph_max_degree); - - auto graph_allocator_handle = allocator_builder.build_for_graph(); - auto graph_allocator = svs::data::Blocked{block_params, graph_allocator_handle}; - auto graph = - GraphType(src_graph.n_nodes(), build_params.graph_max_degree, graph_allocator); - svs::data::copy(src_graph.get_data(), graph.get_data()); - - // Copy/convert the data from the source index to the new data instance - decltype(auto) src_data = src_builder.get_dataset(src_index_impl->view_data()); - - using allocator_type = typename DstDataBuilder::allocator_type; - using value_type = typename allocator_type::value_type; - - auto data_allocator_handle = allocator_builder.build(); - auto data_allocator = allocator_type{block_params, data_allocator_handle}; - auto data = dst_builder.build(src_data, pool, data_allocator); - - svs::index::vamana::detail::VamanaStateLoader state_loader{ - config, src_index_impl->view_translator(), src_index_impl->view_status()}; - - return svs::DynamicVamana::assemble( - std::move(state_loader), - std::move(graph), - std::move(data), - distance, - std::move(pool), - /*debug_load_from_static=*/false, - std::move(logger) - ); -} - -template void register_copy_specializations(Dispatcher& dispatcher) { - // TODO: Enable Compressed -> Compressed specializations by making decompressors, - // decompression accessors and decompression dataset are thread-safe - - // Compression specializations for copy_vamana_index - // To handle cases Simple -> Compressed - auto compression_closure = [&dispatcher]() { - // Skip all distance specializations except one - if constexpr (!std::is_same_v) { - return; - } - auto inner_closure = [&dispatcher]() { - dispatcher.register_target(©_dynamic_vamana_index< - SrcDataBuilder, - DstDataBuilder, - Distance>); - }; - - for_simple_specializations(inner_closure); - for_sq_specializations(inner_closure); - for_lvq_specializations(inner_closure); - for_leanvec_specializations(inner_closure); - }; - for_simple_specializations(compression_closure); - - // Decompression specializations for copy_vamana_index - // To handle cases Compressed -> Simple - auto decompression_closure = [&dispatcher]() { - // Skip all distance specializations except one - if constexpr (!std::is_same_v) { - return; - } - auto inner_closure = [&dispatcher]() { - dispatcher.register_target(©_dynamic_vamana_index< - SrcDataBuilder, - DstDataBuilder, - Distance>); - }; - - for_simple_specializations(inner_closure); - }; - for_sq_specializations(decompression_closure); - for_lvq_specializations(decompression_closure); - for_leanvec_specializations(decompression_closure); -} - -const CopyDynamicIndexDispatcher& copy_dynamic_index_dispatcher() { - static CopyDynamicIndexDispatcher dispatcher = [] { - CopyDynamicIndexDispatcher d{}; - register_copy_specializations(d); - return d; - }(); - return dispatcher; -} - } // namespace svs::DynamicVamana dispatch_dynamic_vamana_index_build( @@ -410,17 +256,19 @@ svs::DynamicVamana dispatch_dynamic_vamana_index_copy( const svs::data::BlockingParameters& block_params, svs::logging::logger_ptr logger ) { - return copy_dynamic_index_dispatcher().invoke( - build_params, - src_index, - src_storage, - dst_storage, - distance_type, - std::move(pool), - allocator_builder, - block_params, - std::move(logger) - ); + using detail::RegularVamanaFlavor; + return detail::copy_dynamic_index_dispatcher() + .invoke( + build_params, + src_index, + src_storage, + dst_storage, + distance_type, + std::move(pool), + allocator_builder, + block_params, + std::move(logger) + ); } svs::index::vamana::MemoryBreakdown dispatch_dynamic_vamana_memory_estimate( diff --git a/bindings/c/src/dispatcher_dynamic_vamana.hpp b/bindings/c/src/dispatcher_dynamic_vamana.hpp index ca3a382ff..47468fbf5 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.hpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.hpp @@ -15,6 +15,7 @@ */ #pragma once +#include "allocator.hpp" #include "storage.hpp" #include diff --git a/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp new file mode 100644 index 000000000..99feb52c2 --- /dev/null +++ b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp @@ -0,0 +1,233 @@ +/* + * Copyright 2026 Intel Corporation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +// Conversion between dynamic Vamana indices, shared by the regular and concurrent +// dispatchers. An index flavor provides: +// - `impl_type`: the typed index implementation; +// - `block_kind`: the data storage layout; +// - `status(impl)`: slot states in `svs::index::vamana::SlotMetadata` terms; +// - `assemble(...)`: a new index from copied state. + +#include "allocator.hpp" +#include "data_builder.hpp" +#include "storage.hpp" +#include "types_support.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace svs::c_runtime::detail { + +using CopyDynamicIndexDispatcher = svs::lib::Dispatcher< + svs::DynamicVamana, + const svs::index::vamana::VamanaBuildParameters&, + const svs::DynamicVamana&, + const Storage*, // src + const Storage*, // dst + svs::DistanceType, + svs::threads::ThreadPoolHandle, + const AllocatorBuilder&, + const svs::data::BlockingParameters&, + svs::logging::logger_ptr>; + +struct RegularVamanaFlavor { + using graph_allocator_type = svs::data::Blocked>; + using graph_type = svs::graphs::SimpleGraph; + static constexpr BlockKind block_kind = BlockKind::Blocked; + + template + using impl_type = svs::index::vamana::MutableVamanaIndex; + + template + static std::vector status(const Impl& impl) { + return impl.view_status(); + } + + template + static svs::DynamicVamana assemble( + const svs::index::vamana::VamanaIndexParameters& config, + Data data, + const SrcGraph& src_graph, + std::vector status, + std::span external_ids, + std::span internal_ids, + Distance distance, + svs::threads::ThreadPoolHandle pool, + const graph_allocator_type& graph_allocator, + svs::logging::logger_ptr logger + ) { + // Copy the graph structure from the source index to the new graph instance + assert(src_graph.max_degree() == config.build_parameters.graph_max_degree); + auto graph = graph_type( + src_graph.n_nodes(), config.build_parameters.graph_max_degree, graph_allocator + ); + svs::data::copy(src_graph.get_data(), graph.get_data()); + + auto translator = svs::IDTranslator{}; + translator.insert(external_ids, internal_ids); + svs::index::vamana::detail::VamanaStateLoader state_loader{ + config, std::move(translator), std::move(status)}; + + return svs::DynamicVamana::assemble( + std::move(state_loader), + std::move(graph), + std::move(data), + std::move(distance), + std::move(pool), + /*debug_load_from_static=*/false, + std::move(logger) + ); + } +}; + +template < + typename SrcFlavor, + typename DstFlavor, + typename SrcDataBuilder, + typename DstDataBuilder, + typename Distance> +svs::DynamicVamana copy_dynamic_vamana_index( + const svs::index::vamana::VamanaBuildParameters& build_params, + const svs::DynamicVamana& src_index, + SrcDataBuilder src_builder, + DstDataBuilder dst_builder, + Distance distance, + svs::threads::ThreadPoolHandle pool, + const AllocatorBuilder& allocator_builder, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger +) { + auto config = src_index.parameters(); + + // A defaulted alpha is resolved per-metric at build time, so it cannot be compared. + constexpr svs::index::vamana::VamanaBuildParameters default_build_params{}; + const bool alpha_is_default = build_params.alpha == default_build_params.alpha; + + // Validate build parameters match + if (config.build_parameters.graph_max_degree != build_params.graph_max_degree || + (!alpha_is_default && config.build_parameters.alpha != build_params.alpha)) { + throw not_implemented("Index build parameters mismatch"); + } + + // Other build parameters that are not explicitly checked above are updated here. + config.build_parameters.apply(build_params); + verify_and_set_default_index_parameters(config.build_parameters, distance); + + using SrcImpl = typename SrcFlavor:: + template impl_type; + auto src_index_impl = + src_index.template get_typed_impl, SrcImpl>(); + if (!src_index_impl) { + throw std::runtime_error("Failed to get typed index implementation"); + } + + // Only valid slots are translated: the concurrent translator keeps soft-deleted + // entries until consolidation. + auto status = SrcFlavor::status(*src_index_impl); + std::vector external_ids; + std::vector internal_ids; + for (auto pair : src_index_impl->view_translator()) { + if (status.at(pair.second) == svs::index::vamana::SlotMetadata::Valid) { + external_ids.push_back(pair.first); + internal_ids.push_back(lib::narrow(pair.second)); + } + } + + decltype(auto) src_data = src_builder.get_dataset(src_index_impl->view_data()); + + using allocator_type = typename DstDataBuilder::allocator_type; + using value_type = typename allocator_type::value_type; + auto data_allocator = + allocator_type{block_params, allocator_builder.build()}; + auto data = dst_builder.build(src_data, pool, data_allocator); + + auto graph_allocator = typename DstFlavor::graph_allocator_type{ + block_params, allocator_builder.build_for_graph()}; + + return DstFlavor::assemble( + config, + std::move(data), + src_index_impl->view_graph(), + std::move(status), + external_ids, + internal_ids, + std::move(distance), + std::move(pool), + graph_allocator, + std::move(logger) + ); +} + +template +void register_copy_specializations(CopyDynamicIndexDispatcher& dispatcher) { + // TODO: Enable Compressed -> Compressed specializations by making decompressors, + // decompression accessors and decompression dataset are thread-safe + auto register_pair = [&dispatcher]() { + return [&dispatcher]() { + dispatcher.register_target(©_dynamic_vamana_index< + SrcFlavor, + DstFlavor, + SrcDataBuilder, + DstDataBuilder, + Distance>); + }; + }; + + // Simple -> any; one distance is enough to enumerate the source types. + auto compression_closure = [&]() { + if constexpr (std::is_same_v) { + auto inner = register_pair.template operator()(); + for_all_specializations(inner); + } + }; + for_simple_specializations(compression_closure); + + // Compressed -> Simple + auto decompression_closure = [&]() { + if constexpr (std::is_same_v) { + auto inner = register_pair.template operator()(); + for_simple_specializations(inner); + } + }; + for_sq_specializations(decompression_closure); + for_lvq_specializations(decompression_closure); + for_leanvec_specializations(decompression_closure); +} + +template +const CopyDynamicIndexDispatcher& copy_dynamic_index_dispatcher() { + static const CopyDynamicIndexDispatcher dispatcher = [] { + CopyDynamicIndexDispatcher d{}; + register_copy_specializations(d); + return d; + }(); + return dispatcher; +} + +} // namespace svs::c_runtime::detail diff --git a/bindings/c/src/dispatcher_vamana.cpp b/bindings/c/src/dispatcher_vamana.cpp index 9cb724259..b3352ce05 100644 --- a/bindings/c/src/dispatcher_vamana.cpp +++ b/bindings/c/src/dispatcher_vamana.cpp @@ -124,18 +124,9 @@ void register_vamana_index_specializations(Dispatcher& dispatcher) { dispatcher.register_target(&load_stream_vamana_index); }; - for_simple_specializations(build_closure); - for_simple_specializations(load_closure); - for_simple_specializations(load_stream_closure); - for_leanvec_specializations(build_closure); - for_leanvec_specializations(load_closure); - for_leanvec_specializations(load_stream_closure); - for_lvq_specializations(build_closure); - for_lvq_specializations(load_closure); - for_lvq_specializations(load_stream_closure); - for_sq_specializations(build_closure); - for_sq_specializations(load_closure); - for_sq_specializations(load_stream_closure); + for_all_specializations(build_closure); + for_all_specializations(load_closure); + for_all_specializations(load_stream_closure); } // Stream load alternative, matched via generic variant DispatchConverter like the @@ -239,6 +230,8 @@ svs::Vamana copy_vamana_index( template void register_copy_vamana_index_specializations(Dispatcher& dispatcher) { + constexpr auto kind = BlockKind::None; + // TODO: Enable Compressed -> Compressed specializations by making decompressors, // decompression accessors and decompression dataset are thread-safe @@ -256,12 +249,9 @@ void register_copy_vamana_index_specializations(Dispatcher& dispatcher) { Distance>); }; - for_simple_specializations(inner_closure); - for_sq_specializations(inner_closure); - for_lvq_specializations(inner_closure); - for_leanvec_specializations(inner_closure); + for_all_specializations(inner_closure); }; - for_simple_specializations(compression_closure); + for_simple_specializations(compression_closure); // Decompression specializations for copy_vamana_index // To handle cases Compressed -> Simple @@ -277,11 +267,11 @@ void register_copy_vamana_index_specializations(Dispatcher& dispatcher) { Distance>); }; - for_simple_specializations(inner_closure); + for_simple_specializations(inner_closure); }; - for_sq_specializations(decompression_closure); - for_lvq_specializations(decompression_closure); - for_leanvec_specializations(decompression_closure); + for_sq_specializations(decompression_closure); + for_lvq_specializations(decompression_closure); + for_leanvec_specializations(decompression_closure); } const CopyIndexDispatcher& copy_vamana_index_dispatcher() { diff --git a/bindings/c/src/index_builder.cpp b/bindings/c/src/index_builder.cpp index 14db878d6..5609b87b3 100644 --- a/bindings/c/src/index_builder.cpp +++ b/bindings/c/src/index_builder.cpp @@ -17,6 +17,7 @@ #include "algorithm.hpp" #include "data_builder.hpp" +#include "dispatcher_concurrent_vamana.hpp" #include "dispatcher_dynamic_vamana.hpp" #include "dispatcher_vamana.hpp" #include "error.hpp" @@ -132,6 +133,22 @@ void validate_builder_compatibility( ); } } + +// FINE_GRAIN is served by the concurrent index; other kinds by the regular one. +bool uses_concurrent_index(svs_sync_kind_t sync_kind) { + return sync_kind == SVS_SYNC_KIND_FINE_GRAIN; +} + +std::shared_ptr make_dynamic_index( + const IndexBuilder& builder, svs::DynamicVamana&& index, svs_sync_kind_t sync_kind +) { + if (uses_concurrent_index(sync_kind)) { + return std::make_shared( + builder, std::move(index), sync_kind + ); + } + return std::make_shared(builder, std::move(index), sync_kind); +} } // namespace std::shared_ptr IndexBuilder::copy(const std::shared_ptr& src_index) { @@ -207,24 +224,46 @@ std::shared_ptr IndexBuilder::copy_dynamic( throw std::invalid_argument("Source index must be a valid Dynamic Vamana index."); } - auto src_lock = vamana_index->read_lock(); - auto index = std::make_shared( - *this, - dispatch_dynamic_vamana_index_copy( - dst_build_parameters, - vamana_index->index, - src_builder.storage.get(), - storage.get(), - to_distance_type(distance_metric), - pool_builder.build(), - allocator_builder, - block_params, - get_logger() - ), - sync_kind - ); + const bool src_concurrent = + std::dynamic_pointer_cast(vamana_index) != nullptr; + const bool dst_concurrent = uses_concurrent_index(sync_kind); + + // The concurrent index may be updated under the shared lock, so reading its whole + // state needs the exclusive one. + std::shared_lock src_read_lock; + std::unique_lock src_write_lock; + if (src_concurrent) { + src_write_lock = vamana_index->write_lock(); + } else { + src_read_lock = vamana_index->read_lock(); + } - return index; + auto copied = (src_concurrent || dst_concurrent) + ? dispatch_concurrent_vamana_index_copy( + dst_build_parameters, + vamana_index->index, + src_concurrent, + src_builder.storage.get(), + storage.get(), + dst_concurrent, + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ) + : dispatch_dynamic_vamana_index_copy( + dst_build_parameters, + vamana_index->index, + src_builder.storage.get(), + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ); + return make_dynamic_index(*this, std::move(copied), sync_kind); } std::shared_ptr IndexBuilder::build_dynamic( @@ -235,25 +274,31 @@ std::shared_ptr IndexBuilder::build_dynamic( ) { if (algorithm->type == SVS_ALGORITHM_TYPE_VAMANA) { auto vamana_algorithm = static_cast(algorithm.get()); - - auto index = std::make_shared( - *this, - // vamana_algorithm, - dispatch_dynamic_vamana_index_build( - vamana_algorithm->build_parameters(), - data, - ids, - storage.get(), - to_distance_type(distance_metric), - pool_builder.build(), - allocator_builder, - block_params, - get_logger() - ), - sync_kind - ); - - return index; + const auto& build_params = vamana_algorithm->build_parameters(); + auto index = uses_concurrent_index(sync_kind) + ? dispatch_concurrent_vamana_index_build( + build_params, + data, + ids, + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ) + : dispatch_dynamic_vamana_index_build( + build_params, + data, + ids, + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ); + return make_dynamic_index(*this, std::move(index), sync_kind); } return nullptr; } @@ -265,24 +310,29 @@ std::shared_ptr IndexBuilder::load_dynamic( ) { if (algorithm->type == SVS_ALGORITHM_TYPE_VAMANA) { auto vamana_algorithm = static_cast(algorithm.get()); - - auto index = std::make_shared( - *this, - // vamana_algorithm, - dispatch_dynamic_vamana_index_load( - vamana_algorithm->build_parameters(), - directory, - storage.get(), - to_distance_type(distance_metric), - pool_builder.build(), - allocator_builder, - block_params, - get_logger() - ), - sync_kind - ); - - return index; + const auto& build_params = vamana_algorithm->build_parameters(); + auto index = uses_concurrent_index(sync_kind) + ? dispatch_concurrent_vamana_index_load( + build_params, + directory, + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ) + : dispatch_dynamic_vamana_index_load( + build_params, + directory, + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ); + return make_dynamic_index(*this, std::move(index), sync_kind); } return nullptr; } @@ -303,7 +353,7 @@ std::shared_ptr IndexBuilder::load_stream_dynamic( std::move(stream), storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params ), @@ -332,14 +382,19 @@ IndexBuilder::estimate_memory_breakdown(size_t num_vectors) const { } svs::index::vamana::MemoryBreakdown IndexBuilder::estimate_memory_breakdown_dynamic( - size_t num_vectors, const svs::data::BlockingParameters& block_params + size_t num_vectors, + const svs::data::BlockingParameters& block_params, + svs_sync_kind_t sync_kind ) const { NOT_IMPLEMENTED_IF( algorithm->type != SVS_ALGORITHM_TYPE_VAMANA, "Memory estimation is currently supported only for Vamana algorithm" ); auto vamana_algorithm = static_cast(algorithm.get()); - return dispatch_dynamic_vamana_memory_estimate( + auto estimate = uses_concurrent_index(sync_kind) + ? &dispatch_concurrent_vamana_memory_estimate + : &dispatch_dynamic_vamana_memory_estimate; + return estimate( vamana_algorithm->build_parameters(), num_vectors, dimension, diff --git a/bindings/c/src/index_builder.hpp b/bindings/c/src/index_builder.hpp index 82eca1066..8d15518cd 100644 --- a/bindings/c/src/index_builder.hpp +++ b/bindings/c/src/index_builder.hpp @@ -150,7 +150,9 @@ struct IndexBuilder { // for `num_vectors` vectors. Mirrors the accounting done by // svs::index::vamana::MutableVamanaIndex::get_memory_breakdown(). svs::index::vamana::MemoryBreakdown estimate_memory_breakdown_dynamic( - size_t num_vectors, const svs::data::BlockingParameters& block_params + size_t num_vectors, + const svs::data::BlockingParameters& block_params, + svs_sync_kind_t sync_kind = SVS_SYNC_KIND_NONE ) const; size_t estimate_search_memory( diff --git a/bindings/c/src/index_vamana.cpp b/bindings/c/src/index_vamana.cpp index bcaa00c3d..3aa966e75 100644 --- a/bindings/c/src/index_vamana.cpp +++ b/bindings/c/src/index_vamana.cpp @@ -100,10 +100,7 @@ DynamicIndexVamana::DynamicIndexVamana( ) : DynamicIndex(std::make_unique(builder), sync_kind) , index(std::move(index)) { - auto all_ids = this->index.all_ids(); - auto [min_it, max_it] = std::minmax_element(all_ids.begin(), all_ids.end()); - min_id = (min_it == all_ids.end()) ? 0 : *min_it; - max_id = (max_it == all_ids.end()) ? 0 : *max_it; + track_id_range(lib::as_span(this->index.all_ids())); // Apply default search parameters to the index auto algorithm_parameters = std::static_pointer_cast( @@ -139,7 +136,9 @@ std::pair, std::vector> DynamicIndexVamana::sea } std::mt19937 rng(42); - std::uniform_int_distribution dist(min_id, max_id); + const size_t lo = min_id.load(std::memory_order_relaxed); + const size_t hi = std::max(lo, max_id.load(std::memory_order_relaxed)); + std::uniform_int_distribution dist(lo, hi); // DynamicVamana index IDs provided by user and may have any values and gaps, so we // need to sample until we find a valid ID. // The most reliable way would be get all IDs and sample from them, but that may be @@ -150,7 +149,7 @@ std::pair, std::vector> DynamicIndexVamana::sea // attempts. This ensures that we have a reasonable chance of finding a valid ID // without excessive sampling. // Note: (index.size() + 1) - to avoid division by zero in case the index is empty. - const size_t max_attempts = std::max((max_id - min_id) / (index.size() + 1), size_t{4}); + const size_t max_attempts = std::max((hi - lo) / (index.size() + 1), size_t{4}); auto sample_generator = [&]() -> size_t { for (size_t attempt = 0; attempt < max_attempts; ++attempt) { @@ -175,14 +174,7 @@ size_t DynamicIndexVamana::add_points( svs::data::ConstSimpleDataView new_points, std::span ids ) { auto lock = write_lock(); - // Track the maximum ID added to the index for ids generator - auto [min_it, max_it] = std::minmax_element(ids.begin(), ids.end()); - if (min_it != ids.end()) { - min_id = std::min(min_id, *min_it); - } - if (max_it != ids.end()) { - max_id = std::max(max_id, *max_it); - } + track_id_range(ids); auto old_size = index.size(); index.add_points(new_points, ids); // TODO: This is a bit of a hack - we should ideally return the number of points @@ -192,6 +184,22 @@ size_t DynamicIndexVamana::add_points( size_t DynamicIndexVamana::delete_points(std::span ids) { auto lock = write_lock(); + return delete_points_unlocked(ids); +} + +void DynamicIndexVamana::track_id_range(std::span ids) { + auto [min_it, max_it] = std::minmax_element(ids.begin(), ids.end()); + if (min_it != ids.end()) { + size_t current = min_id.load(); + while (*min_it < current && !min_id.compare_exchange_weak(current, *min_it)) {} + } + if (max_it != ids.end()) { + size_t current = max_id.load(); + while (*max_it > current && !max_id.compare_exchange_weak(current, *max_it)) {} + } +} + +size_t DynamicIndexVamana::delete_points_unlocked(std::span ids) { std::vector ids_to_delete; ids_to_delete.reserve(ids.size()); @@ -224,6 +232,10 @@ void DynamicIndexVamana::compact(size_t batchsize) { if (!index.is_consolidated()) { index.consolidate(); } + compact_unlocked(batchsize); +} + +void DynamicIndexVamana::compact_unlocked(size_t batchsize) { if (batchsize == 0) { index.compact(); // Use default batch size } else { @@ -234,7 +246,57 @@ void DynamicIndexVamana::compact(size_t batchsize) { void DynamicIndexVamana::set_num_threads(size_t num_threads) { auto lock = write_lock(); this->builder->pool_builder.resize(num_threads); - index.set_threadpool(this->builder->pool_builder.build()); + index.set_threadpool(this->builder->pool_builder.build(sync_kind)); +} + +///////////////////////////////////////// +// ConcurrentIndexVamana Implementation +ConcurrentIndexVamana::ConcurrentIndexVamana( + const IndexBuilder& builder, svs::DynamicVamana&& index, svs_sync_kind_t sync_kind +) + : DynamicIndexVamana(builder, std::move(index), sync_kind) {} + +std::pair, std::vector> ConcurrentIndexVamana::search( + svs::data::ConstSimpleDataView queries, + size_t num_neighbors, + const std::shared_ptr& search_params, + const IDFilterInterface* id_filter +) { + assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN); + if (queries.size() > 1) { + throw not_implemented( + "FINE GRAIN dynamic index only supports searching one query at a time." + ); + } + return DynamicIndexVamana::search(queries, num_neighbors, search_params, id_filter); +} + +size_t ConcurrentIndexVamana::add_points( + svs::data::ConstSimpleDataView new_points, std::span ids +) { + assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN); + if (new_points.size() > 1) { + throw not_implemented( + "FINE GRAIN dynamic index only supports adding one point at a time." + ); + } + auto lock = read_lock(); + track_id_range(ids); + // The size may change concurrently, so it cannot measure this call; a failed add + // throws instead. + index.add_points(new_points, ids); + return ids.size(); +} + +size_t ConcurrentIndexVamana::delete_points(std::span ids) { + assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN); + if (ids.size() > 1) { + throw not_implemented( + "FINE GRAIN dynamic index only supports deleting one point at a time." + ); + } + auto lock = read_lock(); + return delete_points_unlocked(ids); } } // namespace svs::c_runtime diff --git a/bindings/c/src/index_vamana.hpp b/bindings/c/src/index_vamana.hpp index 80e5605e5..2a193981e 100644 --- a/bindings/c/src/index_vamana.hpp +++ b/bindings/c/src/index_vamana.hpp @@ -24,7 +24,9 @@ #include #include +#include #include +#include #include #include #include @@ -75,8 +77,10 @@ struct IndexVamana : public Index { struct DynamicIndexVamana : public DynamicIndex { svs::DynamicVamana index; - size_t min_id = 0; // Track the minimum ID added to the index - size_t max_id = 0; // Track the maximum ID added to the index + // Bounds of the IDs ever added, used to sample IDs for filtered search. Atomic since + // ConcurrentIndexVamana adds points under a shared lock. + std::atomic min_id = std::numeric_limits::max(); + std::atomic max_id = std::numeric_limits::min(); DynamicIndexVamana( const IndexBuilder& builder, svs::DynamicVamana&& index, @@ -144,5 +148,52 @@ struct DynamicIndexVamana : public DynamicIndex { auto lock = read_lock(); return index.size(); } + + protected: + void track_id_range(std::span ids); + size_t delete_points_unlocked(std::span ids); + void compact_unlocked(size_t batchsize); +}; + +/// Dynamic index backed by svs::ConcurrentDynamicVamana. +/// +/// The index synchronizes add/delete/consolidate/compact itself, so they take the shared +/// lock alongside searches. Operations that are not thread-safe in the index (save, +/// set_num_threads) and whole-index reads (memory breakdown, use as a conversion source) +/// take the exclusive lock. +struct ConcurrentIndexVamana : public DynamicIndexVamana { + ConcurrentIndexVamana( + const IndexBuilder& builder, svs::DynamicVamana&& index, svs_sync_kind_t sync_kind + ); + + std::pair, std::vector> search( + svs::data::ConstSimpleDataView queries, + size_t num_neighbors, + const std::shared_ptr& search_params, + const IDFilterInterface* id_filter + ) override; + + size_t add_points( + svs::data::ConstSimpleDataView new_points, std::span ids + ) override; + + size_t delete_points(std::span ids) override; + + void consolidate() override { + auto lock = read_lock(); + index.consolidate(); + } + + void compact(size_t batchsize) override { + // exclusive lock prevents potential deadlock in case of custom threadpool takes + // internal mutex in parallel_for method + auto lock = write_lock(); + compact_unlocked(batchsize); + } + + svs::index::vamana::MemoryBreakdown get_memory_breakdown() const override { + auto lock = write_lock(); + return index.get_memory_breakdown(); + } }; } // namespace svs::c_runtime diff --git a/bindings/c/src/svs_c.cpp b/bindings/c/src/svs_c.cpp index 809dc8b33..6718861bf 100644 --- a/bindings/c/src/svs_c.cpp +++ b/bindings/c/src/svs_c.cpp @@ -662,9 +662,7 @@ namespace { void set_memory_breakdown( svs_memory_breakdown_t* out_breakdown, - size_t graph_bytes, - size_t data_bytes, - size_t metadata_bytes + const svs::index::vamana::MemoryBreakdown& breakdown ) { using namespace svs::c_runtime; INVALID_ARGUMENT_IF( @@ -676,6 +674,10 @@ void set_memory_breakdown( "Incompatible svs_memory_breakdown_t struct_size" ); + // The concurrent index's reverse edges are reported as part of the graph. + const size_t graph_bytes = breakdown.graph_bytes + breakdown.reverse_edges_bytes; + const size_t data_bytes = breakdown.data_bytes; + const size_t metadata_bytes = breakdown.metadata_bytes; if (out_breakdown->struct_size >= offsetof(svs_memory_breakdown_t, graph_bytes) + sizeof(out_breakdown->graph_bytes)) { out_breakdown->graph_bytes = graph_bytes; @@ -758,12 +760,7 @@ extern "C" bool svs_index_builder_estimate_memory( auto builder_ptr = builder->impl; INVALID_ARGUMENT_IF(builder_ptr == nullptr, "Invalid index builder handle"); auto breakdown = builder_ptr->estimate_memory_breakdown(num_vectors); - set_memory_breakdown( - out_breakdown, - breakdown.graph_bytes, - breakdown.data_bytes, - breakdown.metadata_bytes - ); + set_memory_breakdown(out_breakdown, breakdown); return true; }, out_err, @@ -808,12 +805,7 @@ extern "C" bool svs_index_builder_estimate_memory_dynamic( auto breakdown = builder_ptr->estimate_memory_breakdown_dynamic( num_vectors, make_blocking_parameters(blocksize_bytes) ); - set_memory_breakdown( - out_breakdown, - breakdown.graph_bytes, - breakdown.data_bytes, - breakdown.metadata_bytes - ); + set_memory_breakdown(out_breakdown, breakdown); return true; }, out_err, @@ -838,14 +830,9 @@ extern "C" bool svs_index_builder_estimate_memory_dynamic_ex( INVALID_ARGUMENT_IF(builder_ptr == nullptr, "Invalid index builder handle"); auto dynamic_params = read_dynamic_index_params(params); auto breakdown = builder_ptr->estimate_memory_breakdown_dynamic( - num_vectors, dynamic_params.block_params - ); - set_memory_breakdown( - out_breakdown, - breakdown.graph_bytes, - breakdown.data_bytes, - breakdown.metadata_bytes + num_vectors, dynamic_params.block_params, dynamic_params.sync_kind ); + set_memory_breakdown(out_breakdown, breakdown); return true; }, out_err, @@ -1795,12 +1782,7 @@ extern "C" bool svs_index_get_memory_breakdown( auto& index_ptr = index->impl; INVALID_ARGUMENT_IF(index_ptr == nullptr, "Invalid index handle"); auto breakdown = index_ptr->get_memory_breakdown(); - set_memory_breakdown( - out_breakdown, - breakdown.graph_bytes, - breakdown.data_bytes, - breakdown.metadata_bytes - ); + set_memory_breakdown(out_breakdown, breakdown); return true; }, out_err, diff --git a/bindings/c/src/threadpool.hpp b/bindings/c/src/threadpool.hpp index 6f2ae55ae..01cd07994 100644 --- a/bindings/c/src/threadpool.hpp +++ b/bindings/c/src/threadpool.hpp @@ -159,11 +159,16 @@ class ThreadPoolBuilder { num_threads = new_num_threads; } - svs::threads::ThreadPoolHandle build() const { + svs::threads::ThreadPoolHandle + build(svs_sync_kind_t sync_kind = SVS_SYNC_KIND_NONE) const { using namespace svs::threads; switch (kind) { case SVS_THREADPOOL_KIND_NATIVE: - return ThreadPoolHandle(NativeThreadPool(num_threads)); + if (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) { + return ThreadPoolHandle(QueueThreadPoolWrapper(num_threads)); + } else { + return ThreadPoolHandle(NativeThreadPool(num_threads)); + } case SVS_THREADPOOL_KIND_OMP: #if SVS_OMP // OMPThreadPool is only available if OpenMP support is enabled. diff --git a/bindings/c/tests/c_api_dynamic_index_sync.cpp b/bindings/c/tests/c_api_dynamic_index_sync.cpp index b65b1c077..9d4e753c2 100644 --- a/bindings/c/tests/c_api_dynamic_index_sync.cpp +++ b/bindings/c/tests/c_api_dynamic_index_sync.cpp @@ -126,16 +126,19 @@ bool is_known_id(size_t id, size_t num_writer_ids) { void run_concurrent_workload(SyncFixture& fx, svs_index_h index) { constexpr size_t NUM_WRITERS = 2; constexpr size_t NUM_READERS = 4; - constexpr size_t ITERATIONS = 20; + constexpr size_t ITERATIONS = 10; constexpr size_t READER_ITERATIONS = 40; constexpr size_t BATCH = 8; constexpr size_t NUM_WRITER_IDS = NUM_WRITERS * ITERATIONS * BATCH; std::vector batch_data; - generate_test_data(batch_data, BATCH, DIMENSION); + generate_test_data(batch_data, BATCH * ITERATIONS, DIMENSION); FailureLog log; + svs_sync_kind_t sync_kind = SVS_SYNC_KIND_NONE; + CATCH_REQUIRE(svs_index_dynamic_get_sync_kind(index, &sync_kind, fx.error)); + auto writer = [&](size_t w) { svs_error_h err = svs_error_create(); std::vector batch_ids(BATCH); @@ -146,24 +149,53 @@ void run_concurrent_workload(SyncFixture& fx, svs_index_h index) { } size_t added = 0; - log.check( - svs_index_dynamic_add_points( - index, batch_data.data(), batch_ids.data(), BATCH, &added, err - ), - err, - "add_points" - ); - log.check(added == BATCH, err, "add_points count"); + float* batch_ptr = batch_data.data() + iter * BATCH * DIMENSION; + // FINE_GRAIN dynamic index only supports adding one point at a time. + if (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) { + for (size_t i = 0; i < BATCH; ++i) { + log.check( + svs_index_dynamic_add_points( + index, batch_ptr + i * DIMENSION, &batch_ids[i], 1, &added, err + ), + err, + "add_points" + ); + log.check(added == 1, err, "add_points count"); + } + } else { + log.check( + svs_index_dynamic_add_points( + index, batch_ptr, batch_ids.data(), BATCH, &added, err + ), + err, + "add_points" + ); + log.check(added == BATCH, err, "add_points count"); + } size_t deleted = 0; - log.check( - svs_index_dynamic_delete_points( - index, batch_ids.data(), BATCH / 2, &deleted, err - ), - err, - "delete_points" - ); - log.check(deleted == BATCH / 2, err, "delete_points count"); + // FINE_GRAIN dynamic index only supports deleting one point at a time. + if (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) { + for (size_t i = 0; i < BATCH / 2; ++i) { + log.check( + svs_index_dynamic_delete_points( + index, &batch_ids[i], 1, &deleted, err + ), + err, + "delete_points" + ); + log.check(deleted == 1, err, "delete_points count"); + } + } else { + log.check( + svs_index_dynamic_delete_points( + index, batch_ids.data(), BATCH / 2, &deleted, err + ), + err, + "delete_points" + ); + log.check(deleted == BATCH / 2, err, "delete_points count"); + } if (iter % 5 == 4) { log.check(svs_index_dynamic_consolidate(index, err), err, "consolidate"); @@ -182,17 +214,41 @@ void run_concurrent_workload(SyncFixture& fx, svs_index_h index) { for (size_t iter = 0; iter < READER_ITERATIONS; ++iter) { std::this_thread::sleep_for(std::chrono::microseconds(pause_us(rng))); - bool ok = svs_index_search_topk( - index, fx.queries.data(), NUM_QUERIES, K, &results, nullptr, nullptr, err - ); - log.check(ok, err, "search_topk"); - if (ok) { - for (size_t i = 0; i < results.total_results; ++i) { - log.check( - is_known_id(results.indices[i], NUM_WRITER_IDS), - err, - "search returned unknown id" - ); + // FINE_GRAIN dynamic index only supports one query at a time. + if (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) { + bool ok = svs_index_search_topk( + index, fx.queries.data(), 1, K, &results, nullptr, nullptr, err + ); + log.check(ok, err, "search_topk"); + if (ok) { + for (size_t i = 0; i < results.total_results; ++i) { + log.check( + is_known_id(results.indices[i], NUM_WRITER_IDS), + err, + "search returned unknown id" + ); + } + } + } else { + bool ok = svs_index_search_topk( + index, + fx.queries.data(), + NUM_QUERIES, + K, + &results, + nullptr, + nullptr, + err + ); + log.check(ok, err, "search_topk"); + if (ok) { + for (size_t i = 0; i < results.total_results; ++i) { + log.check( + is_known_id(results.indices[i], NUM_WRITER_IDS), + err, + "search returned unknown id" + ); + } } } @@ -436,8 +492,8 @@ CATCH_TEST_CASE("C API Dynamic Index Params", "[c_api][index][dynamic][sync]") { fx.builder, NUM_VECTORS, ¶ms, &by_elements, fx.error )); CATCH_REQUIRE(svs_error_ok(fx.error)); - CATCH_REQUIRE(by_elements.data_bytes < by_bytes.data_bytes); - CATCH_REQUIRE(by_elements.graph_bytes < by_bytes.graph_bytes); + CATCH_REQUIRE(by_elements.data_bytes <= by_bytes.data_bytes); + CATCH_REQUIRE(by_elements.graph_bytes <= by_bytes.graph_bytes); size_t legacy_search = 0; CATCH_REQUIRE(svs_index_builder_estimate_search_memory_dynamic( @@ -489,22 +545,28 @@ CATCH_TEST_CASE("C API Dynamic Index Sync Sequential", "[c_api][index][dynamic][ std::vector new_data; generate_test_data(new_data, new_ids.size(), DIMENSION); size_t count = 0; + // FINE GRAIN dynamic index only supports adding one point at a time. + // If the sync kind is FINE GRAIN, we should only add one point at a time. + size_t points_to_add = (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) ? 1 : new_ids.size(); CATCH_REQUIRE(svs_index_dynamic_add_points( - index, new_data.data(), new_ids.data(), new_ids.size(), &count, fx.error + index, new_data.data(), new_ids.data(), points_to_add, &count, fx.error )); - CATCH_REQUIRE(count == new_ids.size()); - CATCH_REQUIRE( - svs_index_dynamic_delete_points(index, new_ids.data(), 1, &count, fx.error) - ); - CATCH_REQUIRE(count == 1); + CATCH_REQUIRE(count == points_to_add); + size_t points_to_delete = 1; + CATCH_REQUIRE(svs_index_dynamic_delete_points( + index, new_ids.data(), points_to_delete, &count, fx.error + )); + CATCH_REQUIRE(count == points_to_delete); bool has_id = true; CATCH_REQUIRE(svs_index_dynamic_has_id(index, new_ids[0], &has_id, fx.error)); CATCH_REQUIRE_FALSE(has_id); svs_search_results_t results = SVS_INIT_SEARCH_RESULTS(); + // FINE GRAIN dynamic index only supports one query at a time. + size_t queries_to_search = (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) ? 1 : NUM_QUERIES; CATCH_REQUIRE(svs_index_search_topk( - index, fx.queries.data(), NUM_QUERIES, K, &results, nullptr, nullptr, fx.error + index, fx.queries.data(), queries_to_search, K, &results, nullptr, nullptr, fx.error )); svs_search_results_free(&results); @@ -540,7 +602,7 @@ CATCH_TEST_CASE("C API Dynamic Index Sync Sequential", "[c_api][index][dynamic][ size_t size = 0; CATCH_REQUIRE(svs_index_get_size(loaded, &size, fx.error)); - CATCH_REQUIRE(size == NUM_VECTORS + 1); + CATCH_REQUIRE(size == NUM_VECTORS + points_to_add - points_to_delete); CATCH_REQUIRE(svs_index_dynamic_get_sync_kind(loaded, &actual_kind, fx.error)); CATCH_REQUIRE(actual_kind == sync_kind); @@ -553,13 +615,224 @@ CATCH_TEST_CASE("C API Dynamic Index Sync Sequential", "[c_api][index][dynamic][ CATCH_REQUIRE(svs_index_dynamic_get_sync_kind(converted, &actual_kind, fx.error)); CATCH_REQUIRE(actual_kind == SVS_SYNC_KIND_GLOBAL); CATCH_REQUIRE(svs_index_get_size(converted, &size, fx.error)); - CATCH_REQUIRE(size == NUM_VECTORS + 1); + CATCH_REQUIRE(size == NUM_VECTORS + points_to_add - points_to_delete); svs_index_free(converted); svs_index_free(loaded); svs_index_free(index); } +CATCH_TEST_CASE( + "C API Dynamic Index Sync Kind Conversion", "[c_api][index][dynamic][sync]" +) { + // FINE_GRAIN uses a different index implementation; convert between all pairs. + const auto src_kind = GENERATE(SVS_SYNC_KIND_NONE, SVS_SYNC_KIND_FINE_GRAIN); + const auto dst_kind = GENERATE(SVS_SYNC_KIND_NONE, SVS_SYNC_KIND_FINE_GRAIN); + CATCH_CAPTURE(src_kind, dst_kind); + + SyncFixture fx; + svs_index_h source = fx.build(src_kind); + CATCH_REQUIRE(source != nullptr); + + // Leave deletions unconsolidated so the copy must skip them. + // FINE_GRAIN dynamic index only supports deleting one point at a time. + const size_t NUM_DELETED = src_kind == SVS_SYNC_KIND_FINE_GRAIN ? 1 : 10; + size_t count = 0; + CATCH_REQUIRE(svs_index_dynamic_delete_points( + source, fx.ids.data(), NUM_DELETED, &count, fx.error + )); + CATCH_REQUIRE(count == NUM_DELETED); + + svs_dynamic_index_params_t params = SVS_INIT_DYNAMIC_INDEX_PARAMS(); + params.blocksize_bytes = BLOCK_SIZE; + params.sync_kind = dst_kind; + svs_index_h converted = + svs_index_convert_dynamic_ex(fx.builder, source, ¶ms, fx.error); + CATCH_REQUIRE(converted != nullptr); + CATCH_REQUIRE(svs_error_ok(fx.error)); + + size_t size = 0; + CATCH_REQUIRE(svs_index_get_size(converted, &size, fx.error)); + CATCH_REQUIRE(size == NUM_VECTORS - NUM_DELETED); + for (size_t i = 0; i < NUM_VECTORS; ++i) { + bool has_id = false; + CATCH_REQUIRE(svs_index_dynamic_has_id(converted, fx.ids[i], &has_id, fx.error)); + CATCH_REQUIRE(has_id == (i >= NUM_DELETED)); + } + + // The converted index stays fully usable. + svs_search_results_t results = SVS_INIT_SEARCH_RESULTS(); + // FINE_GRAIN dynamic index only supports searching one query at a time. + size_t queries_to_search = (dst_kind == SVS_SYNC_KIND_FINE_GRAIN) ? 1 : NUM_QUERIES; + CATCH_REQUIRE(svs_index_search_topk( + converted, + fx.queries.data(), + queries_to_search, + K, + &results, + nullptr, + nullptr, + fx.error + )); + for (size_t i = 0; i < results.total_results; ++i) { + CATCH_REQUIRE(results.indices[i] >= NUM_DELETED); + CATCH_REQUIRE(results.indices[i] < NUM_VECTORS); + } + svs_search_results_free(&results); + + std::vector new_ids = {fx.ids[0], NUM_VECTORS}; + std::vector new_data; + generate_test_data(new_data, new_ids.size(), DIMENSION); + // FINE_GRAIN dynamic index only supports adding one point at a time. + size_t points_to_add = (dst_kind == SVS_SYNC_KIND_FINE_GRAIN) ? 1 : new_ids.size(); + CATCH_REQUIRE(svs_index_dynamic_add_points( + converted, new_data.data(), new_ids.data(), points_to_add, &count, fx.error + )); + CATCH_REQUIRE(count == points_to_add); + CATCH_REQUIRE(svs_index_dynamic_consolidate(converted, fx.error)); + CATCH_REQUIRE(svs_index_dynamic_compact(converted, 0, fx.error)); + CATCH_REQUIRE(svs_index_get_size(converted, &size, fx.error)); + CATCH_REQUIRE(size == NUM_VECTORS - NUM_DELETED + points_to_add); + + svs_index_free(converted); + svs_index_free(source); +} + +CATCH_TEST_CASE("C API Dynamic Index Fine Grain Storage", "[c_api][index][dynamic][sync]") { + const int storage_id = GENERATE(0, 1, 2, 3, 4); + CATCH_CAPTURE(storage_id); + + SyncFixture fx; + svs_storage_h storage = nullptr; + switch (storage_id) { + case 0: + storage = svs_storage_create_simple(SVS_DATA_TYPE_FLOAT16, fx.error); + break; + case 1: + storage = svs_storage_create_sq(SVS_DATA_TYPE_INT8, fx.error); + break; + case 2: + storage = + svs_storage_create_lvq(SVS_DATA_TYPE_INT4, SVS_DATA_TYPE_INT8, fx.error); + break; + case 3: + storage = + svs_storage_create_lvq(SVS_DATA_TYPE_INT8, SVS_DATA_TYPE_VOID, fx.error); + break; + default: + storage = svs_storage_create_leanvec( + DIMENSION / 2, SVS_DATA_TYPE_INT4, SVS_DATA_TYPE_INT8, fx.error + ); + break; + } + CATCH_REQUIRE(check_storage_support(storage, fx.error)); + if (!storage_usable(storage)) { + svs_storage_free(storage); + return; + } + + svs_index_builder_h builder = svs_index_builder_create( + SVS_DISTANCE_METRIC_EUCLIDEAN, DIMENSION, fx.algorithm, fx.error + ); + CATCH_REQUIRE(builder != nullptr); + CATCH_REQUIRE(svs_index_builder_set_storage(builder, storage, fx.error)); + CATCH_REQUIRE( + svs_index_builder_set_threadpool(builder, SVS_THREADPOOL_KIND_NATIVE, 2, fx.error) + ); + + svs_dynamic_index_params_t params = SVS_INIT_DYNAMIC_INDEX_PARAMS(); + params.blocksize_bytes = BLOCK_SIZE; + params.sync_kind = SVS_SYNC_KIND_FINE_GRAIN; + svs_index_h index = svs_index_build_dynamic_ex( + builder, fx.data.data(), fx.ids.data(), NUM_VECTORS, ¶ms, fx.error + ); + CATCH_REQUIRE(index != nullptr); + CATCH_REQUIRE(svs_error_ok(fx.error)); + + run_concurrent_workload(fx, index); + + // Save and load keep the concurrent index usable. + TempDir dir; + CATCH_REQUIRE(svs_index_save(index, dir.string().c_str(), fx.error)); + svs_index_h loaded = + svs_index_load_dynamic_ex(builder, dir.string().c_str(), ¶ms, fx.error); + CATCH_REQUIRE(loaded != nullptr); + CATCH_REQUIRE(svs_error_ok(fx.error)); + size_t expected_size = 0; + size_t size = 0; + CATCH_REQUIRE(svs_index_get_size(index, &expected_size, fx.error)); + CATCH_REQUIRE(svs_index_get_size(loaded, &size, fx.error)); + CATCH_REQUIRE(size == expected_size); + + svs_search_results_t results = SVS_INIT_SEARCH_RESULTS(); + // FINE_GRAIN dynamic index only supports one query at a time. + CATCH_REQUIRE(svs_index_search_topk( + loaded, fx.queries.data(), 1, K, &results, nullptr, nullptr, fx.error + )); + svs_search_results_free(&results); + + // Compressed (concurrent) -> simple (regular), then simple (concurrent) -> compressed. + params.sync_kind = SVS_SYNC_KIND_NONE; + svs_index_h decompressed = + svs_index_convert_dynamic_ex(fx.builder, loaded, ¶ms, fx.error); + CATCH_REQUIRE(decompressed != nullptr); + CATCH_REQUIRE(svs_index_get_size(decompressed, &size, fx.error)); + CATCH_REQUIRE(size == expected_size); + + svs_index_h simple = fx.build(SVS_SYNC_KIND_FINE_GRAIN); + CATCH_REQUIRE(simple != nullptr); + params.sync_kind = SVS_SYNC_KIND_FINE_GRAIN; + svs_index_h compressed = + svs_index_convert_dynamic_ex(builder, simple, ¶ms, fx.error); + CATCH_REQUIRE(compressed != nullptr); + CATCH_REQUIRE(svs_index_get_size(compressed, &size, fx.error)); + CATCH_REQUIRE(size == NUM_VECTORS); + + svs_index_free(compressed); + svs_index_free(simple); + svs_index_free(decompressed); + svs_index_free(loaded); + svs_index_free(index); + svs_index_builder_free(builder); + svs_storage_free(storage); +} + +CATCH_TEST_CASE("C API Dynamic Index Sync Kind Memory", "[c_api][index][dynamic][sync]") { + SyncFixture fx; + + auto estimate = [&](svs_sync_kind_t kind) { + svs_dynamic_index_params_t params = SVS_INIT_DYNAMIC_INDEX_PARAMS(); + params.blocksize_bytes = BLOCK_SIZE; + params.sync_kind = kind; + svs_memory_breakdown_t breakdown = SVS_INIT_MEMORY_BREAKDOWN(); + CATCH_REQUIRE(svs_index_builder_estimate_memory_dynamic_ex( + fx.builder, NUM_VECTORS, ¶ms, &breakdown, fx.error + )); + return breakdown; + }; + + const auto regular = estimate(SVS_SYNC_KIND_NONE); + const auto global = estimate(SVS_SYNC_KIND_GLOBAL); + const auto fine_grain = estimate(SVS_SYNC_KIND_FINE_GRAIN); + CATCH_REQUIRE(global.graph_bytes == regular.graph_bytes); + // The concurrent index also keeps reverse edges, counted as graph memory. + CATCH_REQUIRE(fine_grain.graph_bytes > regular.graph_bytes); + CATCH_REQUIRE(fine_grain.data_bytes == regular.data_bytes); + CATCH_REQUIRE(fine_grain.metadata_bytes == regular.metadata_bytes); + + svs_index_h index = fx.build(SVS_SYNC_KIND_FINE_GRAIN); + CATCH_REQUIRE(index != nullptr); + svs_memory_breakdown_t breakdown = SVS_INIT_MEMORY_BREAKDOWN(); + CATCH_REQUIRE(svs_index_get_memory_breakdown(index, &breakdown, fx.error)); + size_t usage = 0; + CATCH_REQUIRE(svs_index_get_memory_usage(index, &usage, fx.error)); + CATCH_REQUIRE( + usage == breakdown.graph_bytes + breakdown.data_bytes + breakdown.metadata_bytes + ); + CATCH_REQUIRE(breakdown.graph_bytes > 0); + svs_index_free(index); +} + CATCH_TEST_CASE("C API Dynamic Index Sync Concurrent", "[c_api][index][dynamic][sync]") { const auto sync_kind = GENERATE(SVS_SYNC_KIND_GLOBAL, SVS_SYNC_KIND_FINE_GRAIN); CATCH_CAPTURE(sync_kind); diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index 3ec385b91..b9f7e3361 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -49,8 +49,6 @@ class ConcurrentDynamicVamana : public DynamicVamana { using base_type = DynamicVamana; using default_graph_allocator_type = index::vamana::concurrent::SegmentedBlocked>; - using default_graph_type = - index::vamana::concurrent::graphs::SimpleBlockedGraph; template explicit ConcurrentDynamicVamana(AssembleTag tag, QueryTypes types, Impl impl) @@ -80,7 +78,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { std::span ids, Distance distance, ThreadPoolProto threadpool_proto, - const GraphAllocator& graph_allocator = {} + const GraphAllocator& graph_allocator = {}, + svs::logging::logger_ptr logger = svs::logging::get() ) { auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); auto data = @@ -95,7 +94,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { ids, std::move(distance_function), std::move(threadpool), - graph_allocator + graph_allocator, + std::move(logger) ) ); }; @@ -130,7 +130,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { DataLoader&& data_loader, Distance distance, ThreadPoolProto threadpool_proto, - bool debug_load_from_static = false + bool debug_load_from_static = false, + svs::logging::logger_ptr logger = svs::logging::get() ) requires(!std::is_base_of_v> && !is_stream_loader_v> && !is_stream_loader_v>) { @@ -145,7 +146,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { std::forward(data_loader), std::move(distance_function), std::move(threadpool), - debug_load_from_static + debug_load_from_static, + std::move(logger) ) ); }; @@ -180,7 +182,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { const DataLoader& data_loader, Distance distance, ThreadPoolProto threadpool_proto, - bool debug_load_from_static = false + bool debug_load_from_static = false, + svs::logging::logger_ptr logger = svs::logging::get() ) requires(is_stream_loader_v && is_stream_loader_v) { @@ -204,7 +207,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { [&]() -> Data { return data_loader.load(stream); }, std::move(distance_function), std::move(threadpool), - debug_load_from_static + debug_load_from_static, + std::move(logger) ) ); }; @@ -237,7 +241,8 @@ class ConcurrentDynamicVamana : public DynamicVamana { data_loader.load(data_path), std::move(distance), std::move(threadpool), - debug_load_from_static + debug_load_from_static, + std::move(logger) ); } };