From cfdbd7ef9bbabeb88535ad871a4eaa219776b5d9 Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Fri, 9 Oct 2026 06:04:30 -0700 Subject: [PATCH 1/9] Refactor ConcurrentDynamicVamana to use logger --- .../orchestrators/concurrent_dynamic_vamana.h | 26 ++++++++++--------- 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index 7167f302..4ad03be7 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() ) { auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); auto make = [&](auto distance_function) { @@ -143,7 +144,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) ) ); }; @@ -173,11 +175,13 @@ class ConcurrentDynamicVamana : public DynamicVamana { ThreadPoolProto threadpool_proto, DataLoaderArgs&&... data_args ) { + namespace cc = index::vamana::concurrent; static_assert( - index::vamana::concurrent::is_segmented_blocked_v< - typename Data::allocator_type>, + cc::is_segmented_blocked_v, "The concurrent index requires a dataset with a SegmentedBlocked allocator." ); + using graph_type = cc::graphs::SimpleBlockedGraph; + auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); auto deserializer = svs::lib::detail::Deserializer::build(stream); if (deserializer.is_native()) { @@ -187,9 +191,7 @@ class ConcurrentDynamicVamana : public DynamicVamana { manager::as_typelist(), index::vamana::concurrent::auto_dynamic_assemble( stream, - [&]() -> default_graph_type { - return default_graph_type::load(stream); - }, + [&]() -> graph_type { return graph_type::load(stream); }, [&]() -> Data { return lib::load_from_stream( stream, SVS_FWD(data_args)... @@ -225,7 +227,7 @@ class ConcurrentDynamicVamana : public DynamicVamana { return assemble( config_path, - SVS_LAZY(default_graph_type::load(graph_path)), + SVS_LAZY(graph_type::load(graph_path)), lib::load_from_disk(data_path, SVS_FWD(data_args)...), std::move(distance), std::move(threadpool) From 04bd38d2db19bd332773adf4e8e8c08542a01c25 Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Thu, 15 Oct 2026 09:19:15 -0700 Subject: [PATCH 2/9] [C API] Add SVS_SYNC_KIND_FINE_GRAIN support using ConcurrentDynamicVamana - Added atomic tracking for minimum and maximum IDs in DynamicIndexVamana to support concurrent operations. - Introduced ConcurrentIndexVamana to manage concurrent access and operations on the dynamic index. - Updated memory estimation functions to account for new synchronization parameters. - Implemented comprehensive tests for synchronization kind conversion and memory usage. - Created dispatcher for concurrent Vamana index build and load operations. - Added support for copying between concurrent and regular dynamic indices. --- bindings/c/CMakeLists.txt | 3 + bindings/c/docs/C_API_Design.md | 15 +- bindings/c/include/svs/c/svs_c.h | 25 +- bindings/c/src/allocator.hpp | 35 +- bindings/c/src/data_builder/leanvec.hpp | 12 +- bindings/c/src/data_builder/lvq.hpp | 12 +- bindings/c/src/data_builder/simple.hpp | 7 +- bindings/c/src/data_builder/sq.hpp | 9 +- .../c/src/dispatcher_concurrent_vamana.cpp | 346 ++++++++++++++++++ .../c/src/dispatcher_concurrent_vamana.hpp | 81 ++++ bindings/c/src/dispatcher_dynamic_vamana.cpp | 201 ++-------- bindings/c/src/dispatcher_dynamic_vamana.hpp | 1 + .../c/src/dispatcher_dynamic_vamana_copy.hpp | 235 ++++++++++++ bindings/c/src/index_builder.cpp | 167 ++++++--- bindings/c/src/index_builder.hpp | 4 +- bindings/c/src/index_vamana.cpp | 58 ++- bindings/c/src/index_vamana.hpp | 45 ++- bindings/c/src/svs_c.cpp | 38 +- bindings/c/tests/c_api_dynamic_index_sync.cpp | 197 ++++++++++ 19 files changed, 1193 insertions(+), 298 deletions(-) create mode 100644 bindings/c/src/dispatcher_concurrent_vamana.cpp create mode 100644 bindings/c/src/dispatcher_concurrent_vamana.hpp create mode 100644 bindings/c/src/dispatcher_dynamic_vamana_copy.hpp diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 51a5541b..1cb0b206 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 12616d15..8f3ccae1 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 6ed11d18..41d09799 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -586,23 +586,31 @@ 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. /// * 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. @@ -1158,7 +1166,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 diff --git a/bindings/c/src/allocator.hpp b/bindings/c/src/allocator.hpp index 9b09c700..4af48963 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,37 @@ 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; + +// Accepts a BlockKind, or a bool meaning Blocked (true) / None (false). +template +inline constexpr BlockKind block_kind_v = [] { + if constexpr (std::is_same_v) { + return Kind ? BlockKind::Blocked : BlockKind::None; + } else { + return Kind; + } +}(); class AllocatorBuilder { template class CustomAllocator { diff --git a/bindings/c/src/data_builder/leanvec.hpp b/bindings/c/src/data_builder/leanvec.hpp index 62bad86d..39415163 100644 --- a/bindings/c/src/data_builder/leanvec.hpp +++ b/bindings/c/src/data_builder/leanvec.hpp @@ -41,6 +41,8 @@ #include #endif // SVS_LEANVEC_HEADER +#include + #include #include #include @@ -172,10 +174,10 @@ struct lib:: } }; -template void for_leanvec_specializations(F&& f) { - using byte_alloc = svs::c_runtime:: - MaybeBlockedAlloc>; - +template void for_leanvec_specializations(F&& f) { + constexpr auto Kind = c_runtime::block_kind_v; + 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 f063b451..c6cb3482 100644 --- a/bindings/c/src/data_builder/lvq.hpp +++ b/bindings/c/src/data_builder/lvq.hpp @@ -39,6 +39,9 @@ #include #endif // SVS_LVQ_HEADER +#include "allocator.hpp" +#include + #include #include #include @@ -160,9 +163,10 @@ 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) { + constexpr auto Kind = c_runtime::block_kind_v; + 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 +184,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 070da551..b542a5ce 100644 --- a/bindings/c/src/data_builder/simple.hpp +++ b/bindings/c/src/data_builder/simple.hpp @@ -92,11 +92,12 @@ struct lib::DispatchConverter void for_simple_specializations(F&& f) { +template void for_simple_specializations(F&& f) { + constexpr auto Kind = c_runtime::block_kind_v; 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 571b8cdb..ee5dbc30 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,12 @@ 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) { + constexpr auto Kind = c_runtime::block_kind_v; 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 00000000..c29bfbb9 --- /dev/null +++ b/bindings/c/src/dispatcher_concurrent_vamana.cpp @@ -0,0 +1,346 @@ +/* + * 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; + + 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, + AllocatorHandle graph_handle, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger + ) { + auto graph = ConcurrentGraph( + src_graph.n_nodes(), + config.build_parameters.graph_max_degree, + ConcurrentGraphAllocator{block_params, std::move(graph_handle)} + ); + 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_simple_specializations(build_closure); + for_simple_specializations(load_closure); + for_leanvec_specializations(build_closure); + for_leanvec_specializations(load_closure); + for_lvq_specializations(build_closure); + for_lvq_specializations(load_closure); + for_sq_specializations(build_closure); + for_sq_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. + using index_type = uint32_t; + breakdown.reverse_edges_bytes = + num_vectors * (sizeof(std::vector) + sizeof(cc::SpinLock) + + build_params.graph_max_degree * sizeof(index_type)); + 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 00000000..0710e88b --- /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 9686ebc2..7ac7c961 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, @@ -142,18 +139,19 @@ 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_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); } // Stream load alternative, matched via generic variant DispatchConverter like the @@ -175,7 +173,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; @@ -183,149 +181,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( @@ -405,17 +260,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 ca3a382f..47468fbf 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 00000000..8bf23061 --- /dev/null +++ b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp @@ -0,0 +1,235 @@ +/* + * 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_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, + AllocatorHandle graph_handle, + const svs::data::BlockingParameters& block_params, + svs::logging::logger_ptr logger + ) { + auto graph = graph_type( + src_graph.n_nodes(), + config.build_parameters.graph_max_degree, + svs::data::Blocked{block_params, std::move(graph_handle)} + ); + 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)); + } + } + + const 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); + + 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), + allocator_builder.build_for_graph(), + block_params, + 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_simple_specializations(inner); + for_sq_specializations(inner); + for_lvq_specializations(inner); + for_leanvec_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/index_builder.cpp b/bindings/c/src/index_builder.cpp index 14db878d..5ed7b033 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(), + 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(), + 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(), + 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(), + 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(), + allocator_builder, + block_params, + get_logger() + ) + : dispatch_dynamic_vamana_index_load( + build_params, + directory, + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(), + allocator_builder, + block_params, + get_logger() + ); + return make_dynamic_index(*this, std::move(index), sync_kind); } return nullptr; } @@ -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 82eca106..8d15518c 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 bcaa00c3..41d97372 100644 --- a/bindings/c/src/index_vamana.cpp +++ b/bindings/c/src/index_vamana.cpp @@ -139,7 +139,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 +152,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 +177,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 +187,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 +235,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 { @@ -237,4 +252,27 @@ void DynamicIndexVamana::set_num_threads(size_t num_threads) { index.set_threadpool(this->builder->pool_builder.build()); } +///////////////////////////////////////// +// ConcurrentIndexVamana Implementation +ConcurrentIndexVamana::ConcurrentIndexVamana( + const IndexBuilder& builder, svs::DynamicVamana&& index, svs_sync_kind_t sync_kind +) + : DynamicIndexVamana(builder, std::move(index), sync_kind) {} + +size_t ConcurrentIndexVamana::add_points( + svs::data::ConstSimpleDataView new_points, std::span ids +) { + 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) { + 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 80e5605e..f032e7bb 100644 --- a/bindings/c/src/index_vamana.hpp +++ b/bindings/c/src/index_vamana.hpp @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -75,8 +76,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 = 0; + std::atomic max_id = 0; DynamicIndexVamana( const IndexBuilder& builder, svs::DynamicVamana&& index, @@ -144,5 +147,43 @@ 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 + ); + + 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 { + auto lock = read_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 18482210..5b37c01d 100644 --- a/bindings/c/src/svs_c.cpp +++ b/bindings/c/src/svs_c.cpp @@ -664,9 +664,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( @@ -678,6 +676,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; @@ -760,12 +762,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, @@ -810,12 +807,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, @@ -840,14 +832,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, @@ -1797,12 +1784,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/tests/c_api_dynamic_index_sync.cpp b/bindings/c/tests/c_api_dynamic_index_sync.cpp index b65b1c07..43d0d908 100644 --- a/bindings/c/tests/c_api_dynamic_index_sync.cpp +++ b/bindings/c/tests/c_api_dynamic_index_sync.cpp @@ -560,6 +560,203 @@ CATCH_TEST_CASE("C API Dynamic Index Sync Sequential", "[c_api][index][dynamic][ 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. + constexpr size_t NUM_DELETED = 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(); + CATCH_REQUIRE(svs_index_search_topk( + converted, fx.queries.data(), NUM_QUERIES, 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); + CATCH_REQUIRE(svs_index_dynamic_add_points( + converted, new_data.data(), new_ids.data(), new_ids.size(), &count, fx.error + )); + CATCH_REQUIRE(count == new_ids.size()); + 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 + new_ids.size()); + + 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)) { + 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(); + CATCH_REQUIRE(svs_index_search_topk( + loaded, fx.queries.data(), NUM_QUERIES, 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); From 38f95615f733ac673b8271e598f63ef7a0cb07fe Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Tue, 13 Oct 2026 06:26:42 -0700 Subject: [PATCH 3/9] Some refactoring --- bindings/c/src/allocator.hpp | 10 --- bindings/c/src/data_builder.cpp | 62 ++++++------------- bindings/c/src/data_builder.hpp | 8 +++ bindings/c/src/data_builder/leanvec.hpp | 7 ++- bindings/c/src/data_builder/lvq.hpp | 7 ++- bindings/c/src/data_builder/simple.hpp | 3 +- bindings/c/src/data_builder/sq.hpp | 3 +- .../c/src/dispatcher_concurrent_vamana.cpp | 18 +++--- bindings/c/src/dispatcher_dynamic_vamana.cpp | 15 +---- .../c/src/dispatcher_dynamic_vamana_copy.hpp | 22 +++---- bindings/c/src/dispatcher_vamana.cpp | 32 ++++------ bindings/c/tests/c_api_dynamic_index_sync.cpp | 4 +- 12 files changed, 70 insertions(+), 121 deletions(-) diff --git a/bindings/c/src/allocator.hpp b/bindings/c/src/allocator.hpp index 4af48963..d191ac13 100644 --- a/bindings/c/src/allocator.hpp +++ b/bindings/c/src/allocator.hpp @@ -51,16 +51,6 @@ template struct BlockedAllocSelector> using MaybeBlockedAlloc = typename detail::BlockedAllocSelector::type; -// Accepts a BlockKind, or a bool meaning Blocked (true) / None (false). -template -inline constexpr BlockKind block_kind_v = [] { - if constexpr (std::is_same_v) { - return Kind ? BlockKind::Blocked : BlockKind::None; - } else { - return Kind; - } -}(); - class AllocatorBuilder { template class CustomAllocator { public: diff --git a/bindings/c/src/data_builder.cpp b/bindings/c/src/data_builder.cpp index 1cc1e060..303777db 100644 --- a/bindings/c/src/data_builder.cpp +++ b/bindings/c/src/data_builder.cpp @@ -22,35 +22,25 @@ #include #include +#include #include namespace svs::c_runtime { -// namespace { -template -size_t -estimate_size(DataBuilder builder, size_t num_vectors, size_t dimension, svs::lib::Empty) { - using allocator_type = typename DataBuilder::allocator_type; - static_assert( - !svs::data::is_blocked_v, - "estimate_size requires a non-blocked allocator type." - ); - return builder.estimate_size(num_vectors, dimension, allocator_type{}); -} +namespace { template -size_t estimate_blocked_size( +size_t estimate_size( DataBuilder builder, size_t num_vectors, size_t dimension, - svs::data::BlockingParameters block_params + const svs::data::BlockingParameters& block_params ) { using allocator_type = typename DataBuilder::allocator_type; - static_assert( - svs::data::is_blocked_v, - "estimate_blocked_size requires a blocked allocator type." - ); - auto allocator = allocator_type{block_params}; - return builder.estimate_size(num_vectors, dimension, allocator); + if constexpr (svs::data::is_blocked_v) { + return builder.estimate_size(num_vectors, dimension, allocator_type{block_params}); + } else { + return builder.estimate_size(num_vectors, dimension, allocator_type{}); + } } template @@ -59,25 +49,12 @@ 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); - - 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(size_closure); + for_all_specializations(size_closure); } -using BlocksizeArg = std::variant; - -using EstimateSizeDispatcher = - svs::lib::Dispatcher; +using EstimateSizeDispatcher = svs::lib:: + Dispatcher; const EstimateSizeDispatcher& build_data_size_dispatcher() { static EstimateSizeDispatcher dispatcher = [] { @@ -89,13 +66,16 @@ const EstimateSizeDispatcher& build_data_size_dispatcher() { } size_t dispatch_data_size_estimation( - const Storage* storage, size_t num_vectors, size_t dimension, BlocksizeArg block_params + const Storage* storage, + size_t num_vectors, + size_t dimension, + const svs::data::BlockingParameters& block_params = {} ) { return build_data_size_dispatcher().invoke( - storage, num_vectors, dimension, std::move(block_params) + storage, num_vectors, dimension, block_params ); } -//} // namespace +} // namespace size_t estimate_data_size(const Storage* storage, size_t num_vectors, size_t dimension) { if (storage == nullptr) { @@ -107,9 +87,7 @@ size_t estimate_data_size(const Storage* storage, size_t num_vectors, size_t dim if (dimension == 0) { throw std::invalid_argument("Dimension must be greater than zero."); } - return dispatch_data_size_estimation( - storage, num_vectors, dimension, svs::lib::Empty{} - ); + return dispatch_data_size_estimation(storage, num_vectors, dimension); } size_t estimate_data_size_blocked( diff --git a/bindings/c/src/data_builder.hpp b/bindings/c/src/data_builder.hpp index d341352b..cd8d5d27 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 39415163..98623f5c 100644 --- a/bindings/c/src/data_builder/leanvec.hpp +++ b/bindings/c/src/data_builder/leanvec.hpp @@ -19,6 +19,7 @@ #include "svs/c/svs_c.h" +#include "allocator.hpp" #include "data_builder/lvq.hpp" #include "leanvec_training_data.hpp" #include "storage.hpp" @@ -174,8 +175,7 @@ struct lib:: } }; -template void for_leanvec_specializations(F&& f) { - constexpr auto Kind = c_runtime::block_kind_v; +template void for_leanvec_specializations(F&& f) { using byte_alloc = svs::c_runtime::MaybeBlockedAlloc>; #define X(P, S, D) f.template operator(), D>(); @@ -192,9 +192,10 @@ template void for_leanvec_specializations(F&& f) { } // namespace svs #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled +#include "allocator.hpp" 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 c6cb3482..10dc96c5 100644 --- a/bindings/c/src/data_builder/lvq.hpp +++ b/bindings/c/src/data_builder/lvq.hpp @@ -19,6 +19,7 @@ #include "svs/c/svs_c.h" +#include "allocator.hpp" #include "storage.hpp" #include "types_support.hpp" @@ -163,8 +164,7 @@ struct lib::DispatchConverter< static To convert(From SVS_UNUSED(from)) { return To{}; } }; -template void for_lvq_specializations(F&& f) { - constexpr auto Kind = c_runtime::block_kind_v; +template void for_lvq_specializations(F&& f) { using byte_alloc = svs::c_runtime::MaybeBlockedAlloc>; #define X(P, S, D) f.template operator(), D>(); @@ -182,9 +182,10 @@ template void for_lvq_specializations(F&& f) { } // namespace svs #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled +#include "allocator.hpp" 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 b542a5ce..2744e8ff 100644 --- a/bindings/c/src/data_builder/simple.hpp +++ b/bindings/c/src/data_builder/simple.hpp @@ -92,8 +92,7 @@ struct lib::DispatchConverter void for_simple_specializations(F&& f) { - constexpr auto Kind = c_runtime::block_kind_v; +template void for_simple_specializations(F&& f) { using float_alloc = svs::c_runtime::MaybeBlockedAlloc>; using float16_alloc = svs::c_runtime:: diff --git a/bindings/c/src/data_builder/sq.hpp b/bindings/c/src/data_builder/sq.hpp index ee5dbc30..6d1957dd 100644 --- a/bindings/c/src/data_builder/sq.hpp +++ b/bindings/c/src/data_builder/sq.hpp @@ -100,8 +100,7 @@ struct lib::DispatchConverter static To convert(From SVS_UNUSED(from)) { return To{}; } }; -template void for_sq_specializations(F&& f) { - constexpr auto Kind = c_runtime::block_kind_v; +template void for_sq_specializations(F&& f) { using int8_alloc = svs::c_runtime::MaybeBlockedAlloc>; using uint8_alloc = diff --git a/bindings/c/src/dispatcher_concurrent_vamana.cpp b/bindings/c/src/dispatcher_concurrent_vamana.cpp index c29bfbb9..7fdb0c59 100644 --- a/bindings/c/src/dispatcher_concurrent_vamana.cpp +++ b/bindings/c/src/dispatcher_concurrent_vamana.cpp @@ -76,6 +76,8 @@ cc::SlotMetadata to_concurrent_slot(RegularSlot s) { struct ConcurrentVamanaFlavor { static constexpr BlockKind block_kind = BlockKind::Segmented; + using graph_allocator_type = ConcurrentGraphAllocator; + template using impl_type = cc::MutableVamanaIndex; @@ -99,14 +101,13 @@ struct ConcurrentVamanaFlavor { std::span internal_ids, Distance distance, svs::threads::ThreadPoolHandle pool, - AllocatorHandle graph_handle, - const svs::data::BlockingParameters& block_params, + graph_allocator_type graph_allocator, svs::logging::logger_ptr logger ) { auto graph = ConcurrentGraph( src_graph.n_nodes(), config.build_parameters.graph_max_degree, - ConcurrentGraphAllocator{block_params, std::move(graph_handle)} + std::move(graph_allocator) ); svs::data::copy(src_graph.get_data(), graph.get_data()); @@ -221,14 +222,9 @@ const BuildConcurrentIndexDispatcher& build_concurrent_vamana_index_dispatcher() }; constexpr auto kind = ConcurrentVamanaFlavor::block_kind; - for_simple_specializations(build_closure); - for_simple_specializations(load_closure); - for_leanvec_specializations(build_closure); - for_leanvec_specializations(load_closure); - for_lvq_specializations(build_closure); - for_lvq_specializations(load_closure); - for_sq_specializations(build_closure); - for_sq_specializations(load_closure); + + for_all_specializations(build_closure); + for_all_specializations(load_closure); return d; }(); return dispatcher; diff --git a/bindings/c/src/dispatcher_dynamic_vamana.cpp b/bindings/c/src/dispatcher_dynamic_vamana.cpp index 7ac7c961..96022007 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.cpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.cpp @@ -140,18 +140,9 @@ void register_dynamic_vamana_index_specializations(Dispatcher& dispatcher) { }; constexpr auto kind = BlockKind::Blocked; - 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 diff --git a/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp index 8bf23061..6b21e5b6 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp +++ b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp @@ -57,8 +57,8 @@ using CopyDynamicIndexDispatcher = svs::lib::Dispatcher< svs::logging::logger_ptr>; struct RegularVamanaFlavor { - using graph_type = - svs::graphs::SimpleGraph>>; + using graph_allocator_type = svs::data::Blocked>; + using graph_type = svs::graphs::SimpleGraph; static constexpr BlockKind block_kind = BlockKind::Blocked; template @@ -79,14 +79,11 @@ struct RegularVamanaFlavor { std::span internal_ids, Distance distance, svs::threads::ThreadPoolHandle pool, - AllocatorHandle graph_handle, - const svs::data::BlockingParameters& block_params, + const graph_allocator_type& graph_allocator, svs::logging::logger_ptr logger ) { auto graph = graph_type( - src_graph.n_nodes(), - config.build_parameters.graph_max_degree, - svs::data::Blocked{block_params, std::move(graph_handle)} + src_graph.n_nodes(), config.build_parameters.graph_max_degree, graph_allocator ); svs::data::copy(src_graph.get_data(), graph.get_data()); @@ -168,6 +165,9 @@ svs::DynamicVamana copy_dynamic_vamana_index( 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), @@ -177,8 +177,7 @@ svs::DynamicVamana copy_dynamic_vamana_index( internal_ids, std::move(distance), std::move(pool), - allocator_builder.build_for_graph(), - block_params, + graph_allocator, std::move(logger) ); } @@ -202,10 +201,7 @@ void register_copy_specializations(CopyDynamicIndexDispatcher& dispatcher) { auto compression_closure = [&]() { if constexpr (std::is_same_v) { auto inner = register_pair.template operator()(); - for_simple_specializations(inner); - for_sq_specializations(inner); - for_lvq_specializations(inner); - for_leanvec_specializations(inner); + for_all_specializations(inner); } }; for_simple_specializations(compression_closure); diff --git a/bindings/c/src/dispatcher_vamana.cpp b/bindings/c/src/dispatcher_vamana.cpp index 09c6b3b8..3077ac90 100644 --- a/bindings/c/src/dispatcher_vamana.cpp +++ b/bindings/c/src/dispatcher_vamana.cpp @@ -121,18 +121,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 @@ -236,6 +227,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 @@ -253,12 +246,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 @@ -274,11 +264,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/tests/c_api_dynamic_index_sync.cpp b/bindings/c/tests/c_api_dynamic_index_sync.cpp index 43d0d908..e4c8cea8 100644 --- a/bindings/c/tests/c_api_dynamic_index_sync.cpp +++ b/bindings/c/tests/c_api_dynamic_index_sync.cpp @@ -436,8 +436,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( From 6f9592d4a0a7d691499b799ec632d79e30ab09fe Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Wed, 14 Oct 2026 05:10:43 -0700 Subject: [PATCH 4/9] Code review fixes --- bindings/c/src/data_builder.cpp | 50 ++++++++++++------- .../c/src/dispatcher_concurrent_vamana.cpp | 5 +- bindings/c/src/index_vamana.cpp | 5 +- bindings/c/src/index_vamana.hpp | 5 +- bindings/c/tests/c_api_dynamic_index_sync.cpp | 1 + 5 files changed, 42 insertions(+), 24 deletions(-) diff --git a/bindings/c/src/data_builder.cpp b/bindings/c/src/data_builder.cpp index 303777db..8eea99f9 100644 --- a/bindings/c/src/data_builder.cpp +++ b/bindings/c/src/data_builder.cpp @@ -22,25 +22,35 @@ #include #include -#include #include namespace svs::c_runtime { namespace { +template +size_t +estimate_size(DataBuilder builder, size_t num_vectors, size_t dimension, svs::lib::Empty) { + using allocator_type = typename DataBuilder::allocator_type; + static_assert( + !svs::data::is_blocked_v, + "estimate_size requires a non-blocked allocator type." + ); + return builder.estimate_size(num_vectors, dimension, allocator_type{}); +} template -size_t estimate_size( +size_t estimate_blocked_size( DataBuilder builder, size_t num_vectors, size_t dimension, - const svs::data::BlockingParameters& block_params + svs::data::BlockingParameters block_params ) { using allocator_type = typename DataBuilder::allocator_type; - if constexpr (svs::data::is_blocked_v) { - return builder.estimate_size(num_vectors, dimension, allocator_type{block_params}); - } else { - return builder.estimate_size(num_vectors, dimension, allocator_type{}); - } + static_assert( + svs::data::is_blocked_v, + "estimate_blocked_size requires a blocked allocator type." + ); + auto allocator = allocator_type{block_params}; + return builder.estimate_size(num_vectors, dimension, allocator); } template @@ -50,11 +60,18 @@ void register_data_size_specializations(Dispatcher& dispatcher) { }; for_all_specializations(size_closure); - for_all_specializations(size_closure); + + auto blocked_size_closure = [&dispatcher]() { + dispatcher.register_target(&estimate_blocked_size); + }; + + for_all_specializations(blocked_size_closure); } -using EstimateSizeDispatcher = svs::lib:: - Dispatcher; +using BlocksizeArg = std::variant; + +using EstimateSizeDispatcher = + svs::lib::Dispatcher; const EstimateSizeDispatcher& build_data_size_dispatcher() { static EstimateSizeDispatcher dispatcher = [] { @@ -66,13 +83,10 @@ const EstimateSizeDispatcher& build_data_size_dispatcher() { } size_t dispatch_data_size_estimation( - const Storage* storage, - size_t num_vectors, - size_t dimension, - const svs::data::BlockingParameters& block_params = {} + const Storage* storage, size_t num_vectors, size_t dimension, BlocksizeArg block_params ) { return build_data_size_dispatcher().invoke( - storage, num_vectors, dimension, block_params + storage, num_vectors, dimension, std::move(block_params) ); } } // namespace @@ -87,7 +101,9 @@ size_t estimate_data_size(const Storage* storage, size_t num_vectors, size_t dim if (dimension == 0) { throw std::invalid_argument("Dimension must be greater than zero."); } - return dispatch_data_size_estimation(storage, num_vectors, dimension); + return dispatch_data_size_estimation( + storage, num_vectors, dimension, svs::lib::Empty{} + ); } size_t estimate_data_size_blocked( diff --git a/bindings/c/src/dispatcher_concurrent_vamana.cpp b/bindings/c/src/dispatcher_concurrent_vamana.cpp index 7fdb0c59..dc2b4878 100644 --- a/bindings/c/src/dispatcher_concurrent_vamana.cpp +++ b/bindings/c/src/dispatcher_concurrent_vamana.cpp @@ -332,10 +332,13 @@ svs::index::vamana::MemoryBreakdown dispatch_concurrent_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; - breakdown.reverse_edges_bytes = + 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; } diff --git a/bindings/c/src/index_vamana.cpp b/bindings/c/src/index_vamana.cpp index 41d97372..672728d1 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( diff --git a/bindings/c/src/index_vamana.hpp b/bindings/c/src/index_vamana.hpp index f032e7bb..27eca874 100644 --- a/bindings/c/src/index_vamana.hpp +++ b/bindings/c/src/index_vamana.hpp @@ -26,6 +26,7 @@ #include #include +#include #include #include #include @@ -78,8 +79,8 @@ struct DynamicIndexVamana : public DynamicIndex { svs::DynamicVamana 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 = 0; - std::atomic max_id = 0; + std::atomic min_id = std::numeric_limits::max(); + std::atomic max_id = std::numeric_limits::min(); DynamicIndexVamana( const IndexBuilder& builder, svs::DynamicVamana&& index, diff --git a/bindings/c/tests/c_api_dynamic_index_sync.cpp b/bindings/c/tests/c_api_dynamic_index_sync.cpp index e4c8cea8..5b74705d 100644 --- a/bindings/c/tests/c_api_dynamic_index_sync.cpp +++ b/bindings/c/tests/c_api_dynamic_index_sync.cpp @@ -653,6 +653,7 @@ CATCH_TEST_CASE("C API Dynamic Index Fine Grain Storage", "[c_api][index][dynami } CATCH_REQUIRE(check_storage_support(storage, fx.error)); if (!storage_usable(storage)) { + svs_storage_free(storage); return; } From bfb7b7ecb3661d8cbb829d5152930904336bb503 Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Wed, 14 Oct 2026 05:41:21 -0700 Subject: [PATCH 5/9] Fix copy/ref data issue in copy_dynamic_vamana_index() --- bindings/c/src/dispatcher_dynamic_vamana_copy.hpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp index 6b21e5b6..99feb52c 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp +++ b/bindings/c/src/dispatcher_dynamic_vamana_copy.hpp @@ -82,6 +82,8 @@ struct RegularVamanaFlavor { 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 ); @@ -157,7 +159,7 @@ svs::DynamicVamana copy_dynamic_vamana_index( } } - const auto src_data = src_builder.get_dataset(src_index_impl->view_data()); + 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; From 50b3d6109be812d2885b9cdb9afbbe0f2a70ae90 Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Wed, 14 Oct 2026 06:00:59 -0700 Subject: [PATCH 6/9] Code review s1e2 --- bindings/c/src/data_builder/leanvec.hpp | 5 ++--- bindings/c/src/data_builder/lvq.hpp | 6 ++---- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/bindings/c/src/data_builder/leanvec.hpp b/bindings/c/src/data_builder/leanvec.hpp index 98623f5c..2550461b 100644 --- a/bindings/c/src/data_builder/leanvec.hpp +++ b/bindings/c/src/data_builder/leanvec.hpp @@ -15,11 +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" @@ -192,7 +192,6 @@ template void for_leanvec_specialization } // namespace svs #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled -#include "allocator.hpp" namespace svs { // Define empty stubs for LeanVec-related functions when LVQ/LeanVec support is disabled template void for_leanvec_specializations(F&&) {} diff --git a/bindings/c/src/data_builder/lvq.hpp b/bindings/c/src/data_builder/lvq.hpp index 10dc96c5..116760bf 100644 --- a/bindings/c/src/data_builder/lvq.hpp +++ b/bindings/c/src/data_builder/lvq.hpp @@ -15,11 +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" @@ -40,7 +40,6 @@ #include #endif // SVS_LVQ_HEADER -#include "allocator.hpp" #include #include @@ -182,7 +181,6 @@ template void for_lvq_specializations(F& } // namespace svs #else // SVS_RUNTIME_ENABLE_LVQ_LEANVEC not enabled -#include "allocator.hpp" namespace svs { // Define empty stubs for LVQ-related functions when LVQ/LeanVec support is disabled template void for_lvq_specializations(F&&) {} From 9bbee5f8408bdefad2b56a55b772746f95aa9312 Mon Sep 17 00:00:00 2001 From: Rafik Saliev Date: Thu, 15 Oct 2026 07:38:45 -0700 Subject: [PATCH 7/9] Force fine-grain index use QueueThreadPool instead of Native. --- bindings/c/src/index_builder.cpp | 14 +++++++------- bindings/c/src/threadpool.hpp | 9 +++++++-- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/bindings/c/src/index_builder.cpp b/bindings/c/src/index_builder.cpp index 5ed7b033..5609b87b 100644 --- a/bindings/c/src/index_builder.cpp +++ b/bindings/c/src/index_builder.cpp @@ -247,7 +247,7 @@ std::shared_ptr IndexBuilder::copy_dynamic( storage.get(), dst_concurrent, to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -258,7 +258,7 @@ std::shared_ptr IndexBuilder::copy_dynamic( src_builder.storage.get(), storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -282,7 +282,7 @@ std::shared_ptr IndexBuilder::build_dynamic( ids, storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -293,7 +293,7 @@ std::shared_ptr IndexBuilder::build_dynamic( ids, storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -317,7 +317,7 @@ std::shared_ptr IndexBuilder::load_dynamic( directory, storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -327,7 +327,7 @@ std::shared_ptr IndexBuilder::load_dynamic( directory, storage.get(), to_distance_type(distance_metric), - pool_builder.build(), + pool_builder.build(sync_kind), allocator_builder, block_params, get_logger() @@ -353,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 ), diff --git a/bindings/c/src/threadpool.hpp b/bindings/c/src/threadpool.hpp index e0d3e4e9..dcea5b27 100644 --- a/bindings/c/src/threadpool.hpp +++ b/bindings/c/src/threadpool.hpp @@ -152,11 +152,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: return ThreadPoolHandle(OMPThreadPool(num_threads)); case SVS_THREADPOOL_KIND_SINGLE_THREAD: From 24eccea6bc87ef3e7f9747ce7b93a003ee9b260c Mon Sep 17 00:00:00 2001 From: Dmitry Razdoburdin Date: Fri, 16 Oct 2026 01:07:10 -0700 Subject: [PATCH 8/9] filter deleted ids --- bindings/c/include/svs/c/svs_c.h | 10 ++++++---- bindings/c/src/index_vamana.cpp | 23 ++++++++++++++++++++++- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h index 41d09799..3578e440 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -424,9 +424,9 @@ struct svs_stream_interface { /// Row @p q (results for query @p q) occupies half-open range /// [@p offsets[q], @p offsets[q+1]) in @p indices and @p distances. This supports /// variadic per-query result counts (filtered search, range search) while keeping -/// the data flat and cache-friendly. For fixed top-k searches @p offsets[q] equals -/// @p q * k, so the classical @p indices[q*k+j] / @p distances[q*k+j] access pattern -/// remains valid. +/// the data flat and cache-friendly. When every query returns @p k results, +/// @p offsets[q] equals @p q * k, so the classical @p indices[q*k+j] / +/// @p distances[q*k+j] access pattern remains valid. /// /// Ownership: when @p owns_buffers is true the library allocated @p offsets, /// @p indices, @p distances and svs_search_results_free() will release them. @@ -1460,7 +1460,9 @@ SVS_API bool svs_index_get_size( /// @details Performs a TopK search on the index with the provided queries and an optional /// ID filter. The ID filter allows for filtering the search results based on specific IDs, /// enabling more targeted searches. If the ID filter is NULL, the search will return the -/// top K results. If ID filter is provided, only the results that pass the filter will be +/// top K results, except that SVS_SYNC_KIND_FINE_GRAIN may return fewer when concurrent +/// consolidation removes selected IDs. Use the CSR offsets to determine each query's +/// result count. If ID filter is provided, only the results that pass the filter will be /// returned. Results are written into @p out_results in CSR layout (see /// svs_search_results); library-owned buffers are reused across calls when their /// capacity suffices. If ID filter is provided with `filter_rate > 0.0` then the diff --git a/bindings/c/src/index_vamana.cpp b/bindings/c/src/index_vamana.cpp index 672728d1..64332379 100644 --- a/bindings/c/src/index_vamana.cpp +++ b/bindings/c/src/index_vamana.cpp @@ -22,6 +22,7 @@ #include "index_builder.hpp" #include +#include #include #include #include @@ -132,7 +133,27 @@ std::pair, std::vector> DynamicIndexVamana::sea if (id_filter == nullptr) { index.search(results.view(), queries, params); - return {results, std::vector(queries.size(), num_neighbors)}; + auto found_neighbors_per_query = std::vector(queries.size(), num_neighbors); + if (sync_kind == SVS_SYNC_KIND_FINE_GRAIN) { + // Consolidation can erase buffered IDs before translation. The CSR exporter + // consumes each row's counted prefix, so move IDs and distances together. + for (size_t i = 0; i < results.n_queries(); ++i) { + size_t found = 0; + for (size_t j = 0; j < results.n_neighbors(); ++j) { + const auto id = results.index(i, j); + if (id == svs::index::vamana::concurrent::no_external_id) { + continue; + } + if (found != j) { + results.index(i, found) = id; + results.distance(i, found) = results.distance(i, j); + } + ++found; + } + found_neighbors_per_query[i] = found; + } + } + return {std::move(results), std::move(found_neighbors_per_query)}; } std::mt19937 rng(42); From 4eb80386f89cbbb4c8e23c906a8983c05eed820f Mon Sep 17 00:00:00 2001 From: Dmitry Razdoburdin Date: Fri, 16 Oct 2026 01:12:57 -0700 Subject: [PATCH 9/9] clean-up --- bindings/c/include/svs/c/svs_c.h | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h index 3578e440..39e23426 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -424,9 +424,9 @@ struct svs_stream_interface { /// Row @p q (results for query @p q) occupies half-open range /// [@p offsets[q], @p offsets[q+1]) in @p indices and @p distances. This supports /// variadic per-query result counts (filtered search, range search) while keeping -/// the data flat and cache-friendly. When every query returns @p k results, -/// @p offsets[q] equals @p q * k, so the classical @p indices[q*k+j] / -/// @p distances[q*k+j] access pattern remains valid. +/// the data flat and cache-friendly. For fixed top-k searches @p offsets[q] equals +/// @p q * k, so the classical @p indices[q*k+j] / @p distances[q*k+j] access pattern +/// remains valid. /// /// Ownership: when @p owns_buffers is true the library allocated @p offsets, /// @p indices, @p distances and svs_search_results_free() will release them.