Repository navigation
Add update_entries method to concurrent Vamana Index - #467
razdoburdin wants to merge 1 commit into
Conversation
|
When reusing deleted slots with |
|
Benchmark and profile of Setup: 50K x 128-d FLOAT32, L2, FINE_GRAIN at 9330e49, 8 writer and 8 reader clients. Each churn round deletes, inserts and updates 2,500 keys. Inserts reuse deleted slots (
Profile (perf plus all-thread stack samples, during the insert phase):
Likely path forward on the valkey-search side, pending exploration to validate: keep |
|
Follow-up to the profile above: numbers for the alternative we prototyped on the valkey-search side. Same setup (50K x 128-d, L2, FINE_GRAIN at 9330e49, 8 writer and 8 reader clients, each round deletes, inserts and updates 2,500 keys, EF_RUNTIME 100). The prototype keeps
What we take from it so far:
This is still exploration on our side, so no change is requested yet. If we go this way, the C API piece would be exposing |
|
Follow-up for review: two more prototypes for the remaining consolidation pause, measured from valkey-search. Same setup as above (50K x 128-d FLOAT32, L2, FINE_GRAIN, 8 writer and 8 reader clients, 10 churn rounds that each delete, insert and update 2,500 keys). All SVS changes are local patches on Context: with consolidation moved to a background task in valkey-search, the remaining pauses are (a) the exclusive 1. Batches through 2. A resumable full pass: removes the pauses.
valkey-search runs one step per background task.
A pass takes 12-14 steps and 13-23 s of wall time, and it keeps up with the deletes. Recall@10 after churn is 0.788 against 0.785-0.807 for the other runs. Open points in the prototype:
Proposal for discussion:
valkey-search keeps its current approach (background consolidate + compact) until these exist. Prototype patch on main 0b4b1c3 (experimental, not for merge)diff --git a/bindings/c/include/svs/c/svs_c.h b/bindings/c/include/svs/c/svs_c.h
index 0826c16..628f99e 100644
--- a/bindings/c/include/svs/c/svs_c.h
+++ b/bindings/c/include/svs/c/svs_c.h
@@ -1657,6 +1657,12 @@ SVS_API bool svs_index_reconstruct(
SVS_API bool
svs_index_dynamic_consolidate(svs_index_h index, svs_error_h out_err /*=NULL*/);
+/// EXPERIMENT: one step of a resumable consolidation (FINE_GRAIN only). Repairs up to
+/// @p max_work nodes; sets @p out_done when the pass has finished and freed its slots.
+SVS_API bool svs_index_dynamic_consolidate_step(
+ svs_index_h index, size_t max_work, bool* out_done, svs_error_h out_err /*=NULL*/
+);
+
/// @brief Compact a dynamic index to remove deleted entries and optimize storage
/// @param index The dynamic index handle
/// @param batchsize The batch size for compaction (0 for default)
diff --git a/bindings/c/src/index.hpp b/bindings/c/src/index.hpp
index 5775207..8944ab1 100644
--- a/bindings/c/src/index.hpp
+++ b/bindings/c/src/index.hpp
@@ -94,6 +94,10 @@ struct DynamicIndex : public Index {
virtual size_t delete_points(std::span<const size_t> ids) = 0;
virtual bool has_id(size_t id) const = 0;
virtual void consolidate() = 0;
+ // EXPERIMENT: one step of a resumable consolidation.
+ virtual bool consolidate_step(size_t /*max_work*/) {
+ throw not_implemented("consolidate_step needs SVS_SYNC_KIND_FINE_GRAIN");
+ }
virtual void compact(size_t batchsize) = 0;
};
} // namespace svs::c_runtime
diff --git a/bindings/c/src/index_vamana.cpp b/bindings/c/src/index_vamana.cpp
index addc5eb..e2f09d4 100644
--- a/bindings/c/src/index_vamana.cpp
+++ b/bindings/c/src/index_vamana.cpp
@@ -305,7 +305,7 @@ size_t ConcurrentIndexVamana::add_points(
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);
+ index.add_points(new_points, ids, /*reuse_empty=*/true); // EXPERIMENT
return ids.size();
}
diff --git a/bindings/c/src/index_vamana.hpp b/bindings/c/src/index_vamana.hpp
index 2a19398..999e8cf 100644
--- a/bindings/c/src/index_vamana.hpp
+++ b/bindings/c/src/index_vamana.hpp
@@ -184,6 +184,11 @@ struct ConcurrentIndexVamana : public DynamicIndexVamana {
index.consolidate();
}
+ bool consolidate_step(size_t max_work) override {
+ auto lock = read_lock();
+ return index.consolidate_step(max_work);
+ }
+
void compact(size_t batchsize) override {
// exclusive lock prevents potential deadlock in case of custom threadpool takes
// internal mutex in parallel_for method
diff --git a/bindings/c/src/svs_c.cpp b/bindings/c/src/svs_c.cpp
index 6718861..a3de170 100644
--- a/bindings/c/src/svs_c.cpp
+++ b/bindings/c/src/svs_c.cpp
@@ -1588,6 +1588,25 @@ extern "C" bool svs_index_dynamic_delete_points(
);
}
+extern "C" bool svs_index_dynamic_consolidate_step(
+ svs_index_h index, size_t max_work, bool* out_done, svs_error_h out_err
+) {
+ using namespace svs::c_runtime;
+ return wrap_exceptions(
+ [&]() {
+ EXPECT_ARG_NOT_NULL(index);
+ EXPECT_ARG_NOT_NULL(out_done);
+ auto dynamic_index_ptr = std::dynamic_pointer_cast<DynamicIndex>(index->impl);
+ INVALID_ARGUMENT_IF(
+ dynamic_index_ptr == nullptr, "Index does not support dynamic updates"
+ );
+ *out_done = dynamic_index_ptr->consolidate_step(max_work);
+ 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 1fc1a89..7a6f87b 100644
--- a/include/svs/concurrent/dynamic_index.h
+++ b/include/svs/concurrent/dynamic_index.h
@@ -1617,6 +1617,87 @@ class MutableVamanaIndex {
consolidate_locked();
}
+ // EXPERIMENT: resumable consolidation. The first step snapshots the Deleted set
+ // and gathers its in-neighbors once; each step repairs up to `max_work` of them;
+ // the last step frees the slots as consolidate_locked() does. Returns true when
+ // the pass is complete (or there was nothing to do).
+ struct ConsolidationProgress {
+ tsl::robin_set<Idx> deleted{};
+ std::vector<size_t> work{};
+ size_t next = 0;
+ Idx anchor{};
+ bool active = false;
+ };
+ std::unique_ptr<ConsolidationProgress> progress_ =
+ std::make_unique<ConsolidationProgress>();
+ std::unique_ptr<std::mutex> progress_mutex_ = std::make_unique<std::mutex>();
+
+ bool consolidate_step(size_t max_work) {
+ auto compact_lock = lock_storage_shared();
+ std::lock_guard step_lock{*progress_mutex_};
+ auto& p = *progress_;
+ ConsolidationParameters params{200'000, prune_to_, max_candidates_, alpha_};
+ auto consolidator = GraphConsolidator{graph_, data_, threadpool_, distance_, params};
+ if (!p.active) {
+ p.deleted.clear();
+ for (size_t i = 0, imax = status_.size(); i < imax; ++i) {
+ if (std::atomic_ref<SlotMetadata>(status_[i])
+ .load(std::memory_order_acquire) == SlotMetadata::Deleted) {
+ p.deleted.insert(lib::narrow_cast<Idx>(i));
+ }
+ }
+ if (p.deleted.empty()) {
+ return true;
+ }
+ p.anchor = entry_point();
+ auto gather_remove = [&](size_t i) {
+ return i != p.anchor && p.deleted.contains(lib::narrow_cast<Idx>(i));
+ };
+ p.work = consolidator.gather_work_set(p.deleted, gather_remove);
+ p.next = 0;
+ p.active = true;
+ }
+ auto should_remove = [&](size_t i) {
+ return i != p.anchor && p.deleted.contains(lib::narrow_cast<Idx>(i));
+ };
+ const size_t end = std::min(p.work.size(), p.next + std::max<size_t>(max_work, 1));
+ std::vector<size_t> slice(p.work.begin() + p.next, p.work.begin() + end);
+ if (!slice.empty()) {
+ consolidator.run_driver(slice, should_remove);
+ }
+ p.next = end;
+ if (p.next < p.work.size()) {
+ return false;
+ }
+ {
+ std::lock_guard lock{*translator_mutex_};
+ std::vector<size_t> deleted_internal_ids;
+ for (auto i : p.deleted) {
+ if (status_[i] == SlotMetadata::Deleted && translator_.has_internal(i)) {
+ deleted_internal_ids.push_back(i);
+ }
+ }
+ if (!deleted_internal_ids.empty()) {
+ translator_.delete_internal(deleted_internal_ids, false);
+ }
+ size_t min_freed = std::numeric_limits<size_t>::max();
+ for (auto i : p.deleted) {
+ if (i != p.anchor && status_[i] == SlotMetadata::Deleted) {
+ std::atomic_ref<SlotMetadata>(status_[i])
+ .store(SlotMetadata::Empty, std::memory_order_release);
+ min_freed = std::min(min_freed, static_cast<size_t>(i));
+ }
+ }
+ if (min_freed != std::numeric_limits<size_t>::max()) {
+ detail::atomic_min(*first_reusable_, min_freed);
+ }
+ }
+ p.active = false;
+ p.deleted.clear();
+ p.work.clear();
+ return true;
+ }
+
// Full consolidation: gather the Deleted set (one cheap byte scan of status_) then
// delegate. Concurrent deletes arriving after the scan are picked up next round.
void consolidate_locked() {
diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h
index 261278d..f99eaa1 100644
--- a/include/svs/orchestrators/dynamic_vamana.h
+++ b/include/svs/orchestrators/dynamic_vamana.h
@@ -45,6 +45,8 @@ class DynamicVamanaInterface : public VamanaInterface {
virtual void delete_points(std::span<const size_t> ids) = 0;
virtual void consolidate() = 0;
+ // EXPERIMENT: one step of a resumable consolidation; true when complete.
+ virtual bool consolidate_step(size_t max_work) = 0;
virtual void compact(size_t batchsize = 1'000'000) = 0;
// ID inspection.
@@ -85,6 +87,13 @@ class DynamicVamanaImpl : public VamanaImpl<QueryTypes, Impl, DynamicVamanaInter
void delete_points(std::span<const size_t> ids) override { impl().delete_entries(ids); }
void consolidate() override { impl().consolidate(); }
+ bool consolidate_step(size_t max_work) override {
+ if constexpr (requires { impl().consolidate_step(max_work); }) {
+ return impl().consolidate_step(max_work);
+ } else {
+ throw ANNEXCEPTION("consolidate_step needs the concurrent index");
+ }
+ }
void compact(size_t batchsize) override { impl().compact(batchsize); }
// ID inspection.
@@ -177,6 +186,7 @@ class DynamicVamana : public manager::IndexManager<DynamicVamanaInterface> {
impl_->consolidate();
return *this;
}
+ bool consolidate_step(size_t max_work) { return impl_->consolidate_step(max_work); }
DynamicVamana& compact(size_t batchsize = 1'000'000) {
impl_->compact(batchsize); |
|
Follow-up to the comment above, with two corrections. 1. The prototype crashed. In one of three uniform-data runs, the valkey-server process died during the SAVE that follows the churn. At that point 13,395 slots were deleted and a resumable pass was probably in progress. My reading, not yet confirmed with a stack trace (that run kept no server log):
So a resumable pass has to coordinate with save and compaction. Either finish or abandon the pending pass first, restoring the reverse entries it cleared, or have save and compaction skip the slots the pass holds. Treat the resumable numbers below as an estimate until that is fixed. 2. Repeated runs at about 98% recall. The earlier numbers were single runs at
The conclusions from the earlier comment hold: a resumable pass, plus refilling freed slots, removes the multi-second pauses on both datasets, and saves need #421. The crash makes coordination with save and compaction a requirement for any real implementation, not an edge case. |
|
Correction to the follow-up above: the "Cohere: insert p99" row (60 ms / 1.34 s / 52 ms) is misleading. It is not a regression in the background-consolidation build. I timed every churn write and lined them up with the consolidation and compaction windows in the server log (Cohere, 8K live x 768-d, 8 writer clients):
The benchmark now reports whole-run percentiles per phase. The other rows and conclusions above are unchanged: compaction is the remaining pause in that build, and the resumable pass removes it. |
This PR adds a way to change the vectors already stored in the index in a concurrent regime with search/delete/insert/conslidate
update_entries(points, external_ids)replaces the vectors with existing IDs. Both the internal ID and the external ID stay the same.
During an update, the slots are
Pending. They are hidden from search. After the update they are Valid again. A delete of the same ID waits until update finisghes.Implementation add a new shared mutex
payload_mutex_it is taken exclusive only byupdate_entriesand only for a short payload replacement phase. All readers of the vectors take it shared. After the payload is replaced, consolidation of the incoming connections is called to repair the links to the old (deleted) vector, after that new outcoming connections are established (this logic is shared with theadd_pointsmethod).