Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions bindings/c/include/svs/c/svs_c.h
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 9 additions & 1 deletion bindings/c/src/filtered_search.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ std::vector<size_t> 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())) {
Expand All @@ -228,6 +228,14 @@ std::vector<size_t> 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()},
Expand Down
4 changes: 4 additions & 0 deletions bindings/c/src/index.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<float> points, std::span<const size_t> ids
) = 0;
virtual void replace_id(size_t old_id, size_t new_id) = 0;
};
} // namespace svs::c_runtime
25 changes: 25 additions & 0 deletions bindings/c/src/index_vamana.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,18 @@ size_t DynamicIndexVamana::delete_points_unlocked(std::span<const size_t> ids) {
return ids_to_delete.size();
}

void DynamicIndexVamana::update_points(
svs::data::ConstSimpleDataView<float> /*points*/, std::span<const size_t> /*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();
Expand Down Expand Up @@ -309,6 +321,19 @@ size_t ConcurrentIndexVamana::add_points(
return ids.size();
}

void ConcurrentIndexVamana::update_points(
svs::data::ConstSimpleDataView<float> points, std::span<const size_t> 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<const size_t> ids) {
assert(sync_kind == SVS_SYNC_KIND_FINE_GRAIN);
if (ids.size() > 1) {
Expand Down
15 changes: 15 additions & 0 deletions bindings/c/src/index_vamana.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,12 @@ struct DynamicIndexVamana : public DynamicIndex {

void compact(size_t batchsize) override;

void update_points(
svs::data::ConstSimpleDataView<float> points, std::span<const size_t> 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();
Expand Down Expand Up @@ -195,5 +201,14 @@ struct ConcurrentIndexVamana : public DynamicIndexVamana {
auto lock = write_lock();
return index.get_memory_breakdown();
}

void update_points(
svs::data::ConstSimpleDataView<float> points, std::span<const size_t> 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
46 changes: 46 additions & 0 deletions bindings/c/src/svs_c.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<DynamicIndex>(index->impl);
INVALID_ARGUMENT_IF(
dynamic_index_ptr == nullptr, "Index does not support dynamic updates"
);
auto src_data = svs::data::ConstSimpleDataView<float>(
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<DynamicIndex>(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
) {
Expand Down
85 changes: 66 additions & 19 deletions include/svs/concurrent/dynamic_index.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 <class Dims, class Base>
requires(std::tuple_size_v<Dims> == 2)
void translate_to_external(DenseArray<size_t, Dims, Base>& 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<Idx>(ids.at(i, j));
Expand Down Expand Up @@ -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<std::shared_mutex> lock_payload_shared() const {
std::shared_lock gatekeeper_lock{*payload_gatekeeper_};
return std::shared_lock<std::shared_mutex>(*payload_mutex_);
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 <std::unsigned_integral I, svs::Arithmetic T>
void reconstruct_at(data::SimpleDataView<T> dst, std::span<const I> ids) {
const size_t ids_size = ids.size();
Expand All @@ -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<size_t> 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<SlotMetadata>(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,
Expand Down
Loading
Loading