diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h index 0826c16f9..2cde57d8a 100644 --- a/bindings/c/include/svs/c/svs_c.h +++ b/bindings/c/include/svs/c/svs_c.h @@ -1598,6 +1598,37 @@ SVS_API bool svs_index_dynamic_delete_points( svs_error_h out_err /*=NULL*/ ); +/// @brief Update the vectors of existing points in a dynamic index +/// @param index The dynamic index handle +/// @param points Pointer to the new vector data (float array) +/// @param ids Pointer to the IDs of the points to update (size_t array); every ID must +/// already exist in the index +/// @param num_points The number of points to update +/// @param out_err An optional error handle to capture errors +/// @return true on success, false on failure +/// @error SVS_ERROR_NOT_IMPLEMENTED if @p index uses SVS_SYNC_KIND_NONE or +/// SVS_SYNC_KIND_GLOBAL, or if @p index uses SVS_SYNC_KIND_FINE_GRAIN and @p num_points +/// is greater than 1 +SVS_API bool svs_index_dynamic_update_points( + svs_index_h index, + const float* points, + const size_t* ids, + size_t num_points, + svs_error_h out_err /*=NULL*/ +); + +/// @brief Rename the ID of a point in a dynamic index, keeping its vector data +/// @param index The dynamic index handle +/// @param old_id The existing vector ID to rename +/// @param new_id The new vector ID; must not already exist in the index +/// @param out_err An optional error handle to capture errors +/// @return true on success, false on failure +/// @error SVS_ERROR_INVALID_ARGUMENT if @p old_id does not exist in the index or +/// @p new_id already exists +SVS_API bool svs_index_dynamic_replace_id( + svs_index_h index, size_t old_id, size_t new_id, svs_error_h out_err /*=NULL*/ +); + /// @brief Check if a dynamic index has a specific ID /// @param index The dynamic index handle /// @param id The vector ID to check for diff --git a/bindings/c/src/filtered_search.hpp b/bindings/c/src/filtered_search.hpp index 8e45ac691..932558c0a 100644 --- a/bindings/c/src/filtered_search.hpp +++ b/bindings/c/src/filtered_search.hpp @@ -202,7 +202,7 @@ std::vector filtered_topk_search( batch_size = estimate_batch_size( total_checked, found, num_neighbors, batch_size, index_size ); - iterator.next(batch_size); + iterator.unsafe_next(batch_size); total_checked += iterator.size(); for (auto& neighbor : iterator.results()) { if (id_filter->is_member(neighbor.id())) { @@ -228,6 +228,14 @@ std::vector filtered_topk_search( } }; + // Follow the concurrent backend's get_threadpool_handle() execution contract. + [[maybe_unused]] auto read_guards = [&] { + if constexpr (requires { index.lock_read_guards(); }) { + return index.lock_read_guards(); + } else { + return svs::lib::Empty{}; + } + }(); svs::threads::parallel_for( index.get_threadpool_handle(), svs::threads::StaticPartition{queries.size()}, diff --git a/bindings/c/src/index.hpp b/bindings/c/src/index.hpp index 577520780..493dbff24 100644 --- a/bindings/c/src/index.hpp +++ b/bindings/c/src/index.hpp @@ -95,5 +95,9 @@ struct DynamicIndex : public Index { virtual bool has_id(size_t id) const = 0; virtual void consolidate() = 0; virtual void compact(size_t batchsize) = 0; + virtual void update_points( + svs::data::ConstSimpleDataView points, std::span ids + ) = 0; + virtual void replace_id(size_t old_id, size_t new_id) = 0; }; } // namespace svs::c_runtime diff --git a/bindings/c/src/index_vamana.cpp b/bindings/c/src/index_vamana.cpp index addc5eb34..30fb10e2e 100644 --- a/bindings/c/src/index_vamana.cpp +++ b/bindings/c/src/index_vamana.cpp @@ -236,6 +236,18 @@ size_t DynamicIndexVamana::delete_points_unlocked(std::span ids) { return ids_to_delete.size(); } +void DynamicIndexVamana::update_points( + svs::data::ConstSimpleDataView /*points*/, std::span /*ids*/ +) { + throw not_implemented("Updating points is only supported by FINE GRAIN dynamic indices." + ); +} + +void DynamicIndexVamana::replace_id(size_t old_id, size_t new_id) { + auto lock = write_lock(); + index.replace_external_id(old_id, new_id); +} + void DynamicIndexVamana::save(const std::filesystem::path& directory) { // Saving consolidates and compacts the index, so it requires exclusive access. auto lock = write_lock(); @@ -309,6 +321,19 @@ size_t ConcurrentIndexVamana::add_points( return ids.size(); } +void ConcurrentIndexVamana::update_points( + svs::data::ConstSimpleDataView points, std::span ids +) { + assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN); + if (points.size() > 1) { + throw not_implemented( + "FINE GRAIN dynamic index only supports updating one point at a time." + ); + } + auto lock = read_lock(); + index.update_entries(points, ids); +} + size_t ConcurrentIndexVamana::delete_points(std::span ids) { assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN); if (ids.size() > 1) { diff --git a/bindings/c/src/index_vamana.hpp b/bindings/c/src/index_vamana.hpp index 2a193981e..34b49624c 100644 --- a/bindings/c/src/index_vamana.hpp +++ b/bindings/c/src/index_vamana.hpp @@ -132,6 +132,12 @@ struct DynamicIndexVamana : public DynamicIndex { void compact(size_t batchsize) override; + void update_points( + svs::data::ConstSimpleDataView points, std::span ids + ) override; + + void replace_id(size_t old_id, size_t new_id) override; + size_t get_num_threads() const override { auto lock = read_lock(); return index.get_num_threads(); @@ -195,5 +201,14 @@ struct ConcurrentIndexVamana : public DynamicIndexVamana { auto lock = write_lock(); return index.get_memory_breakdown(); } + + void update_points( + svs::data::ConstSimpleDataView points, std::span ids + ) override; + + void replace_id(size_t old_id, size_t new_id) override { + auto lock = read_lock(); + index.replace_external_id(old_id, new_id); + } }; } // namespace svs::c_runtime diff --git a/bindings/c/src/svs_c.cpp b/bindings/c/src/svs_c.cpp index 6718861bf..ec85b8982 100644 --- a/bindings/c/src/svs_c.cpp +++ b/bindings/c/src/svs_c.cpp @@ -1588,6 +1588,52 @@ extern "C" bool svs_index_dynamic_delete_points( ); } +extern "C" bool svs_index_dynamic_update_points( + svs_index_h index, + const float* points, + const size_t* ids, + size_t num_points, + svs_error_h out_err +) { + using namespace svs::c_runtime; + return wrap_exceptions( + [&]() { + EXPECT_ARG_NOT_NULL(index); + EXPECT_ARG_NOT_NULL(points); + EXPECT_ARG_NOT_NULL(ids); + EXPECT_ARG_GT_THAN(num_points, 0); + auto dynamic_index_ptr = std::dynamic_pointer_cast(index->impl); + INVALID_ARGUMENT_IF( + dynamic_index_ptr == nullptr, "Index does not support dynamic updates" + ); + auto src_data = svs::data::ConstSimpleDataView( + points, num_points, dynamic_index_ptr->dimensions() + ); + dynamic_index_ptr->update_points(src_data, std::span(ids, num_points)); + return true; + }, + out_err + ); +} + +extern "C" bool svs_index_dynamic_replace_id( + svs_index_h index, size_t old_id, size_t new_id, svs_error_h out_err +) { + using namespace svs::c_runtime; + return wrap_exceptions( + [&]() { + EXPECT_ARG_NOT_NULL(index); + auto dynamic_index_ptr = std::dynamic_pointer_cast(index->impl); + INVALID_ARGUMENT_IF( + dynamic_index_ptr == nullptr, "Index does not support dynamic updates" + ); + dynamic_index_ptr->replace_id(old_id, new_id); + return true; + }, + out_err + ); +} + extern "C" bool svs_index_dynamic_has_id( svs_index_h index, size_t id, bool* out_has_id, svs_error_h out_err ) { diff --git a/include/svs/concurrent/dynamic_index.h b/include/svs/concurrent/dynamic_index.h index dacf5b2b5..c74480d3b 100644 --- a/include/svs/concurrent/dynamic_index.h +++ b/include/svs/concurrent/dynamic_index.h @@ -813,14 +813,17 @@ class MutableVamanaIndex { /// /// (3) Exclude compaction from the production of these internal IDs through this call. /// + /// (4) Each worker partition uses one mapping snapshot; concurrent mapping changes + /// may be observed between partitions. + /// template requires(std::tuple_size_v == 2) void translate_to_external(DenseArray& ids) { - std::shared_lock lock{*translator_mutex_}; threads::parallel_for( threadpool_, threads::StaticPartition{getsize<0>(ids)}, [&](const auto is, uint64_t /*tid*/) { + std::shared_lock lock{*translator_mutex_}; for (auto i : is) { for (size_t j = 0, jmax = getsize<1>(ids); j < jmax; ++j) { auto internal = lib::narrow_cast(ids.at(i, j)); @@ -910,6 +913,7 @@ class MutableVamanaIndex { /// @brief Hold shared payload protection for a complete read operation. /// Acquire compact_mutex_ first, then this guard, then translator/node locks. + /// Follow the pool execution contract in get_threadpool_handle(). [[nodiscard]] std::shared_lock lock_payload_shared() const { std::shared_lock gatekeeper_lock{*payload_gatekeeper_}; return std::shared_lock(*payload_mutex_); @@ -1832,11 +1836,33 @@ class MutableVamanaIndex { /// /// @brief Return the current thread pool handle. /// - /// Do not call index methods that acquire storage/payload protection from tasks on - /// this index's underlying pool, including through another handle to that pool. - /// The submitting thread must acquire those guards before submitting work; tasks - /// use `unsafe_*` read paths under those guards. Otherwise a writer can close the - /// gate and wait for the pool while its tasks wait at that gate. + /// Thread pool lock rules: + /// A task is a function run by this pool, even on the caller's thread. + /// + /// 1. Outside tasks only: + /// compact_mutex_, payload_mutex_, compact_gatekeeper_, payload_gatekeeper_. + /// Take the needed locks before submitting work. Keep the storage/payload locks + /// until that work finishes. Unlock them on the thread that locked them. + /// A task must not call methods that take these locks. + /// unsafe_next() uses the caller's storage and payload locks. + /// + /// 2. Other index locks may be taken inside or outside tasks. + /// Before submitting or waiting for work, the caller must release locks that + /// any task on this pool may take, including tasks of other callers. + /// This includes translator_mutex_, node locks and reverse-edge locks. + /// Shared locks follow this rule too. + /// For unsafe_has_id(), unsafe_on_ids() and unsafe_translate_*(), the task + /// must hold translator_mutex_ itself. + /// + /// 3. A task must not submit work to this pool and wait for it, + /// unless the pool explicitly supports this. + /// + /// 4. A task must not wait for a Pending slot. + /// The slot's owner may need this pool to finish. + /// + /// Another handle to this pool follows the same rules. + /// Follow the lock order listed beside the mutexes. + /// Update holds payload_mutex_ exclusive only while copying data. threads::ThreadPoolHandle& get_threadpool_handle() { return threadpool_; } ///// Window Interface @@ -2078,8 +2104,10 @@ class MutableVamanaIndex { /// - ``svs::getsize<1>(dst) == dimensions()``: The space allocated for each vector in /// ``dst`` is correct. /// - /// An exception will be thrown if any of these pre-conditions does not hold. - /// If such an exception is thrown, the argument ``dst`` will be left unmodified. + /// Initial validation checks these preconditions before writing to ``dst``; failure + /// leaves it unmodified. Each worker protects its ID mappings and accepts mapped + /// Valid or Deleted slots. A missing mapping or a Pending or Empty slot causes an + /// exception after the workers finish. ``dst`` may then contain partial results. template void reconstruct_at(data::SimpleDataView dst, std::span ids) { const size_t ids_size = ids.size(); @@ -2098,34 +2126,53 @@ class MutableVamanaIndex { ); } - // Lock order: compact_mutex_ then translator_mutex_ (global order). - // compact_mutex_ shared guards data_/graph_ against compact()'s shrink; - // translator_mutex_ shared guards the ID translation reads below. auto compact_lock = lock_storage_shared(); auto payload_lock = lock_payload_shared(); - std::shared_lock lock{*translator_mutex_}; - // Bounds checking. - for (size_t i = 0; i < ids_size; ++i) { - I id = ids[i]; // inbounds by loop bounds. - if (!unsafe_has_id(id)) { - throw ANNEXCEPTION("ID {} with value {} is out of bounds!", i, id); + { + std::shared_lock lock{*translator_mutex_}; + for (size_t i = 0; i < ids_size; ++i) { + I id = ids[i]; // inbounds by loop bounds. + if (!unsafe_has_id(id)) { + throw ANNEXCEPTION("ID {} with value {} is out of bounds!", i, id); + } } } + // Some pools do not propagate task exceptions. Report a concurrent ID failure + // on the caller after all workers have finished using the guarded storage. + std::atomic unavailable_id{ids_size}; // Prerequisites checked - proceed with the operation. // TODO: Communicate the requested decompression type to the backend dataset to // allow more fine-grained specialization? auto threaded_function = [&](auto is, uint64_t SVS_UNUSED(tid)) { + // Keep translation and reconstruction under one lock so consolidation cannot + // reclaim the resolved slot before its payload is read. + std::shared_lock lock{*translator_mutex_}; auto accessor = extensions::reconstruct_accessor(data_); for (auto i : is) { - auto id = unsafe_translate_external_id(ids[i]); - dst.set_datum(i, accessor(data_, id)); + if (translator_.has_external(ids[i])) { + auto id = unsafe_translate_external_id(ids[i]); + auto state = std::atomic_ref(status_[id]) + .load(std::memory_order_acquire); + // Pending insertions can still write despite our shared payload lock. + if (state == SlotMetadata::Valid || state == SlotMetadata::Deleted) { + dst.set_datum(i, accessor(data_, id)); + continue; + } + } + unavailable_id.store(i, std::memory_order_relaxed); + return; } }; threads::parallel_for( threadpool_, threads::StaticPartition{ids_size}, threaded_function ); + if (auto i = unavailable_id.load(std::memory_order_relaxed); i != ids_size) { + throw ANNEXCEPTION( + "ID {} with value {} became unavailable during reconstruction!", i, ids[i] + ); + } } /// Invoke the provided callable with constant references to the contained graph, data, diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h index 261278d7b..43d51976d 100644 --- a/include/svs/orchestrators/dynamic_vamana.h +++ b/include/svs/orchestrators/dynamic_vamana.h @@ -22,7 +22,9 @@ #include "svs/orchestrators/vamana.h" // stdlib +#include #include +#include namespace svs { @@ -56,6 +58,27 @@ class DynamicVamanaInterface : public VamanaInterface { // Check if the index has been consolidated - no `Deleted` entries remain. virtual bool is_consolidated() const = 0; + + // Overwrite the vectors of existing external IDs. Only supported by backends that + // provide `update_entries` (the concurrent index). + virtual void update_entries( + const float* data, size_t dim0, size_t dim1, std::span ids + ) = 0; + + // Rename an external ID. Throws `std::invalid_argument` if `old_id` is missing or + // `new_id` already exists. + virtual void replace_external_id(size_t old_id, size_t new_id) = 0; + + using ReadGuards = + std::pair, std::shared_lock>; + + /// @brief Acquire shared storage and payload protection before submitting read work. + /// Follow the concurrent backend's get_threadpool_handle() execution contract. + /// Backends without these locks return empty guards. The default reports this + /// operation as unsupported. + virtual ReadGuards lock_read_guards() const { + throw ANNEXCEPTION("Guarded reads are not supported by this index adapter."); + } }; template @@ -109,6 +132,70 @@ class DynamicVamanaImpl : public VamanaImpl ids + ) override { + if constexpr (requires { + impl().update_entries( + std::declval>(), ids + ); + }) { + auto points = data::ConstSimpleDataView(data, dim0, dim1); + impl().update_entries(points, ids); + } else { + throw ANNEXCEPTION("update_entries is not supported by this index type."); + } + } + + void replace_external_id(size_t old_id, size_t new_id) override { + using Result = decltype(impl().replace_external_id(old_id, new_id)); + if constexpr (std::is_void_v) { + // Non-concurrent index: throws ANNException, so check up front to report + // invalid arguments uniformly. + if (!impl().has_id(old_id)) { + throw std::invalid_argument( + fmt::format("Index does not contain external ID {}!", old_id) + ); + } + if (impl().has_id(new_id)) { + throw std::invalid_argument( + fmt::format("Index already contains external ID {}!", new_id) + ); + } + impl().replace_external_id(old_id, new_id); + } else { + switch (impl().replace_external_id(old_id, new_id)) { + case Result::Ok: + return; + case Result::OldIdMissing: + throw std::invalid_argument( + fmt::format("Index does not contain external ID {}!", old_id) + ); + case Result::NewIdExists: + throw std::invalid_argument( + fmt::format("Index already contains external ID {}!", new_id) + ); + } + } + } + + DynamicVamanaInterface::ReadGuards lock_read_guards() const override { + if constexpr (requires { + impl().lock_storage_shared(); + impl().lock_payload_shared(); + }) { + return {impl().lock_storage_shared(), impl().lock_payload_shared()}; + } else if constexpr (requires { + impl().get_parent_index().lock_storage_shared(); + impl().get_parent_index().lock_payload_shared(); + }) { + const auto& parent = impl().get_parent_index(); + return {parent.lock_storage_shared(), parent.lock_payload_shared()}; + } else { + return {}; + } + } }; // Forward Declaractions. @@ -199,6 +286,29 @@ class DynamicVamana : public manager::IndexManager { return *this; } + /// + /// @brief Overwrite the vectors of existing points. Only supported by the + /// concurrent index; throws otherwise. + /// + DynamicVamana& + update_entries(data::ConstSimpleDataView points, std::span ids) { + impl_->update_entries(points.data(), points.size(), points.dimensions(), ids); + return *this; + } + + /// + /// @brief Rename external ID `old_id` to `new_id`. + /// + /// Throws `std::invalid_argument` if `old_id` is missing or `new_id` already exists. + /// + DynamicVamana& replace_external_id(size_t old_id, size_t new_id) { + impl_->replace_external_id(old_id, new_id); + return *this; + } + + /// @copydoc DynamicVamanaInterface::lock_read_guards + [[nodiscard]] auto lock_read_guards() const { return impl_->lock_read_guards(); } + // Accessors /// @copydoc svs::index::vamana::MutableVamanaIndex::get_alpha float get_alpha() const { return impl_->get_alpha(); } diff --git a/include/svs/orchestrators/vamana_iterator.h b/include/svs/orchestrators/vamana_iterator.h index 76ca8c9fc..881ea26d7 100644 --- a/include/svs/orchestrators/vamana_iterator.h +++ b/include/svs/orchestrators/vamana_iterator.h @@ -45,6 +45,9 @@ class VamanaIterator { virtual void update(svs::AnonymousArray<1> newquery) = 0; virtual ~Interface() = default; + + virtual void + unsafe_next(size_t batch_size, const lib::DefaultPredicate& cancel) = 0; }; template struct Implementation : Interface { @@ -81,6 +84,13 @@ class VamanaIterator { ) override { impl_.next(batch_size, cancel); } + void unsafe_next(size_t batch_size, const lib::DefaultPredicate& cancel) override { + if constexpr (requires { impl_.unsafe_next(batch_size, cancel); }) { + impl_.unsafe_next(batch_size, cancel); + } else { + impl_.next(batch_size, cancel); + } + } bool done() const override { return impl_.done(); } // Query Updates @@ -152,6 +162,16 @@ class VamanaIterator { impl_->next(batch_size, cancel); } + /// @copydoc next + /// The caller must hold the index's shared storage and payload guards through + /// this call. Backends without guarded reads use their normal next(). + void unsafe_next( + size_t batch_size, + const lib::DefaultPredicate& cancel = lib::Returns(lib::Const()) + ) { + impl_->unsafe_next(batch_size, cancel); + } + /// @brief Signal that the next batch search should begin entirely from scratch. /// /// The iterator records some internal state to accelerate future calls to ``next()``.