diff --git a/bindings/c/docs/C_API_Design.md b/bindings/c/docs/C_API_Design.md index aa4fdb276..a45af6a93 100644 --- a/bindings/c/docs/C_API_Design.md +++ b/bindings/c/docs/C_API_Design.md @@ -258,7 +258,8 @@ The main search structure providing vector similarity search operations. **Dynamic Index Parameters:** The `_ex` variants of the dynamic functions (`svs_index_build_dynamic_ex`, -`svs_index_load_dynamic_ex`, `svs_index_convert_dynamic_ex`, +`svs_index_load_dynamic_ex`, `svs_index_load_stream_dynamic_ex`, +`svs_index_convert_dynamic_ex`, `svs_index_builder_estimate_memory_dynamic_ex`, `svs_index_builder_estimate_search_memory_dynamic_ex`) take a versioned `const svs_dynamic_index_params_t*` instead of a bare `blocksize_bytes` argument. @@ -277,7 +278,7 @@ 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` | 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 | +| `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 takes the shared lock too: it runs alongside searches, waits for running writers, and does not consolidate or compact. `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. @@ -654,7 +655,7 @@ for full signatures, parameters, and Doxygen documentation. | **Search params** | `svs_search_params_create_vamana`, `svs_search_params_free` | | **Builder** | `svs_index_builder_create`, `svs_index_builder_set_{storage,threadpool,threadpool_custom,allocator,allocator_custom}`, `svs_index_builder_free` | | **Memory estimation** | `svs_index_builder_estimate_memory`, `svs_index_builder_estimate_memory_dynamic[_ex]`, `svs_index_builder_estimate_search_memory`, `svs_index_builder_estimate_search_memory_dynamic[_ex]`, `svs_index_builder_get_default_blocksize_bytes` | -| **Index lifecycle** | `svs_index_build`, `svs_index_build_dynamic[_ex]`, `svs_index_convert`, `svs_index_convert_dynamic[_ex]`, `svs_index_load`, `svs_index_load_dynamic[_ex]`, `svs_index_load_stream`, `svs_index_load_stream_dynamic`, `svs_index_save`, `svs_index_save_stream`, `svs_index_free` | +| **Index lifecycle** | `svs_index_build`, `svs_index_build_dynamic[_ex]`, `svs_index_convert`, `svs_index_convert_dynamic[_ex]`, `svs_index_load`, `svs_index_load_dynamic[_ex]`, `svs_index_load_stream`, `svs_index_load_stream_dynamic[_ex]`, `svs_index_save`, `svs_index_save_stream`, `svs_index_free` | | **Dynamic ops** | `svs_index_dynamic_{add_points,delete_points,has_id,consolidate,compact,get_sync_kind}` | | **Introspection** | `svs_index_get_num_threads` / `set_num_threads`, `svs_index_get_distance`, `svs_index_reconstruct`, `svs_index_get_memory_usage`, `svs_index_get_memory_breakdown` | | **Search** | `svs_index_search_topk` (+ deprecated `svs_index_search`), `svs_search_results_free` | diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h index 0826c16f9..a00423a9d 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -597,7 +597,9 @@ struct svs_memory_breakdown { /// 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 +/// in-flight operations inside the index). save takes a shared lock and runs alongside +/// searches; the index blocks its own writers while saving and keeps deleted and empty +/// slots as they are (no consolidate or compact). Only set_num_threads, memory queries /// and use as a conversion source take an exclusive lock. /// /// @remarks Synchronization covers operations on the index only: @@ -1411,6 +1413,27 @@ SVS_API svs_index_h svs_index_load_stream_dynamic( svs_error_h out_err /*=NULL*/ ); +/// @brief Load a dynamic index from a caller-supplied stream with explicit parameters +/// @param builder The index builder handle (used for configuration) +/// @param stream The stream interface to read the index from +/// @param params Pointer to the dynamic index parameters structure, or NULL for defaults +/// @param out_err An optional error handle to capture errors +/// @return A handle to the loaded dynamic index +/// @remarks The block parameters only affect the in-memory layout, not the on-disk +/// format. The sync kind is not persisted with the index and must be passed on every load. +/// @remarks The operations table is copied, but @p stream->self is retained as-is. It only +/// needs to remain valid until this function returns. +/// @remarks Accepts both the native stream encoding and a packed directory archive. The +/// native encoding carries no total length, so a load may read up to 64 KiB past the end +/// of the index; frame the payload if embedding it in a larger stream. +/// @error SVS_ERROR_INVALID_ARGUMENT if @p params is invalid (see svs_dynamic_index_params) +SVS_API svs_index_h svs_index_load_stream_dynamic_ex( + svs_index_builder_h builder, + svs_stream_i stream, + const svs_dynamic_index_params_t* params /*=NULL*/, + svs_error_h out_err /*=NULL*/ +); + /// @brief Convert an index using new builder configuration /// @param builder The index builder handle (used for configuration) /// @param src_index The source index handle to convert from diff --git a/bindings/c/src/dispatcher_concurrent_vamana.cpp b/bindings/c/src/dispatcher_concurrent_vamana.cpp index af6daa95c..36e67a362 100644 --- a/bindings/c/src/dispatcher_concurrent_vamana.cpp +++ b/bindings/c/src/dispatcher_concurrent_vamana.cpp @@ -32,6 +32,8 @@ #include #include +#include +#include #include #include #include @@ -196,9 +198,38 @@ svs::DynamicVamana load_concurrent_vamana_index( ); } +template +svs::DynamicVamana load_stream_concurrent_vamana_index( + const svs::index::vamana::VamanaBuildParameters& SVS_UNUSED(build_params), + std::unique_ptr stream, + DataLoader SVS_UNUSED(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; + using data_type = typename DataLoader::data_type; + auto allocator = allocator_type{block_params, allocator_builder.build()}; + auto graph_allocator = ConcurrentGraphAllocator{ + block_params, allocator_builder.build_for_graph()}; + + return svs::ConcurrentDynamicVamana::assemble( + *stream, + svs::make_stream_loader(graph_allocator), + svs::make_stream_loader(allocator), + std::move(D), + std::move(pool), + std::move(logger) + ); +} + using ConcurrentVamanaSource = std::variant< std::pair, std::span>, - std::filesystem::path>; + std::filesystem::path, + std::unique_ptr>; using BuildConcurrentIndexDispatcher = svs::lib::Dispatcher< svs::DynamicVamana, @@ -221,10 +252,15 @@ const BuildConcurrentIndexDispatcher& build_concurrent_vamana_index_dispatcher() d.register_target(&load_concurrent_vamana_index); }; + auto load_stream_closure = [&d]() { + d.register_target(&load_stream_concurrent_vamana_index); + }; + constexpr auto kind = ConcurrentVamanaFlavor::block_kind; for_all_specializations(build_closure); for_all_specializations(load_closure); + for_all_specializations(load_stream_closure); return d; }(); return dispatcher; @@ -293,6 +329,28 @@ svs::DynamicVamana dispatch_concurrent_vamana_index_load( ); } +svs::DynamicVamana dispatch_concurrent_vamana_index_load_stream( + const svs::index::vamana::VamanaBuildParameters& build_params, + std::unique_ptr stream, + 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::move(stream)}, + 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, diff --git a/bindings/c/src/dispatcher_concurrent_vamana.hpp b/bindings/c/src/dispatcher_concurrent_vamana.hpp index 0710e88bb..59f366dba 100644 --- a/bindings/c/src/dispatcher_concurrent_vamana.hpp +++ b/bindings/c/src/dispatcher_concurrent_vamana.hpp @@ -53,6 +53,17 @@ svs::DynamicVamana dispatch_concurrent_vamana_index_load( svs::logging::logger_ptr logger ); +svs::DynamicVamana dispatch_concurrent_vamana_index_load_stream( + const svs::index::vamana::VamanaBuildParameters& build_params, + std::unique_ptr stream, + 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( diff --git a/bindings/c/src/index_builder.cpp b/bindings/c/src/index_builder.cpp index 89ae15ba0..ee971365e 100644 --- a/bindings/c/src/index_builder.cpp +++ b/bindings/c/src/index_builder.cpp @@ -346,23 +346,29 @@ std::shared_ptr IndexBuilder::load_stream_dynamic( ) { if (algorithm->type == SVS_ALGORITHM_TYPE_VAMANA) { auto vamana_algorithm = static_cast(algorithm.get()); - - auto index = std::make_shared( - *this, - dispatch_dynamic_vamana_index_load_stream( - vamana_algorithm->build_parameters(), - std::move(stream), - storage.get(), - to_distance_type(distance_metric), - pool_builder.build(sync_kind), - 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_stream( + build_params, + std::move(stream), + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ) + : dispatch_dynamic_vamana_index_load_stream( + build_params, + std::move(stream), + storage.get(), + to_distance_type(distance_metric), + pool_builder.build(sync_kind), + allocator_builder, + block_params, + get_logger() + ); + return make_dynamic_index(*this, std::move(index), sync_kind); } return nullptr; } diff --git a/bindings/c/src/index_vamana.hpp b/bindings/c/src/index_vamana.hpp index 2a193981e..7aa643d27 100644 --- a/bindings/c/src/index_vamana.hpp +++ b/bindings/c/src/index_vamana.hpp @@ -158,8 +158,9 @@ struct DynamicIndexVamana : public DynamicIndex { /// 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) +/// lock alongside searches. save() also takes the shared lock: the index excludes its own +/// writers while saving. Operations that are not thread-safe in the index +/// (set_num_threads) and whole-index reads (memory breakdown, use as a conversion source) /// take the exclusive lock. struct ConcurrentIndexVamana : public DynamicIndexVamana { ConcurrentIndexVamana( @@ -179,6 +180,16 @@ struct ConcurrentIndexVamana : public DynamicIndexVamana { size_t delete_points(std::span ids) override; + void save(const std::filesystem::path& directory) override { + auto lock = read_lock(); + index.save(directory / "config", directory / "graph", directory / "data"); + } + + void save(std::ostream& stream) override { + auto lock = read_lock(); + index.save(stream); + } + void consolidate() override { auto lock = read_lock(); index.consolidate(); diff --git a/bindings/c/src/svs_c.cpp b/bindings/c/src/svs_c.cpp index 6718861bf..5a549b3c4 100644 --- a/bindings/c/src/svs_c.cpp +++ b/bindings/c/src/svs_c.cpp @@ -1245,6 +1245,40 @@ extern "C" svs_index_h svs_index_load_stream_dynamic( ); } +extern "C" svs_index_h svs_index_load_stream_dynamic_ex( + svs_index_builder_h builder, + svs_stream_i stream, + const svs_dynamic_index_params_t* params, + svs_error_h out_err +) { + using namespace svs::c_runtime; + return wrap_exceptions( + [&]() { + EXPECT_ARG_NOT_NULL(builder); + EXPECT_ARG_NOT_NULL(stream); + NOT_IMPLEMENTED_IF( + (builder->impl->algorithm->type != SVS_ALGORITHM_TYPE_VAMANA), + "Only Vamana algorithm is currently supported for dynamic index loading" + ); + StreamBuf::validate(stream, /*need_write=*/false); + auto dynamic_params = read_dynamic_index_params(params); + auto index = builder->impl->load_stream_dynamic( + std::make_unique(*stream->ops, stream->self), + dynamic_params.block_params, + dynamic_params.sync_kind + ); + if (index == nullptr) { + SET_ERROR(out_err, SVS_ERROR_RUNTIME, "Dynamic index load failed"); + return svs_index_h{nullptr}; + } + auto result = new svs_index; + result->impl = index; + return result; + }, + out_err + ); +} + extern "C" svs_index_h svs_index_convert(svs_index_builder_h builder, svs_index_h src_index, svs_error_h out_err) { using namespace svs::c_runtime; diff --git a/data/serialization.toml b/data/serialization.toml index 24554ef08..921ac6d5a 100644 --- a/data/serialization.toml +++ b/data/serialization.toml @@ -98,6 +98,14 @@ schema_version = "v0.0.0" parameters = "table" translation = "table" +[[schemas]] + __schema__ = "vamana_dynamic_auxiliary_parameters" + __version__ = "v0.0.1" + name = "str" + parameters = "table" + translation = "table" + status = "table" + [[schemas]] __schema__ = "vamana_build_parameters" __version__ = "v0.0.1" diff --git a/include/svs/concurrent/README.md b/include/svs/concurrent/README.md index a8ba73bf0..8a946dc43 100644 --- a/include/svs/concurrent/README.md +++ b/include/svs/concurrent/README.md @@ -84,12 +84,18 @@ storage smaller, so it must wait for every reader to finish first. It is the onl caller that takes `compact_mutex_` exclusive. It runs `consolidate_locked()` before it compacts. -### 1d. Not safe at all +### 1d. `save()` + +`save()` writes the slots as they are, with deleted and empty slots included. It does +not call `consolidate()` or `compact()`. It takes `writers_gatekeeper_` and then +`writers_mutex_` exclusive, so it waits for running writers and blocks new ones. New +writers hold the gatekeeper only while acquiring `writers_mutex_`, so they cannot +starve `save()`. Searches keep running. + +### 1e. Not safe at all Call these only when no other thread uses the index: -- `save()` - it holds no lock while it writes the files. It calls `consolidate()` - and `compact()` first, but it releases both locks before writing. - `set_alpha`, `set_prune_to`, `set_max_candidates`, `set_construction_window_size`, `set_threadpool` - plain writes, no lock. @@ -142,6 +148,8 @@ There are three groups: | Lock | Protects | Taken exclusive by | | --- | --- | --- | +| `writers_mutex_` | the index state stays unchanged while `save()` writes it | `save()` only | +| `writers_gatekeeper_` | new writers wait behind a pending `save()` | `save()` only | | `compact_mutex_` | the storage stays alive | `compact()` only | | `translator_mutex_` | the two id hash maps | `add_points`, `consolidate`, `compact`, `replace_external_id` | | `pending_insertions_mutex_` | `pending_insertions_` | `add_points` | @@ -159,6 +167,7 @@ so it waits for all of them. deadlock. ``` +writers_gatekeeper_ -> writers_mutex_ -> compact_gatekeeper_ -> compact_mutex_ compact_mutex_ -> slot_alloc_mutex_ compact_mutex_ -> translator_mutex_ l2e_mutex_ -> e2l_mutex_ (multi index) @@ -247,8 +256,8 @@ The dataset must use a `SegmentedBlocked` allocator; this is checked at compile time. `distance` may be a functor or a `svs::DistanceType`. `batch_iterator()` returns this index's own `BatchIterator`, so it takes the locks described in 1a. -The rules from section 1 apply unchanged: `save()`, `set_threadpool()` and the -build-parameter setters still need exclusive access. +The rules from section 1 apply unchanged: `set_threadpool()` and the +build-parameter setters still need exclusive access; `save()` follows 1d. ```cpp using Data = svs::index::vamana::concurrent::SegmentedBlockedData; diff --git a/include/svs/concurrent/dynamic_index.h b/include/svs/concurrent/dynamic_index.h index 1fc1a8980..bb88b03ab 100644 --- a/include/svs/concurrent/dynamic_index.h +++ b/include/svs/concurrent/dynamic_index.h @@ -250,6 +250,8 @@ class MutableVamanaIndex { // compact() holds it exclusive. // // Lock acquisition order across the codebase: + // writers_gatekeeper_ -> writers_mutex_ (never reversed) + // writers_mutex_ -> compact_gatekeeper_ (never reversed) // compact_gatekeeper_ -> compact_mutex_ (never reversed) // compact_mutex_ -> slot_alloc_mutex_ (never reversed) // compact_mutex_ -> translator_mutex_ (never reversed) @@ -263,6 +265,14 @@ class MutableVamanaIndex { std::make_unique()}; // Iterators check this under compact_mutex_ before reusing cached internal IDs. size_t compaction_epoch_ = 0; + // Outermost lock. Every writer (add_points, delete_entries, consolidate, compact, + // replace_external_id) holds it shared; save() holds it exclusive so the state it + // writes is stable. Searches and other readers never take it. + std::unique_ptr writers_mutex_{ + std::make_unique()}; + // Once save() owns this gate, new writers cannot keep it waiting on writers_mutex_. + std::unique_ptr writers_gatekeeper_{ + std::make_unique()}; // Writer-only mutex serializing slot allocation in add_points std::unique_ptr slot_alloc_mutex_{std::make_unique()}; @@ -447,7 +457,7 @@ class MutableVamanaIndex { /// /// * data.size() == graph.n_nodes() == status.size(). /// * ``status`` contains no ``Pending`` slots. - /// * ``translator`` maps exactly the ``Valid`` slots of ``status``. + /// * ``translator`` maps every ``Valid`` slot and no ``Empty`` slot of ``status``. template MutableVamanaIndex( const VamanaIndexParameters& config, @@ -513,17 +523,12 @@ class MutableVamanaIndex { status_[i] = s; } - if (translator_.size() != num_valid) { - throw ANNEXCEPTION( - "Translator has {} IDs but status has {} valid slots", - translator_.size(), - num_valid - ); - } for (auto pair : translator_) { - if (pair.second >= n || status[pair.second] != SlotMetadata::Valid) { + if (pair.second >= n || (status[pair.second] != SlotMetadata::Valid && + status[pair.second] != SlotMetadata::Deleted)) { throw ANNEXCEPTION( - "Translator maps external id {} to non-valid slot {}", + "Translator maps external id {} to slot {} that is not valid or " + "deleted", pair.first, pair.second ); @@ -826,6 +831,7 @@ class MutableVamanaIndex { /// @param new_id The external ID to assign. Must not be mapped at all. /// ReplaceExternalIdResult replace_external_id(size_t old_id, size_t new_id) { + auto writers_lock = lock_writers_shared(); auto compact_lock = lock_storage_shared(); std::lock_guard lock{*translator_mutex_}; @@ -882,6 +888,12 @@ class MutableVamanaIndex { // This is an internal method, mostly used to help implement the batch iterator. ValidBuilder internal_search_builder() const { return ValidBuilder{status_}; } + /// @brief Shared lock taken by every writer; waits behind a pending save(). + [[nodiscard]] std::shared_lock lock_writers_shared() const { + std::shared_lock gatekeeper_lock{*writers_gatekeeper_}; + return std::shared_lock(*writers_mutex_); + } + /// @brief RAII shared lock guarding data_/graph_ against compact()'s shrink /// (which frees segments). Used by searches, iterators and mutations that /// may overlap each other but must exclude compaction. @@ -1108,6 +1120,8 @@ class MutableVamanaIndex { return {}; } + auto writers_lock = lock_writers_shared(); + // Reserve slot ownership against compact(). Held for the entire call, // including the lock-free Phase 2-4 below; compact() takes this // exclusive and so will block until every in-flight add finishes. @@ -1315,6 +1329,7 @@ class MutableVamanaIndex { /// graph. /// template size_t delete_entries(const T& ids) { + auto writers_lock = lock_writers_shared(); auto compact_lock = lock_storage_shared(); std::shared_lock lock{*translator_mutex_}; size_t deleted = 0; @@ -1400,6 +1415,7 @@ class MutableVamanaIndex { /// improve performance but requires more working memory. /// void compact(Idx batch_size = 1'000) { + auto writers_lock = lock_writers_shared(); std::lock_guard gatekeeper_lock{*compact_gatekeeper_}; std::lock_guard compact_lock{*compact_mutex_}; @@ -1613,6 +1629,7 @@ class MutableVamanaIndex { ///// Mutation void consolidate() { + auto writers_lock = lock_writers_shared(); auto compact_lock = lock_storage_shared(); consolidate_locked(); } @@ -1702,6 +1719,7 @@ class MutableVamanaIndex { /// @returns The number of listed IDs that were consolidated. /// template size_t consolidate(const T& ids) { + auto writers_lock = lock_writers_shared(); auto compact_lock = lock_storage_shared(); // Collect the internal slots of the listed, already-soft-deleted IDs. @@ -1741,18 +1759,29 @@ class MutableVamanaIndex { get_search_parameters()}; } - static constexpr lib::Version save_version = lib::Version(0, 0, 0); + static constexpr lib::Version save_version = vamana::detail::dynamic_status_version; + + // Saves the slots as they are (holes included); does not consolidate or compact. + // Blocks writers while saving; searches may continue. void save( const std::filesystem::path& config_directory, const std::filesystem::path& graph_directory, const std::filesystem::path& data_directory ) { - vamana::detail::save_dynamic_index( - *this, config_directory, graph_directory, data_directory + std::lock_guard gatekeeper_lock{*writers_gatekeeper_}; + std::unique_lock writers_lock{*writers_mutex_}; + auto status = status_bytes_for_save(); + vamana::detail::write_dynamic_index( + *this, translator_, &status, config_directory, graph_directory, data_directory ); } - void save(std::ostream& os) { vamana::detail::save_dynamic_index(*this, os); } + void save(std::ostream& os) { + std::lock_guard gatekeeper_lock{*writers_gatekeeper_}; + std::unique_lock writers_lock{*writers_mutex_}; + auto status = status_bytes_for_save(); + vamana::detail::write_dynamic_index(*this, translator_, &status, os); + } ///// ///// Calibrate @@ -1873,6 +1902,20 @@ class MutableVamanaIndex { // Not synchronized: callers must exclude concurrent mutation. const IDTranslator& view_translator() const { return translator_; } + // Slot status bytes. The caller holds writers_mutex_ exclusive. + std::vector status_bytes_for_save() const { + auto status = std::vector{}; + auto slots = status_snapshot(); + status.reserve(slots.size()); + for (size_t i = 0, imax = slots.size(); i < imax; ++i) { + if (slots[i] == SlotMetadata::Pending) { + throw ANNEXCEPTION("Cannot save an index with pending slot {}", i); + } + status.push_back(static_cast(slots[i])); + } + return status; + } + // Not synchronized: callers must exclude concurrent mutation. std::vector status_snapshot() const { auto result = std::vector(status_.size()); @@ -2320,18 +2363,9 @@ auto auto_dynamic_assemble( bool SVS_UNUSED(debug_load_from_static) = false, svs::logging::logger_ptr logger = svs::logging::get() ) { - // Read the combined TOML (parameters + translation) - // and the translator binary data. - auto table = lib::detail::read_metadata(is); - - auto parameters = lib::load( - table.template cast().at("parameters").template cast() - ); - - auto translation = - table.template cast().at("translation").template cast(); - - auto translator = IDTranslator::load(translation, is); + // Read the combined TOML (parameters + translation + status) and the binary data. + auto [parameters, translator, status] = + vamana::detail::load_stream_state(is); auto data = data_loader(); auto graph = graph_loader(); @@ -2344,11 +2378,8 @@ auto auto_dynamic_assemble( ); } - auto translator_size = translator.size(); - if (translator_size != datasize) { - throw ANNEXCEPTION( - "Translator has {} IDs but should have {}", translator_size, datasize - ); + if (!status) { + status.emplace(datasize, SlotMetadata::Valid); } auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); @@ -2357,7 +2388,7 @@ auto auto_dynamic_assemble( std::move(data), std::move(graph), std::move(distance), - std::vector(datasize, SlotMetadata::Valid), + *status, std::move(translator), std::move(threadpool), std::move(logger)}; diff --git a/include/svs/index/vamana/dynamic_index.h b/include/svs/index/vamana/dynamic_index.h index 8a9a64dd1..af26c4342 100644 --- a/include/svs/index/vamana/dynamic_index.h +++ b/include/svs/index/vamana/dynamic_index.h @@ -19,6 +19,7 @@ // stdlib #include #include +#include #include // Include the flat index to spin-up exhaustive searches on demand. @@ -111,31 +112,37 @@ class ValidBuilder { namespace detail { +// Version of "vamana_dynamic_auxiliary_parameters" that also stores per-slot status. +inline constexpr lib::Version dynamic_status_version = lib::Version(0, 0, 1); + +// Writes the index as-is. If `status` is given (one byte per slot), it is saved too and +// `translator` may also map `Deleted` slots. template -void save_dynamic_index( +void write_dynamic_index( Index& index, + const IDTranslator& translator, + const std::vector* status, const std::filesystem::path& config_directory, const std::filesystem::path& graph_directory, const std::filesystem::path& data_directory ) { - // Post-consolidation, all entries should be "valid". - // Therefore, we don't need to save the slot metadata. - index.consolidate(); - index.compact(); - // Save auxiliary data structures. lib::save_to_disk( lib::SaveOverride([&](const lib::SaveContext& ctx) { // Save the construction parameters. - return lib::SaveTable( + auto table = lib::SaveTable( "vamana_dynamic_auxiliary_parameters", Index::save_version, { {"name", lib::save(index.name())}, {"parameters", lib::save(index.parameters(), ctx)}, - {"translation", lib::save(index.view_translator(), ctx)}, + {"translation", lib::save(translator, ctx)}, } ); + if (status != nullptr) { + table.insert("status", lib::save(lib::BinaryBlobSaver(*status), ctx)); + } + return table; }), config_directory ); @@ -146,12 +153,13 @@ void save_dynamic_index( lib::save_to_disk(index.view_graph(), graph_directory); } -template void save_dynamic_index(Index& index, std::ostream& os) { - // Post-consolidation, all entries should be "valid". - // Therefore, we don't need to save the slot metadata. - index.consolidate(); - index.compact(); - +template +void write_dynamic_index( + Index& index, + const IDTranslator& translator, + const std::vector* status, + std::ostream& os +) { lib::begin_serialization(os); auto save_table = lib::SaveTable( "vamana_dynamic_auxiliary_parameters", @@ -159,11 +167,17 @@ template void save_dynamic_index(Index& index, std::ostream& os { {"name", lib::save(index.name())}, {"parameters", lib::save(index.parameters())}, - {"translation", lib::detail::exit_hook(index.view_translator().metadata())}, + {"translation", lib::detail::exit_hook(translator.metadata())}, } ); + if (status != nullptr) { + save_table.insert("num_status", lib::save(status->size())); + } lib::save_to_stream(save_table, os); - index.view_translator().save(os); + translator.save(os); + if (status != nullptr) { + lib::write_binary(os, *status); + } // Save the dataset. lib::save_to_stream(index.view_data(), os); @@ -171,6 +185,35 @@ template void save_dynamic_index(Index& index, std::ostream& os lib::save_to_stream(index.view_graph(), os); } +template +void save_dynamic_index( + Index& index, + const std::filesystem::path& config_directory, + const std::filesystem::path& graph_directory, + const std::filesystem::path& data_directory +) { + // Post-consolidation, all entries should be "valid". + // Therefore, we don't need to save the slot metadata. + index.consolidate(); + index.compact(); + write_dynamic_index( + index, + index.view_translator(), + nullptr, + config_directory, + graph_directory, + data_directory + ); +} + +template void save_dynamic_index(Index& index, std::ostream& os) { + // Post-consolidation, all entries should be "valid". + // Therefore, we don't need to save the slot metadata. + index.consolidate(); + index.compact(); + write_dynamic_index(index, index.view_translator(), nullptr, os); +} + template < typename Index, data::ImmutableMemoryDataset Queries, @@ -1502,6 +1545,61 @@ MutableVamanaIndex( ) -> MutableVamanaIndex, Data, Dist>; namespace detail { +// Convert raw status bytes read from a version 0.0.1 file. Only 0, 1 and 2 +// (`Empty`, `Valid`, `Deleted`) are valid on disk. +template +std::vector decode_slot_status(const std::vector& raw) { + auto result = std::vector(raw.size()); + for (size_t i = 0, imax = raw.size(); i < imax; ++i) { + if (raw[i] > 2) { + throw ANNEXCEPTION("Invalid slot status {} at slot {}", unsigned(raw[i]), i); + } + result[i] = static_cast(raw[i]); + } + return result; +} + +// Saved concurrent indexes keep soft-deleted IDs mapped; this index erases them on delete. +template +void drop_deleted_translations( + IDTranslator& translator, const std::vector& status +) { + auto to_drop = std::vector(); + for (auto pair : translator) { + if (pair.second < status.size() && status[pair.second] == Metadata::Deleted) { + to_drop.push_back(pair.second); + } + } + translator.delete_internal(to_drop, false); +} + +// The translator must cover exactly the `Valid` slots. +template +void check_translator_matches_status( + const IDTranslator& translator, const std::vector& status +) { + size_t num_valid = 0; + for (size_t i = 0, imax = status.size(); i < imax; ++i) { + if (status[i] == Metadata::Valid) { + ++num_valid; + if (!translator.has_internal(i)) { + throw ANNEXCEPTION("Translator is missing internal id {}", i); + } + } + } + + auto translator_size = translator.size(); + if (translator_size != num_valid) { + throw ANNEXCEPTION( + "Translator has {} IDs but should have {}", translator_size, num_valid + ); + } +} + +inline bool is_supported_dynamic_version(const lib::Version& version) { + return version == lib::Version(0, 0, 0) || version == dynamic_status_version; +} + template struct VamanaStateLoader { ///// Loading static bool @@ -1509,7 +1607,7 @@ template struct VamanaStateLoader { // We provide the option to load from a static index. return VamanaIndexParameters::check_load_compatibility(schema, version) || (schema == "vamana_dynamic_auxiliary_parameters" && - version == lib::Version(0, 0, 0)); + is_supported_dynamic_version(version)); } // Provide a compatibility path for loading static datasets. @@ -1522,10 +1620,22 @@ template struct VamanaStateLoader { std::vector(assume_datasize, Metadata::Valid)}; } + auto status = std::vector(assume_datasize, Metadata::Valid); + if (table.version() == dynamic_status_version) { + std::vector raw = + lib::load_at>(table, "status"); + if (raw.size() != assume_datasize) { + throw ANNEXCEPTION( + "Status has {} entries but should have {}", raw.size(), assume_datasize + ); + } + status = decode_slot_status(raw); + } + return VamanaStateLoader{ SVS_LOAD_MEMBER_AT_(table, parameters), svs::lib::load_at(table, "translation"), - std::vector(assume_datasize, Metadata::Valid), + std::move(status), }; } @@ -1535,6 +1645,36 @@ template struct VamanaStateLoader { std::vector status_; }; +// Contents of a stream before the dataset: parameters, translator and (version 0.0.1) +// slot status. `status` is empty for version 0.0.0, where every slot is `Valid`. +template struct StreamState { + VamanaIndexParameters parameters; + IDTranslator translator; + std::optional> status; +}; + +template StreamState load_stream_state(std::istream& is) { + auto table = lib::detail::read_metadata(is); + auto root = table.template cast(); + if (!is_supported_dynamic_version(root.version())) { + throw ANNEXCEPTION("Unsupported dynamic index stream version {}", root.version()); + } + + auto parameters = + lib::load(root.at("parameters").template cast() + ); + auto translator = + IDTranslator::load(root.at("translation").template cast(), is); + + std::optional> status{}; + if (root.version() == dynamic_status_version) { + auto raw = std::vector(lib::load_at(root, "num_status")); + lib::read_binary(is, raw); + status = decode_slot_status(raw); + } + return {std::move(parameters), std::move(translator), std::move(status)}; +} + template VamanaStateLoader auto_load_state( ConfigProto&& config_proto, @@ -1661,24 +1801,9 @@ auto auto_dynamic_assemble( ); } - // The translator must cover exactly the valid slots; copied state may contain - // deleted or empty slots. - size_t num_valid = 0; - for (size_t i = 0; i < datasize; ++i) { - if (status[i] == SlotMetadata::Valid) { - ++num_valid; - if (!translator.has_internal(i)) { - throw ANNEXCEPTION("Translator is missing internal id {}", i); - } - } - } - - auto translator_size = translator.size(); - if (translator_size != num_valid) { - throw ANNEXCEPTION( - "Translator has {} IDs but should have {}", translator_size, num_valid - ); - } + // Copied state may contain deleted or empty slots. + detail::drop_deleted_translations(translator, status); + detail::check_translator_matches_status(translator, status); // At this point, we should be completely validated. // Construct the index! @@ -1707,18 +1832,9 @@ auto auto_dynamic_assemble( bool SVS_UNUSED(debug_load_from_static) = false, svs::logging::logger_ptr logger = svs::logging::get() ) { - // Read the combined TOML (parameters + translation) - // and the translator binary data. - auto table = lib::detail::read_metadata(is); - - auto parameters = lib::load( - table.template cast().at("parameters").template cast() - ); - - auto translation = - table.template cast().at("translation").template cast(); - - auto translator = IDTranslator::load(translation, is); + // Read the combined TOML (parameters + translation + status) and the binary data. + auto [parameters, translator, loaded_status] = + detail::load_stream_state(is); auto data = data_loader(); auto graph = graph_loader(); @@ -1731,14 +1847,16 @@ auto auto_dynamic_assemble( ); } - auto translator_size = translator.size(); - if (translator_size != datasize) { + auto status = + loaded_status.value_or(std::vector(datasize, SlotMetadata::Valid)); + if (status.size() != datasize) { throw ANNEXCEPTION( - "Translator has {} IDs but should have {}", translator_size, datasize + "Status has {} entries but should have {}", status.size(), datasize ); } + detail::drop_deleted_translations(translator, status); + detail::check_translator_matches_status(translator, status); - auto status = std::vector(datasize, SlotMetadata::Valid); auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); return MutableVamanaIndex{ parameters, diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index 884e67804..e729b9fcd 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -38,8 +38,10 @@ namespace svs { /// The wrapped index may be searched and mutated from several threads at once: search, /// batch iteration, ``add_points``, ``delete_points``, ``consolidate``, ``compact``, ID /// inspection, ``get_distance`` and ``reconstruct_at`` are safe to call concurrently. -/// ``save``, ``set_threadpool`` and the build-parameter setters are not; call them only -/// when no other thread uses the index. See ``include/svs/concurrent/README.md``. +/// ``save`` writes the slots as they are, without consolidating or compacting; it blocks +/// writers while it runs. ``set_threadpool`` and the build-parameter setters are not safe; +/// call them only when no other thread uses the index. See +/// ``include/svs/concurrent/README.md``. /// /// The dataset must use a ``svs::index::vamana::concurrent::SegmentedBlocked`` allocator so /// that growing it never relocates storage under concurrent readers.