diff --git a/include/svs/concurrent/README.md b/include/svs/concurrent/README.md new file mode 100644 index 000000000..06caab7b0 --- /dev/null +++ b/include/svs/concurrent/README.md @@ -0,0 +1,116 @@ +# `svs::concurrent` — a dynamic Vamana index with lock-free search + +A second dynamic Vamana index type whose searches never block, and never need to be excluded +from concurrent insertion or deletion. + +**Nothing outside this directory changes.** No existing header is modified. The static +`VamanaIndex` and the existing `MutableVamanaIndex` are bit-for-bit unaffected and pay +nothing — no extra atomic load, no extra indirection, no extra byte per node. + +## What it gives you, and what it does not + +| | `svs::index::vamana::MutableVamanaIndex` | `svs::concurrent::MutableVamanaIndex` | +|---|---|---| +| search ‖ search | yes | yes | +| search ‖ insert | **caller must exclude** | **lock-free** | +| search ‖ delete | **caller must exclude** | **lock-free** | +| insert ‖ insert | caller must exclude | serialized internally | +| `consolidate()` / `compact()` | caller must exclude | stop-the-world (internally) | +| multi-label (`MultiMutableVamanaIndex`) | yes | `static_assert` — not supported | +| save / load | yes | **not supported** (`supports_saving = false`) | + +The trade is deliberate: writers are serialized behind one mutex so that the only +synchronization on the *search* path is a per-node sequence lock. Applications where query +throughput matters and ingest is a single background stream get lock-free search for a +one-line type change. Applications that need concurrent writers or serialization should keep +using the existing index. + +## How it works + +1. **`SeqLockGraph`** (`graph.h`) stores adjacency lists in a `lib::SegmentedVector`, so + growing the graph never moves an existing node's slot — a reader holding a pointer into + node *i* is unaffected by a concurrent `unsafe_resize`. Each node carries a 1-byte + `SeqLockCounter`. Elements are accessed through relaxed `std::atomic_ref`, which is a + plain `MOV` on x86 but makes the concurrent access race-free rather than UB. + +2. **`seqlock_greedy_search`** (`greedy_search.h`) wraps each node expansion in + `read_begin` / `read_validate`. A rejected read is safe to retry: the graph never + publishes a degree covering an unwritten slot, so anything already inserted into the + search buffer has a valid ID and a correctly computed distance, and `insert` dedupes by + ID. A retry costs redundant work, never correctness. + +3. **`SeqLockGraphView`** (`graph_view.h`) pushes that same retry down into `get_node`, + returning a certified snapshot copied into per-thread scratch. This is what lets + **unmodified** upstream code — notably the 340-line `BatchIterator` — read the graph + safely with no changes at all. + +4. **`MutableVamanaIndex`** (`mutable_vamana_index.h`) holds one `writer_mutex_` to + serialize writers, and two `WriterPriorityMutex`es held *shared* by searches: one for + structural changes (dataset capacity growth) and one for ID-translator remaps. + +`SeqLockGraph` satisfies the **unmodified** `svs::graphs::MemoryGraph` concept — including +`add_edge`'s `size_t` return — which is why `VamanaBuilder`, `prune` and `GraphConsolidator` +all work against it with no edits. There are `static_assert`s to that effect in +`tests/svs/concurrent/graph.cpp`; if a future concept change breaks the arrangement, those +fire at the concept boundary rather than deep inside a template. + +## Using it + +```cpp +#include "svs/concurrent/mutable_vamana_index.h" + +using Index = svs::concurrent::MutableVamanaIndex< + uint32_t, svs::data::BlockedData, svs::distance::DistanceL2>; + +// Build with the *stock* builder over a SeqLockGraph. +auto graph = svs::concurrent::SeqLockGraph{data.size(), max_degree}; +auto builder = svs::index::vamana::VamanaBuilder{ + graph, data, distance, parameters, threadpool, prefetch}; +builder.construct(alpha, entry_point); + +auto index = Index{ + std::move(graph), std::move(data), entry_point, distance, ids, threadpool}; + +// From here on, `index.search(...)` may run on any number of threads while another thread +// calls `add_points` / `delete_entries`. No external lock. +``` + +`consolidate()` and `compact()` are the exceptions: they take the structure lock +exclusively, so searches stall for their duration. + +## Testing + +`tests/svs/concurrent/{graph,mutable_vamana_index}.cpp`, tagged `[concurrent]`. + +Because the correctness argument here is almost entirely about memory ordering, a +non-instrumented test can only fail to disprove it. Two opt-in ThreadSanitizer targets close +that gap: + +```sh +cmake -DSVS_BUILD_TESTS=YES -DSVS_EXPERIMENTAL_ENABLE_CONCURRENT_TSAN=YES ... +ctest -L tsan +``` + +`concurrent_tsan` must be clean. `concurrent_tsan_negative` compiles the same graph test +with `-DSVS_CONCURRENT_UNSAFE_PLAIN_GRAPH_ACCESS`, which swaps the relaxed `atomic_ref` +accessors for plain loads and stores, and is registered `WILL_FAIL` — it must report races +in `SeqLockGraph::load_`. Without that negative control a clean run proves only that TSan +was not looking at the interesting memory. + +## Known limitations + +* **No serialization.** SVS's save/load is templated on the graph type; `SeqLockGraph` needs + its own serializer and a matching loader. Mechanical, but not done. +* **No allocator.** `SegmentedVector` takes no allocator, so graph memory is invisible to + allocator-based accounting. +* **`SeqLockCounter` is a `uint8_t`.** A reader descheduled across ≥128 writes to the *same* + node can observe a matching counter after wraparound and accept a torn read. Widening it is + a one-line change in `lib/concurrency/seqlock.h`. +* **Uncompressed storage only.** The analysis of which accesses may race was done against + `svs::data::SimpleData<..., Blocked<...>>`. LVQ and LeanVec have their own layouts and + growth behaviour and have not been reviewed. +* **`WriterPriorityMutex` has a slower fallback off glibc.** See + `lib/concurrency/writer_priority_mutex.h`: the glibc path keeps `std::shared_mutex`'s + atomic fast path, the portable path takes an uncontended `std::mutex` per `lock_shared`. +* **Code duplication.** The graph and the search loop are parallel copies of upstream's and + will drift. This is the standing cost of keeping the change additive. diff --git a/include/svs/concurrent/graph.h b/include/svs/concurrent/graph.h new file mode 100644 index 000000000..19a8ec7a4 --- /dev/null +++ b/include/svs/concurrent/graph.h @@ -0,0 +1,360 @@ +/* + * 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 "svs/lib/concurrency/atomic_span.h" +#include "svs/lib/concurrency/seqlock.h" +#include "svs/lib/segmented_vector.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace svs::concurrent { + +/// +/// @brief A grow-stable adjacency-list graph with per-node sequence locks. +/// +/// Layout mirrors ``svs::graphs::SimpleGraphBase``: node ``i`` owns ``stride = +/// max_degree + 1`` contiguous ``Idx`` slots, the first holding the out-degree and the +/// remainder holding neighbor IDs. Unlike ``SimpleBlockedGraph`` -- whose block +/// *descriptor* array is a ``std::vector`` and therefore reallocates on growth -- nodes +/// live in fixed-size segments held by a ``lib::SegmentedVector``, so the address of a +/// node's slot is stable for that node's lifetime. +/// +/// Concurrency contract: +/// +/// * **One writer at a time.** ``add_edge`` / ``clear_node`` / ``replace_node`` / +/// ``unsafe_resize`` must be serialized by the caller. Because there is only ever one +/// writer, the per-node sequence counters need no writer-writer serialization and no +/// per-node spinlock is required. +/// * **Many concurrent readers.** ``get_node_atomic`` paired with ``seqlock(i)`` gives a +/// race-free read of node ``i``'s adjacency list concurrently with a writer. See +/// ``svs::concurrent::seqlock_greedy_search`` for the retry protocol. +/// * ``get_node`` performs *plain* (non-atomic) loads and is intended for the writer +/// itself and for build-time code that runs before the graph is published to readers +/// (e.g. ``svs::index::vamana::VamanaBuilder``). +/// * **Shrinking frees segments.** ``unsafe_resize`` to a smaller size may free storage, +/// so the caller must have drained readers first (e.g. via an exclusive lock). +/// +/// This type satisfies the *unmodified* ``svs::graphs::MemoryGraph`` concept -- in +/// particular ``add_edge`` keeps its ``size_t`` return -- so upstream ``VamanaBuilder`` +/// and ``prune`` work against it with no changes to SVS. +/// +template class SeqLockGraph { + public: + using index_type = Idx; + using reference = std::span; + using const_reference = std::span; + // Mirrors ``svs::graphs::SimpleGraphBase``. Not required by the graph concepts, but + // ``svs::index::vamana::GraphConsolidator`` reads ``Graph::const_value_type``. + using value_type = std::span; + using const_value_type = std::span; + + /// @brief Default number of nodes per segment. + static constexpr size_t default_segment_size = 512; + + private: + /// + /// @brief One contiguous, never-relocated run of ``segment_size * stride`` slots. + /// + /// Slots are zero-initialized on allocation. That matters for concurrent readers: a + /// reader that observes a torn degree can only ever read neighbor slots that were + /// zero-initialized or validly written, never indeterminate memory, so it cannot + /// index out of the dataset before the sequence-lock validation rejects the read. + /// + class Segment { + public: + Segment() = default; + explicit Segment(size_t n) + : n_{n} + , storage_{new Idx[n]()} {} + + Segment(const Segment& other) + : n_{other.n_} + , storage_{other.storage_ ? new Idx[other.n_] : nullptr} { + if (storage_) { + std::copy(other.storage_.get(), other.storage_.get() + n_, storage_.get()); + } + } + Segment& operator=(const Segment& other) { + if (this != &other) { + auto tmp = Segment(other); + *this = std::move(tmp); + } + return *this; + } + Segment(Segment&&) noexcept = default; + Segment& operator=(Segment&&) noexcept = default; + ~Segment() = default; + + Idx* data() noexcept { return storage_.get(); } + const Idx* data() const noexcept { return storage_.get(); } + + private: + size_t n_{0}; + std::unique_ptr storage_{}; + }; + + public: + SeqLockGraph() = default; + + /// + /// @brief Construct a graph with ``num_nodes`` nodes, each with capacity + /// ``max_degree``. + /// + SeqLockGraph( + size_t num_nodes, size_t max_degree, size_t segment_size = default_segment_size + ) + : max_degree_{max_degree} + , stride_{max_degree + 1} + , segment_size_{segment_size} { + assert(segment_size_ > 0); + grow_to(num_nodes); + n_nodes_.store(num_nodes, std::memory_order_release); + } + + SeqLockGraph(const SeqLockGraph&) = delete; + SeqLockGraph& operator=(const SeqLockGraph&) = delete; + + // Hand-written because ``n_nodes_`` is an atomic and so is not implicitly movable. + // Moving is only ever done before the graph is published to readers. + SeqLockGraph(SeqLockGraph&& other) noexcept + : max_degree_{other.max_degree_} + , stride_{other.stride_} + , segment_size_{other.segment_size_} + , n_nodes_{other.n_nodes_.load(std::memory_order_relaxed)} + , segments_{std::move(other.segments_)} + , seqlocks_{std::move(other.seqlocks_)} { + other.n_nodes_.store(0, std::memory_order_relaxed); + } + + SeqLockGraph& operator=(SeqLockGraph&& other) noexcept { + if (this != &other) { + max_degree_ = other.max_degree_; + stride_ = other.stride_; + segment_size_ = other.segment_size_; + n_nodes_.store( + other.n_nodes_.load(std::memory_order_relaxed), std::memory_order_relaxed + ); + segments_ = std::move(other.segments_); + seqlocks_ = std::move(other.seqlocks_); + other.n_nodes_.store(0, std::memory_order_relaxed); + } + return *this; + } + + ~SeqLockGraph() = default; + + ///// ImmutableMemoryGraph + + /// @brief Maximum out-degree any node can hold. + size_t max_degree() const noexcept { return max_degree_; } + + /// @brief Number of nodes currently in the graph. + size_t n_nodes() const noexcept { return n_nodes_.load(std::memory_order_acquire); } + + /// @brief Alias for ``n_nodes`` (used by ``svs::graphs::graphs_equal``). + size_t num_nodes() const noexcept { return n_nodes(); } + + /// + /// @brief Return node ``i``'s adjacency list using plain loads. + /// + /// Only safe when no writer can be concurrently mutating node ``i``. Use + /// ``get_node_atomic`` on the concurrent read path. + /// + const_reference get_node(Idx i) const noexcept { + const Idx* base = slot(i); + return const_reference{base + 1, static_cast(base[0])}; + } + + /// + /// @brief Return node ``i``'s adjacency list as an ``AtomicSpan``. + /// + /// Every element access is an atomic relaxed load, so the read is race-free even + /// while the single writer mutates node ``i``. The *contents* may still be a torn + /// mixture of pre- and post-write state; pair this with ``seqlock(i)`` to detect and + /// retry such reads. + /// + svs::AtomicSpan get_node_atomic(Idx i) const noexcept { + const Idx* base = slot(i); + return svs::AtomicSpan{base + 1, static_cast(load_(base[0]))}; + } + + /// @brief Out-degree of node ``i`` (atomic load). + size_t get_node_degree(Idx i) const noexcept { + return static_cast(load_(slot(i)[0])); + } + + /// @brief Prefetch node ``i``'s adjacency list. Performance hint only. + void prefetch_node(Idx i) const noexcept { + const Idx* base = slot(i); + for (size_t offset = 0; offset < stride_ * sizeof(Idx); offset += 64) { + __builtin_prefetch(reinterpret_cast(base) + offset); + } + } + + /// @brief Access node ``i``'s sequence-lock counter. + const svs::SeqLockCounter& seqlock(size_t i) const noexcept { return seqlocks_[i]; } + + ///// MemoryGraph (single writer; caller serializes) + + /// + /// @brief Add the edge ``src -> dst``, returning ``src``'s out-degree afterwards. + /// + /// A no-op (returning the current degree) if the edge already exists, if + /// ``src == dst``, or if ``src``'s adjacency list is already full. + /// + size_t add_edge(Idx src, Idx dst) { + Idx* base = mutable_slot(src); + const size_t degree = static_cast(base[0]); + if (src == dst || degree >= max_degree_) { + return degree; + } + for (size_t i = 0; i < degree; ++i) { + if (base[1 + i] == dst) { + return degree; + } + } + auto seq = seqlocks_[src].begin_write(); + // Publish the neighbor before the degree that exposes it, so a reader can never + // see a degree covering a slot that has not been written. + store_(base[1 + degree], dst); + store_(base[0], static_cast(degree + 1)); + seqlocks_[src].end_write(seq); + return degree + 1; + } + + /// @brief Drop every edge out of node ``i``. + void clear_node(Idx i) { + Idx* base = mutable_slot(i); + auto seq = seqlocks_[i].begin_write(); + store_(base[0], Idx{0}); + seqlocks_[i].end_write(seq); + } + + /// + /// @brief Overwrite node ``src``'s adjacency list with ``neighbors``. + /// + /// The degree is dropped to zero first so that a reader which began before the write + /// only ever observes a validly-written *prefix* of the list. + /// + template void replace_node(Idx src, const R& neighbors) { + Idx* base = mutable_slot(src); + const size_t n = std::size(neighbors); + assert(n <= max_degree_); + + auto seq = seqlocks_[src].begin_write(); + store_(base[0], Idx{0}); + size_t k = 0; + for (auto id : neighbors) { + store_(base[1 + k], static_cast(id)); + ++k; + } + store_(base[0], static_cast(n)); + seqlocks_[src].end_write(seq); + } + + /// + /// @brief Resize the graph to ``new_size`` nodes. + /// + /// Growing is safe against concurrent readers: storage is allocated and the sequence + /// counters extended before the new node count is published. Shrinking may free + /// segments and therefore requires that readers have been drained. + /// + void unsafe_resize(size_t new_size) { + const size_t current = n_nodes_.load(std::memory_order_relaxed); + if (new_size == current) { + return; + } + if (new_size < current) { + n_nodes_.store(new_size, std::memory_order_release); + seqlocks_.resize(new_size); + return; + } + grow_to(new_size); + n_nodes_.store(new_size, std::memory_order_release); + } + + /// @brief Append a single node. + void add_node() { unsafe_resize(n_nodes() + 1); } + + /// @brief Number of nodes addressable without allocating another segment. + size_t capacity() const noexcept { return segments_.size() * segment_size_; } + + /// @brief Bytes held by the adjacency storage and sequence counters. + size_t bytes_reserved() const noexcept { + return capacity() * stride_ * sizeof(Idx) + + seqlocks_.capacity() * sizeof(svs::SeqLockCounter); + } + + private: + // Allocate segments and sequence counters so that every index < n is addressable. + // Does not publish the new node count. + void grow_to(size_t n) { + while (capacity() < n) { + segments_.push_back(Segment(segment_size_ * stride_)); + } + if (seqlocks_.size() < n) { + seqlocks_.resize(n); + } + } + + const Idx* slot(size_t i) const noexcept { + return segments_[i / segment_size_].data() + (i % segment_size_) * stride_; + } + Idx* mutable_slot(size_t i) noexcept { + return segments_[i / segment_size_].data() + (i % segment_size_) * stride_; + } + + // Relaxed atomics, not plain accesses. The sequence counter tells a reader whether what + // it read was *coherent*, but it does not make the reads themselves defined behaviour: + // a reader legitimately touches slots a writer is modifying, which is a data race + // unless the accesses are atomic. Relaxed is the weakest ordering that removes the + // race, and on x86 both of these compile to a plain MOV -- so this costs nothing but + // buys a well-defined program that ThreadSanitizer can actually verify. + // + // Define SVS_CONCURRENT_UNSAFE_PLAIN_GRAPH_ACCESS to drop the atomics. That build is + // *wrong* on purpose: it exists so the TSan targets can be shown to fail when they + // should, which is the only way a clean TSan run means anything. +#ifdef SVS_CONCURRENT_UNSAFE_PLAIN_GRAPH_ACCESS + static void store_(Idx& s, Idx v) noexcept { s = v; } + static Idx load_(const Idx& s) noexcept { return s; } +#else + static void store_(Idx& s, Idx v) noexcept { + std::atomic_ref(s).store(v, std::memory_order_relaxed); + } + static Idx load_(const Idx& s) noexcept { + return std::atomic_ref(const_cast(s)).load(std::memory_order_relaxed); + } +#endif + + size_t max_degree_{0}; + size_t stride_{1}; + size_t segment_size_{default_segment_size}; + std::atomic n_nodes_{0}; + // Grow-stable: appending a segment never relocates existing segments or the + // directory that addresses them. + svs::lib::SegmentedVector segments_{}; + svs::SeqLockArray seqlocks_{}; +}; + +} // namespace svs::concurrent diff --git a/include/svs/concurrent/graph_view.h b/include/svs/concurrent/graph_view.h new file mode 100644 index 000000000..8c70e8200 --- /dev/null +++ b/include/svs/concurrent/graph_view.h @@ -0,0 +1,113 @@ +/* + * 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 "svs/concurrent/graph.h" + +#include "svs/concepts/graph.h" +#include "svs/lib/spinlock.h" // svs::detail::pause + +#include +#include +#include + +namespace svs::concurrent { + +/// +/// @brief A read-only view of a ``SeqLockGraph`` that makes *unmodified* upstream +/// graph-traversal code safe against a concurrent writer. +/// +/// The trick is where the sequence lock lives. ``seqlock_greedy_search`` puts the whole +/// node expansion -- distance computations included -- inside the read section, which is +/// zero-copy but means every traversal routine has to be rewritten around the retry loop. +/// This class instead pushes the read section *down into ``get_node``*: it copies the +/// adjacency list into a scratch buffer, retrying until the sequence counter certifies the +/// copy, and hands back a span over that stable buffer. The caller sees an ordinary, +/// immutable adjacency list and needs to know nothing about concurrency. +/// +/// So this type satisfies ``svs::graphs::ImmutableMemoryGraph`` and can be handed to +/// upstream's ``greedy_search``, ``BatchIterator``, and anything else that only *reads* a +/// graph, with no changes to any of them. That is how the prototype supports range/batch +/// queries without duplicating upstream's 340-line ``iterator.h``. +/// +/// Cost: one copy of up to ``max_degree`` indices per node visited (128 bytes at degree 32 +/// with ``uint32_t`` IDs), against the ~32 full distance computations that visit provokes. +/// Correctness cost: none -- the copy is certified by the same sequence counter. +/// +/// **Not thread-safe.** One instance per searching thread; the scratch buffer is per-view. +/// This is the same contract as a search buffer, and it is why views are created inside the +/// search call rather than stored on the index. +/// +template class SeqLockGraphView { + public: + using graph_type = SeqLockGraph; + using index_type = Idx; + // Both alias the same const span: this view is read-only, so there is no mutable + // reference to hand out. `reference` exists only because the concept requires the name. + using reference = std::span; + using const_reference = std::span; + + explicit SeqLockGraphView(const graph_type& graph) + : graph_{&graph} + , scratch_(graph.max_degree()) {} + + size_t max_degree() const { return graph_->max_degree(); } + size_t n_nodes() const { return graph_->n_nodes(); } + void prefetch_node(Idx i) const { graph_->prefetch_node(i); } + + /// @brief A certified-consistent snapshot of node ``i``'s adjacency list. + /// + /// The returned span is valid until the next call to ``get_node`` on this view. + const_reference get_node(Idx i) const { + const auto& seqlock = graph_->seqlock(i); + for (;;) { + auto maybe_seq = seqlock.read_begin(); + if (!maybe_seq) { + svs::detail::pause(); + continue; + } + auto neighbors = graph_->get_node_atomic(i); + // The degree is a single aligned store of a value never exceeding max_degree, + // so it cannot tear into something larger -- but clamping is free and keeps a + // future writer-side bug from turning into a buffer overrun here. + const size_t degree = std::min(neighbors.size(), scratch_.size()); + for (size_t j = 0; j < degree; ++j) { + scratch_[j] = neighbors[j]; + } + if (seqlock.read_validate(*maybe_seq)) { + return const_reference{scratch_.data(), degree}; + } + svs::detail::pause(); + } + } + + /// @brief Degree of node ``i``. + /// + /// Callers must treat this as a hint that may already be stale by the time it returns: + /// only the degree bundled with the snapshot from ``get_node`` is certified. Upstream + /// uses ``get_node_degree`` only for reporting and capacity checks, never to bound a + /// loop over memory, so a stale value is harmless there. + size_t get_node_degree(Idx i) const { return graph_->get_node_degree(i); } + + private: + const graph_type* graph_; + mutable std::vector scratch_; +}; + +static_assert(svs::graphs::ImmutableMemoryGraph>); + +} // namespace svs::concurrent diff --git a/include/svs/concurrent/greedy_search.h b/include/svs/concurrent/greedy_search.h new file mode 100644 index 000000000..f0de583b0 --- /dev/null +++ b/include/svs/concurrent/greedy_search.h @@ -0,0 +1,220 @@ +/* + * 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 "svs/concurrent/graph.h" + +#include "svs/concepts/data.h" +#include "svs/concepts/distance.h" + +// ``distance::compute`` is called with a *qualified* name from both this header and +// upstream's ``greedy_search.h``, so the concrete overloads must be declared before either +// definition is parsed -- qualified lookup in a template happens at the point of +// definition, not instantiation. Upstream's ``dynamic_index.h`` gets this for free because +// its alphabetically-sorted include list puts ``core/distance.h`` above +// ``index/vamana/greedy_search.h``; being explicit here removes the ordering hazard. +#include "svs/core/distance.h" + +#include "svs/index/vamana/greedy_search.h" +#include "svs/lib/spinlock.h" // svs::detail::pause + +#include +#include + +namespace svs::concurrent { + +// Reuse upstream's trackers, prefetch parameters, initializer and neighbor builder +// verbatim -- only the node-expansion loop needs to change. +using svs::index::vamana::EntryPointInitializer; +using svs::index::vamana::GreedySearchPrefetchParameters; +using svs::index::vamana::GreedySearchTracker; +using svs::index::vamana::NeighborBuilder; +using svs::index::vamana::NullTracker; + +/// +/// @brief Greedy graph search that tolerates a concurrent writer. +/// +/// Behaviourally identical to ``svs::index::vamana::greedy_search``; the only difference +/// is that each node expansion is wrapped in a sequence-lock read section: +/// +/// 1. ``read_begin`` -- bail out and spin if a write to this node is in flight. +/// 2. Read the adjacency list through an ``AtomicSpan`` (relaxed atomic loads, which +/// compile to plain ``MOV`` on x86) and expand its neighbors. +/// 3. ``read_validate`` -- if the writer touched this node meanwhile, retry the +/// expansion. +/// +/// Retrying is safe rather than merely tolerable: a rejected read can only have inserted +/// neighbors with *valid* IDs and *correctly computed* distances into the search buffer +/// (the graph never publishes a degree covering an unwritten slot), and +/// ``search_buffer.insert`` deduplicates by ID. So a retry can add redundant work but +/// cannot corrupt the result. +/// +/// This lives in ``svs::concurrent`` and is a separate routine from upstream's, so +/// ``svs/index/vamana/greedy_search.h`` -- shared with the *static* Vamana index -- is +/// left completely untouched and pays nothing for this feature. +/// +template < + std::unsigned_integral Idx, + svs::data::ImmutableMemoryDataset Dataset, + svs::data::AccessorFor Accessor, + typename QueryType, + svs::distance::Distance Dist, + typename Buffer, + typename Initializer, + typename Builder, + GreedySearchTracker Tracker> +void seqlock_greedy_search( + const SeqLockGraph& graph, + const Dataset& dataset, + Accessor& accessor, + const QueryType& query, + Dist& distance_function, + Buffer& search_buffer, + const Initializer& initializer, + const Builder& builder, + Tracker& search_tracker, + GreedySearchPrefetchParameters prefetch_parameters = {}, + const svs::lib::DefaultPredicate& cancel = svs::lib::Returns(svs::lib::Const()) +) { + using I = Idx; + + // Fix the query if needed by the distance function. + svs::distance::maybe_fix_argument(distance_function, query); + + // Initialize the search buffer. + { + auto computer = [&](std::integral auto id) { + return svs::distance::compute(distance_function, query, accessor(dataset, id)); + }; + initializer(search_buffer, computer, graph, builder, search_tracker); + } + + // Main search routine. + while (!search_buffer.done()) { + // Check if request to cancel the search + if (cancel()) { + return; + } + // Get the next unvisited vertex. + // + // Copy it out by value rather than holding the reference: unlike upstream, the + // expansion below can run more than once, and the ``search_buffer.insert`` calls + // it performs may reorder the buffer and invalidate a reference into it. + const auto tracked = svs::Neighbor{search_buffer.next()}; + const auto node_id = tracked.id(); + + const auto& node_seqlock = graph.seqlock(node_id); + + for (;;) { // Sequence-lock read section. + auto maybe_seq = node_seqlock.read_begin(); + if (!maybe_seq) { + // A write to this node is in flight; wait for it to land. + svs::detail::pause(); + continue; + } + + // Get the adjacency list for this vertex and prepare prefetching logic. + auto neighbors = graph.get_node_atomic(node_id); + const size_t num_neighbors = neighbors.size(); + search_tracker.visited(tracked, num_neighbors); + + auto prefetcher = svs::lib::make_prefetcher( + svs::lib::PrefetchParameters{ + prefetch_parameters.lookahead, prefetch_parameters.step}, + num_neighbors, + [&](size_t i) { accessor.prefetch(dataset, neighbors[i]); }, + [&](size_t i) { + // Perform the visited set enabled check just once. + if (search_buffer.visited_set_enabled()) { + // Prefetch next bucket so it's (hopefully) in the cache when we + // next consult the visited filter. + if (i + 1 < num_neighbors) { + search_buffer.unsafe_prefetch_visited(neighbors[i + 1]); + } + return !search_buffer.unsafe_is_visited(neighbors[i]); + } + + // Otherwise, always prefetch the next data item. + return true; + } + ); + + ///// Neighbor expansion. + prefetcher(); + for (auto id : neighbors) { + if (search_buffer.emplace_visited(id)) { + continue; + } + + // Run the prefetcher. + prefetcher(); + + // Compute distance and update search buffer. + auto dist = + svs::distance::compute(distance_function, query, accessor(dataset, id)); + search_buffer.insert(builder(id, dist)); + } + + if (node_seqlock.read_validate(*maybe_seq)) { + break; // Consistent read -- move on to the next node. + } + // The writer modified this node mid-expansion. Anything we inserted is valid + // but possibly stale, so re-expand to pick up the current adjacency list. + svs::detail::pause(); + } + } +} + +/// @brief Overload supplying a default (null) search tracker. +template < + std::unsigned_integral Idx, + svs::data::ImmutableMemoryDataset Dataset, + svs::data::AccessorFor Accessor, + typename QueryType, + svs::distance::Distance Dist, + typename Buffer, + typename Initializer, + typename Builder = NeighborBuilder> +void seqlock_greedy_search( + const SeqLockGraph& graph, + const Dataset& dataset, + Accessor& accessor, + QueryType query, + Dist& distance_function, + Buffer& search_buffer, + const Initializer& initializer, + const Builder& builder = NeighborBuilder(), + GreedySearchPrefetchParameters prefetch_parameters = {}, + const svs::lib::DefaultPredicate& cancel = svs::lib::Returns(svs::lib::Const()) +) { + auto null_tracker = NullTracker{}; + seqlock_greedy_search( + graph, + dataset, + accessor, + query, + distance_function, + search_buffer, + initializer, + builder, + null_tracker, + prefetch_parameters, + cancel + ); +} + +} // namespace svs::concurrent diff --git a/include/svs/concurrent/mutable_vamana_index.h b/include/svs/concurrent/mutable_vamana_index.h new file mode 100644 index 000000000..a7ad2309b --- /dev/null +++ b/include/svs/concurrent/mutable_vamana_index.h @@ -0,0 +1,790 @@ +/* + * 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 "svs/concurrent/graph.h" +#include "svs/concurrent/graph_view.h" +#include "svs/concurrent/greedy_search.h" +#include "svs/lib/concurrency/writer_priority_mutex.h" + +#include "svs/core/translation.h" +#include "svs/index/vamana/consolidate.h" +#include "svs/index/vamana/dynamic_index.h" // SlotMetadata +#include "svs/index/vamana/extensions.h" +#include "svs/index/vamana/index.h" +#include "svs/index/vamana/iterator.h" +#include "svs/index/vamana/search_params.h" +#include "svs/index/vamana/vamana_build.h" +#include "svs/lib/concurrency/readwrite_protected.h" +#include "svs/lib/threads.h" + +#include +#include +#include +#include +#include + +namespace svs::concurrent { + +using svs::index::vamana::SlotMetadata; + +/// +/// @brief Neighbor builder that filters deleted slots using atomic status reads. +/// +/// Equivalent to ``svs::index::vamana::ValidBuilder`` except that the status byte is read +/// with an atomic load, because a concurrent writer may be flipping it from ``Valid`` to +/// ``Deleted`` while a search is in flight. +/// +class AtomicValidBuilder { + public: + explicit AtomicValidBuilder(const uint8_t* status) + : status_{status} {} + + template + svs::PredicatedSearchNeighbor operator()(I i, float distance) const { + auto raw = std::atomic_ref(const_cast(status_[i])) + .load(std::memory_order_relaxed); + bool invalid = static_cast(raw) == SlotMetadata::Deleted; + return svs::PredicatedSearchNeighbor(i, distance, !invalid); + } + + private: + const uint8_t* status_; +}; + +/// +/// @brief A dynamic Vamana index whose searches do not block on insertions or deletions. +/// +/// This is a *new index type*, not a modification of +/// ``svs::index::vamana::MutableVamanaIndex``. It is **purely additive**: it lives in its +/// own namespace and modifies no existing header. ``VamanaBuilder``, ``prune``, +/// ``consolidate``, the search-extension hooks, the search buffers and the graph concepts +/// are all consumed unmodified. The static Vamana index and the existing mutable index are +/// therefore completely unaffected by this feature -- they pay no extra atomic load, no +/// extra indirection and no extra byte per node. +/// +/// ## Synchronization model +/// +/// | Operation | Search blocked? | Mechanism | +/// |---|---|---| +/// | ``search`` | -- | shared lock on ``structure_mutex_`` + per-node seqlock reads | +/// | ``add_points`` (no growth) | no | ``writer_mutex_`` only | +/// | ``add_points`` (growth) | briefly | exclusive ``structure_mutex_`` around the resize | +/// | ``delete_entries`` | no | ``writer_mutex_`` + atomic status store | +/// | ``consolidate`` / ``compact`` | yes | exclusive ``structure_mutex_`` | +/// +/// Three locks, with a strict acquisition order of ``writer_mutex_`` -> +/// ``structure_mutex_`` -> ``translator_mutex_``: +/// +/// * ``writer_mutex_`` serializes whole mutating *operations* against each other. Note +/// this is coarser than "one writer thread": ``VamanaBuilder`` still mutates the graph +/// with the full thread pool from inside ``add_points``. That is safe because the +/// builder already guarantees *at most one writer per node* (it holds per-vertex +/// ``SpinLock``s while adding reverse edges, and partitions nodes disjointly across +/// threads elsewhere), which is exactly the precondition ``SeqLockCounter`` requires. +/// * ``structure_mutex_`` is held *shared* for the duration of a search and *exclusive* +/// only when the container structure changes under readers' feet. The graph is +/// grow-stable so it needs no such protection, but ``svs::data::BlockedData`` holds its +/// block descriptors in a ``std::vector`` that reallocates on growth, so capacity +/// growth and compaction must exclude readers. Steady-state edge rewiring -- the bulk +/// of insertion work -- does not. +/// * ``translator_mutex_`` guards the external<->internal ID maps, whose hash tables +/// rehash on insert. Searches take it shared only to convert their final k results. +/// +/// The alternative to the exclusive-on-growth window is to make the dataset itself +/// grow-stable, which means changing ``svs::data::SimpleData<..., Blocked<...>>`` and so +/// adding an indirection to ``get_datum`` for *every* index type in the library. This +/// prototype deliberately keeps that cost out of the shared data path. +/// +template +class MutableVamanaIndex { + public: + // Traits + static constexpr bool supports_insertions = true; + static constexpr bool supports_deletions = true; + static constexpr bool supports_saving = false; // Not implemented in the prototype. + static constexpr bool needs_id_translation = true; + + /// @brief Placeholder written into a result slot whose vector was deleted mid-search. + /// + /// Concurrent deletion is legal here, so a search can legitimately select a neighbor + /// that is retired before its ID is translated back. Results carrying this value must + /// be dropped by the caller. ``svs::index::vamana::MutableVamanaIndex`` has no analogue + /// because it forbids concurrent writers outright. + static constexpr size_t invalid_external_id = std::numeric_limits::max(); + + // Type aliases -- deliberately mirroring svs::index::vamana::MutableVamanaIndex so that + // call sites can swap between the two. + using Idx = Index; + using internal_id_type = Idx; + using external_id_type = size_t; + using value_type = typename Data::value_type; + using const_value_type = typename Data::const_value_type; + static constexpr size_t extent = Data::extent; + + using distance_type = Dist; + using search_buffer_type = + svs::index::vamana::MutableBuffer>; + + using graph_type = SeqLockGraph; + using data_type = Data; + using entry_point_type = std::vector; + using search_parameters_type = svs::index::vamana::VamanaSearchParameters; + using inner_scratch_type = svs::tag_t< + svs::index::vamana::extensions::single_search_setup>::result_t; + using scratchspace_type = + svs::index::vamana::SearchScratchspace; + + /// + /// @brief Construct from a pre-built graph and dataset. + /// + /// Signature matches ``svs::index::vamana::MutableVamanaIndex``'s corresponding + /// constructor so that integrators (e.g. VecSim) can select between the two with a type + /// alias. + /// + template + MutableVamanaIndex( + graph_type graph, + Data data, + Idx entry_point, + Dist distance_function, + const ExternalIds& external_ids, + ThreadPoolProto threadpool_proto, + svs::logging::logger_ptr logger = svs::logging::get() + ) + : graph_{std::move(graph)} + , data_{std::move(data)} + , entry_point_{entry_point} + , status_(data_.size(), static_cast(SlotMetadata::Valid)) + , first_empty_{data_.size()} + , translator_{} + , distance_{std::move(distance_function)} + , threadpool_{svs::threads::as_threadpool(std::move(threadpool_proto))} + , search_parameters_{svs::index::vamana::construct_default_search_parameters(data_)} + , construction_window_size_{2 * graph_.max_degree()} + , max_candidates_{750} + , prune_to_{graph_.max_degree()} + , logger_{std::move(logger)} { + translator_.insert(external_ids, svs::threads::UnitRange(0, data_.size())); + } + + ///// Accessors + + svs::logging::logger_ptr get_logger() const { return logger_; } + size_t dimensions() const { return data_.dimensions(); } + const Data& view_data() const { return data_; } + const graph_type& view_graph() const { return graph_; } + size_t max_degree() const { return graph_.max_degree(); } + + float get_alpha() const { return alpha_; } + void set_alpha(float alpha) { alpha_ = alpha; } + size_t get_construction_window_size() const { return construction_window_size_; } + void set_construction_window_size(size_t s) { construction_window_size_ = s; } + size_t get_max_candidates() const { return max_candidates_; } + void set_max_candidates(size_t n) { max_candidates_ = n; } + size_t get_prune_to() const { return prune_to_; } + void set_prune_to(size_t n) { prune_to_ = n; } + bool get_full_search_history() const { return use_full_search_history_; } + void set_full_search_history(bool b) { use_full_search_history_ = b; } + + search_parameters_type get_search_parameters() const { + return search_parameters_.get(); + } + void set_search_parameters(const search_parameters_type& sp) { + search_parameters_.set(sp); + } + + void reset_performance_parameters() { + auto sp = get_search_parameters(); + auto pp = svs::index::vamana::extensions::estimate_prefetch_parameters(data_); + sp.prefetch_lookahead_ = pp.lookahead; + sp.prefetch_step_ = pp.step; + set_search_parameters(sp); + } + + size_t get_num_threads() const { return threadpool_.size(); } + void set_threadpool(svs::threads::ThreadPoolHandle pool) { + threadpool_ = std::move(pool); + } + + Dist distance_function() const { return svs::threads::shallow_copy(distance_); } + + ///// ID translation + // + // All of these take the translator lock shared: a concurrent ``add_points`` may be + // rehashing the underlying maps. + + Idx translate_external_id(size_t e) const { + std::shared_lock lock{translator_mutex_}; + return translator_.get_internal(e); + } + size_t translate_internal_id(Idx i) const { + std::shared_lock lock{translator_mutex_}; + return translator_.get_external(i); + } + /// @brief As ``translate_internal_id``, but yields ``invalid_external_id`` instead of + /// throwing if the slot was retired by a concurrent ``delete_entries``. + size_t try_translate_internal_id(Idx i) const { + std::shared_lock lock{translator_mutex_}; + return translator_.has_internal(i) ? translator_.get_external(i) + : invalid_external_id; + } + bool has_id(size_t e) const { + std::shared_lock lock{translator_mutex_}; + return translator_.has_external(e); + } + /// @brief Number of valid (non-deleted) entries. + size_t size() const { + std::shared_lock lock{translator_mutex_}; + return translator_.size(); + } + + template void on_ids(F&& f) const { + std::shared_lock lock{translator_mutex_}; + for (auto pair : translator_) { + f(pair.first); + } + } + + std::vector external_ids() const { + std::vector ids{}; + on_ids([&ids](size_t id) { ids.push_back(id); }); + return ids; + } + + auto get_datum(size_t e) const { return data_.get_datum(translate_external_id(e)); } + + bool is_deleted(size_t i) const { return load_status(i) != SlotMetadata::Valid; } + + /// @brief Bytes reserved by the graph, data and dynamic metadata. + size_t get_memory_usage() const { + return graph_.bytes_reserved() + + data_.size() * data_.dimensions() * sizeof(value_type) + + status_.capacity() * sizeof(uint8_t); + } + + ///// Search + + AtomicValidBuilder internal_search_builder() const { + return AtomicValidBuilder{status_.data()}; + } + + scratchspace_type scratchspace(const search_parameters_type& sp) const { + return scratchspace_type{ + search_buffer_type{sp.buffer_config_, svs::distance::comparator(distance_)}, + svs::index::vamana::extensions::single_search_setup(data_, distance_), + svs::index::vamana::GreedySearchPrefetchParameters{ + sp.prefetch_lookahead_, sp.prefetch_step_}}; + } + scratchspace_type scratchspace() const { return scratchspace(get_search_parameters()); } + + /// + /// @brief The search closure handed to the extension hooks. + /// + /// Identical to ``svs::index::vamana::MutableVamanaIndex::greedy_search_closure`` apart + /// from calling + /// ``svs::concurrent::seqlock_greedy_search`` instead of + /// ``svs::index::vamana::greedy_search``. + /// + auto greedy_search_closure( + svs::index::vamana::GreedySearchPrefetchParameters prefetch_parameters, + const svs::lib::DefaultPredicate& cancel = + svs::lib::Returns(svs::lib::Const()) + ) const { + return [&, prefetch_parameters]( + const auto& query, auto& accessor, auto& distance, auto& buffer + ) { + seqlock_greedy_search( + graph_, + data_, + accessor, + query, + distance, + buffer, + EntryPointInitializer{svs::lib::as_const_span(entry_point_)}, + internal_search_builder(), + prefetch_parameters, + cancel + ); + buffer.cleanup(); + }; + } + + /// @brief Single-query search into a caller-provided scratch space. + template + void search( + const Query& query, + scratchspace_type& scratch, + const svs::lib::DefaultPredicate& cancel = + svs::lib::Returns(svs::lib::Const()) + ) const { + // Shared: excludes capacity growth and GC, admits concurrent searches and + // concurrent steady-state insertions/deletions. + std::shared_lock structure_lock{structure_mutex_}; + svs::index::vamana::extensions::single_search( + data_, + scratch.buffer, + scratch.scratch, + query, + greedy_search_closure(scratch.prefetch_parameters, cancel), + *this + ); + } + + /// @brief Batch search over ``queries``, writing external IDs into ``results``. + template + void search( + svs::QueryResultView results, + const Queries& queries, + const search_parameters_type& sp, + const svs::lib::DefaultPredicate& cancel = + svs::lib::Returns(svs::lib::Const()) + ) { + { + std::shared_lock structure_lock{structure_mutex_}; + svs::threads::parallel_for( + threadpool_, + svs::threads::StaticPartition{queries.size()}, + [&](const auto is, uint64_t /*tid*/) { + size_t num_neighbors = results.n_neighbors(); + auto buffer = search_buffer_type{ + sp.buffer_config_, svs::distance::comparator(distance_)}; + auto prefetch_parameters = + svs::index::vamana::GreedySearchPrefetchParameters{ + sp.prefetch_lookahead_, sp.prefetch_step_}; + if (buffer.target_capacity() < num_neighbors) { + buffer.change_maxsize(num_neighbors); + } + auto scratch = + svs::index::vamana::extensions::per_thread_batch_search_setup( + data_, distance_ + ); + svs::index::vamana::extensions::per_thread_batch_search( + data_, + buffer, + scratch, + queries, + results, + svs::threads::UnitRange{is}, + greedy_search_closure(prefetch_parameters, cancel), + *this, + cancel + ); + } + ); + } + + if (cancel()) { + return; + } + translate_to_external(results.indices()); + } + + /// @brief Distance between the vector stored for ``external_id`` and ``query``. + template + double get_distance(const ExternalId& external_id, const Query& query) const { + if (!has_id(external_id)) { + throw ANNEXCEPTION( + "ID {} is out of bounds for index of size {}!", external_id, size() + ); + } + if (query.size() != dimensions()) { + throw ANNEXCEPTION( + "Incompatible dimensions. Query has {} while the index expects {}.", + query.size(), + dimensions() + ); + } + std::shared_lock structure_lock{structure_mutex_}; + auto internal_id = translate_external_id(external_id); + return svs::index::vamana::extensions::get_distance_ext( + data_, distance_, internal_id, query + ); + } + + ///// Mutation + + /// + /// @brief Insert ``points`` under the given external IDs. + /// + /// Searches run concurrently throughout, except for a brief exclusive window if the + /// dataset must grow to make room. + /// + template + std::vector add_points( + const Points& points, const ExternalIds& external_ids, bool reuse_empty = false + ) { + const size_t num_points = points.size(); + if (num_points != external_ids.size()) { + throw ANNEXCEPTION( + "Number of points ({}) not equal to the number of external ids ({})!", + num_points, + external_ids.size() + ); + } + + std::lock_guard writer_lock{writer_mutex_}; + + // Gather reusable slots. Only this thread mutates ``status_``'s structure, so a + // plain scan is fine. + std::vector slots{}; + slots.reserve(num_points); + for (size_t s = reuse_empty ? 0 : first_empty_, smax = status_.size(); + s < smax && slots.size() < num_points; + ++s) { + if (load_status(s) == SlotMetadata::Empty) { + slots.push_back(s); + } + } + + if (slots.size() < num_points) { + const size_t needed = num_points - slots.size(); + const size_t current_size = data_.size(); + const size_t new_size = current_size + needed; + + // The only place searches are excluded during insertion: ``BlockedData``'s + // block-descriptor vector may reallocate. The graph and the status array are + // resized here too so the window covers all of them at once. + { + std::unique_lock structure_lock{structure_mutex_}; + data_.resize(new_size); + graph_.unsafe_resize(new_size); + status_.resize(new_size, static_cast(SlotMetadata::Empty)); + } + + for (size_t s = current_size; s < new_size; ++s) { + slots.push_back(s); + } + } + assert(slots.size() == num_points); + + // Publish the ID mapping before any edges point at the new slots, so a searcher + // that reaches a new node can always translate it back to an external ID. + { + std::unique_lock translator_lock{translator_mutex_}; + translator_.insert(external_ids, slots); + } + + // Write the vectors before wiring the nodes in: no in-edges exist yet, so the + // data is fully visible by the time a searcher can reach these slots. + svs::threads::parallel_for( + threadpool_, + svs::threads::StaticPartition{slots.size()}, + [&](auto is, uint64_t /*tid*/) { + for (auto i : is) { + data_.set_datum(slots[i], points.get_datum(i)); + } + } + ); + for (auto slot : slots) { + graph_.clear_node(static_cast(slot)); + } + + // Wire up the new nodes using the *unmodified* upstream builder. It mutates the + // graph with the whole thread pool, but never two threads on one node, which is + // all the per-node seqlocks require. + auto parameters = svs::index::vamana::VamanaBuildParameters{ + alpha_, + graph_.max_degree(), + construction_window_size_, + max_candidates_, + prune_to_, + use_full_search_history_}; + + auto sp = get_search_parameters(); + auto prefetch_parameters = svs::index::vamana::GreedySearchPrefetchParameters{ + sp.prefetch_lookahead_, sp.prefetch_step_}; + + auto builder = svs::index::vamana::VamanaBuilder{ + graph_, + data_, + distance_, + parameters, + threadpool_, + prefetch_parameters, + logger_, + svs::logging::Level::Trace}; + builder.construct( + alpha_, entry_point(), slots, svs::logging::Level::Trace, logger_ + ); + + for (auto slot : slots) { + store_status(slot, SlotMetadata::Valid); + } + if (!slots.empty()) { + first_empty_ = std::max(first_empty_, slots.back() + 1); + } + return slots; + } + + /// + /// @brief Soft-delete the given external IDs. + /// + /// Never blocks searches: flipping a status byte is a single atomic store, and + /// in-flight searches simply stop returning the affected slots. + /// + template size_t delete_entries(const T& ids) { + std::lock_guard writer_lock{writer_mutex_}; + { + std::shared_lock translator_lock{translator_mutex_}; + translator_.check_external_exist(ids.begin(), ids.end()); + } + for (auto i : ids) { + Idx internal; + { + std::shared_lock translator_lock{translator_mutex_}; + internal = translator_.get_internal(i); + } + store_status(internal, SlotMetadata::Deleted); + } + { + std::unique_lock translator_lock{translator_mutex_}; + translator_.delete_external(ids); + } + return ids.size(); + } + + /// + /// @brief Remove deleted entries from the graph's adjacency lists. + /// + /// Stop-the-world: takes ``structure_mutex_`` exclusively. Reuses the *unmodified* + /// upstream ``svs::index::vamana::consolidate``, which is generic over the graph type + /// -- possible only because this prototype left the graph concepts alone. + /// + void consolidate() { + std::lock_guard writer_lock{writer_mutex_}; + std::unique_lock structure_lock{structure_mutex_}; + + auto check_is_deleted = [&](size_t i) { return this->is_deleted(i); }; + std::function valid = [&](size_t i) { return !this->is_deleted(i); }; + + // Replace the entry point if it was deleted. + if (load_status(entry_point_[0]) == SlotMetadata::Deleted) { + auto new_entry_point = svs::index::vamana::extensions::compute_entry_point( + data_, threadpool_, valid + ); + entry_point_[0] = new_entry_point; + } + + svs::index::vamana::consolidate( + graph_, + data_, + threadpool_, + prune_to_, + max_candidates_, + alpha_, + distance_, + check_is_deleted + ); + + for (size_t i = 0, imax = status_.size(); i < imax; ++i) { + if (load_status(i) == SlotMetadata::Deleted) { + store_status(i, SlotMetadata::Empty); + } + } + } + + /// + /// @brief Squeeze out empty slots so IDs are dense again. + /// + /// Stop-the-world. Builds a fresh graph rather than remapping in place -- simpler and + /// safe, at the cost of transiently holding two graphs. + /// + void compact(size_t batch_size = 1'000'000) { + std::lock_guard writer_lock{writer_mutex_}; + std::unique_lock structure_lock{structure_mutex_}; + + // new_to_old[new_id] == old_id + std::vector new_to_old{}; + for (size_t i = 0, imax = status_.size(); i < imax; ++i) { + if (load_status(i) != SlotMetadata::Empty) { + new_to_old.push_back(static_cast(i)); + } + } + const size_t new_size = new_to_old.size(); + if (new_size == status_.size()) { + return; // Already dense. + } + + auto old_to_new = tsl::robin_map{}; + for (Idx new_id = 0; new_id < static_cast(new_size); ++new_id) { + old_to_new.insert({new_to_old[new_id], new_id}); + } + + // Remap adjacency into a fresh graph. Edges into now-empty slots are dropped; + // consolidate() should have removed them already, but be defensive. + auto compacted = graph_type{new_size, graph_.max_degree()}; + std::vector buffer{}; + for (Idx new_id = 0; new_id < static_cast(new_size); ++new_id) { + buffer.clear(); + for (auto old_neighbor : graph_.get_node(new_to_old[new_id])) { + auto found = old_to_new.find(old_neighbor); + if (found != old_to_new.end()) { + buffer.push_back(found->second); + } + } + compacted.replace_node(new_id, buffer); + } + graph_ = std::move(compacted); + + data_.compact(svs::lib::as_const_span(new_to_old), threadpool_, batch_size); + data_.resize(new_size); + + // Remap metadata and the translator. + { + std::unique_lock translator_lock{translator_mutex_}; + std::vector new_status(new_size); + for (size_t new_id = 0; new_id < new_size; ++new_id) { + auto old_id = new_to_old[new_id]; + new_status[new_id] = static_cast(load_status(old_id)); + if (static_cast(new_status[new_id]) == SlotMetadata::Valid && + old_id != static_cast(new_id)) { + translator_.remap_internal_id(old_id, static_cast(new_id)); + } + } + status_ = std::move(new_status); + } + first_empty_ = new_size; + + for (auto& ep : entry_point_) { + ep = old_to_new.at(ep); + } + } + + Idx entry_point() const { return entry_point_[0]; } + + /// + /// @brief Hand the raw graph, data, distance and entry points to ``f``. + /// + /// Same contract as + /// ``svs::index::vamana::MutableVamanaIndex::experimental_escape_hatch``, with one + /// difference that matters: the graph passed to ``f`` is a ``SeqLockGraphView``, not + /// the + /// ``SeqLockGraph`` itself. ``f`` therefore gets an ordinary immutable graph whose + /// adjacency lists are certified snapshots, and needs no awareness of concurrency -- + /// which is what lets upstream's ``BatchIterator`` traverse this index unmodified. + /// + /// The structure lock is held shared for the duration of the callback, so capacity + /// growth and compaction cannot run underneath it. + /// + template void experimental_escape_hatch(F&& f) const { + std::shared_lock structure_lock{structure_mutex_}; + auto view = SeqLockGraphView{graph_}; + std::invoke( + SVS_FWD(f), view, data_, distance_, svs::lib::as_const_span(entry_point_) + ); + } + + /// + /// @brief Batch (range-query) iterator over ``query``. + /// + /// This is upstream's ``BatchIterator``, verbatim, made safe by ``SeqLockGraphView`` + /// rather than by editing it. Note what it does *not* promise: the iterator caches a + /// search buffer across ``next()`` calls, so a vector inserted or deleted between two + /// batches may be missed or repeated. That is a semantic question about what a + /// long-lived cursor over a mutating index should mean, and it is not something a lock + /// discipline can answer -- it needs a product decision. What is guaranteed is memory + /// safety and per-batch internal consistency. + /// + template + auto make_batch_iterator( + std::span query, + size_t extra_search_buffer_capacity = svs::UNSIGNED_INTEGER_PLACEHOLDER + ) const { + return svs::index::vamana::BatchIterator( + *this, query, extra_search_buffer_capacity + ); + } + + static std::string name() { return "concurrent vamana index"; } + + private: + SlotMetadata load_status(size_t i) const { + return static_cast( + std::atomic_ref(const_cast(status_[i])) + .load(std::memory_order_relaxed) + ); + } + void store_status(size_t i, SlotMetadata m) { + std::atomic_ref(status_[i]) + .store(static_cast(m), std::memory_order_relaxed); + } + + // Signature mirrors ``svs::index::vamana::MutableVamanaIndex::translate_to_external``: + // the argument is the + // ``DenseArray`` view handed back by ``QueryResultView::indices()``, rewritten in + // place. + template + void translate_to_external(svs::DenseArray& ids) { + std::shared_lock translator_lock{translator_mutex_}; + svs::threads::parallel_for( + threadpool_, + svs::threads::StaticPartition{svs::getsize<0>(ids)}, + [&](auto is, uint64_t /*tid*/) { + for (auto i : is) { + for (size_t j = 0, jmax = svs::getsize<1>(ids); j < jmax; ++j) { + auto internal = svs::lib::narrow_cast(ids.at(i, j)); + // A concurrent ``delete_entries`` can retire a slot after the + // search selected it but before we get here, dropping its + // translator entry. Upstream can simply call ``get_external`` and + // let it throw, because no writer can run during its search. Here + // the honest answer is "this neighbor was deleted mid-flight", so + // report the sentinel and let the caller drop it. Throwing would + // turn a benign, expected race into a failed query. + ids.at(i, j) = translator_.has_internal(internal) + ? translator_.get_external(internal) + : invalid_external_id; + } + } + } + ); + } + + ///// Members + + graph_type graph_; + data_type data_; + entry_point_type entry_point_; + // Flat, so the search hot path pays one indexed load. Resized only under an exclusive + // ``structure_mutex_``; individual bytes are read/written atomically. + std::vector status_; + size_t first_empty_ = 0; + svs::IDTranslator translator_; + + distance_type distance_; + svs::threads::ThreadPoolHandle threadpool_; + svs::lib::ReadWriteProtected search_parameters_; + + // Configuration + size_t construction_window_size_; + size_t max_candidates_; + size_t prune_to_; + float alpha_ = 1.2; + bool use_full_search_history_ = true; + + svs::logging::logger_ptr logger_; + + ///// Synchronization. Acquire in this order. + mutable std::mutex writer_mutex_; + // Both must be writer-preferring, not ``std::shared_mutex``. Searches hold these + // shared for their whole duration, so under continuous query load a reader-preferring + // lock leaves the writer parked forever. See ``writer_priority_mutex.h``. + mutable svs::WriterPriorityMutex structure_mutex_; + mutable svs::WriterPriorityMutex translator_mutex_; +}; + +} // namespace svs::concurrent diff --git a/include/svs/lib/concurrency/atomic_span.h b/include/svs/lib/concurrency/atomic_span.h new file mode 100644 index 000000000..4d7d4ae50 --- /dev/null +++ b/include/svs/lib/concurrency/atomic_span.h @@ -0,0 +1,93 @@ +/* + * 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 +#include +#include + +namespace svs { + +/// +/// @brief A non-owning, zero-copy view over a contiguous range of ``T`` that performs +/// atomic loads on every element access. +/// +/// Each dereference uses ``std::atomic_ref::load(std::memory_order_relaxed)``. +/// On x86, this compiles to a plain MOV instruction — identical to non-atomic access. +/// +/// This type is designed to be used as a drop-in replacement for ``std::span`` +/// when concurrent reads and writes are possible, ensuring no undefined behavior +/// while maintaining zero-copy semantics. +/// +template class AtomicSpan { + public: + using value_type = std::remove_const_t; + + class iterator { + public: + using value_type = AtomicSpan::value_type; + using difference_type = std::ptrdiff_t; + using iterator_category = std::input_iterator_tag; + + explicit iterator(const T* p) + : ptr_(p) {} + + value_type operator*() const { + return std::atomic_ref(const_cast(*ptr_)) + .load(std::memory_order_relaxed); + } + + iterator& operator++() { + ++ptr_; + return *this; + } + + iterator operator++(int) { + auto tmp = *this; + ++ptr_; + return tmp; + } + + bool operator==(const iterator& other) const { return ptr_ == other.ptr_; } + bool operator!=(const iterator& other) const { return ptr_ != other.ptr_; } + + private: + const T* ptr_; + }; + + AtomicSpan(const T* data, size_t size) + : data_(data) + , size_(size) {} + + size_t size() const { return size_; } + bool empty() const { return size_ == 0; } + const T* data() const { return data_; } + + value_type operator[](size_t i) const { + return std::atomic_ref(const_cast(data_[i])) + .load(std::memory_order_relaxed); + } + + iterator begin() const { return iterator{data_}; } + iterator end() const { return iterator{data_ + size_}; } + + private: + const T* data_; + size_t size_; +}; + +} // namespace svs diff --git a/include/svs/lib/concurrency/seqlock.h b/include/svs/lib/concurrency/seqlock.h new file mode 100644 index 000000000..7fe3ea70e --- /dev/null +++ b/include/svs/lib/concurrency/seqlock.h @@ -0,0 +1,141 @@ +/* + * 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 "svs/lib/segmented_vector.h" + +#include +#include +#include +#include +#include + +namespace svs { + +/// +/// @brief Per-element sequence lock counter for reader-writer synchronization. +/// +/// Uses a uint8_t counter: odd values indicate a write in progress, even values indicate +/// a stable state. +/// +/// **Writer-writer serialization is the caller's responsibility.** Only one writer +/// at a time may call ``begin_write``/``end_write`` on a given counter. Use an external +/// lock (e.g., per-node ``SpinLock``) to serialize concurrent writers to the same element. +/// +class SeqLockCounter { + using counter_type = uint8_t; + + public: + SeqLockCounter() = default; + + SeqLockCounter(const SeqLockCounter& other) + : seq_(other.seq_.load(std::memory_order_relaxed)) {} + + SeqLockCounter& operator=(const SeqLockCounter& other) { + seq_.store(other.seq_.load(std::memory_order_relaxed), std::memory_order_relaxed); + return *this; + } + + SeqLockCounter(SeqLockCounter&& other) noexcept + : seq_(other.seq_.load(std::memory_order_relaxed)) {} + + SeqLockCounter& operator=(SeqLockCounter&& other) noexcept { + seq_.store(other.seq_.load(std::memory_order_relaxed), std::memory_order_relaxed); + return *this; + } + + /// + /// @brief Begin a write operation. Returns the pre-write sequence value. + /// + /// Increments the counter to an odd value, signaling to readers that a write is in + /// progress. The returned value must be passed to ``end_write``. + /// + counter_type begin_write() { + auto seq = seq_.load(std::memory_order_relaxed); + seq_.store(seq + 1, std::memory_order_relaxed); + std::atomic_thread_fence(std::memory_order_release); + return seq; + } + + /// + /// @brief End a write operation. + /// + /// @param seq The value returned by the corresponding ``begin_write`` call. + /// + /// Increments the counter to an even value, signaling that the write is complete + /// and data is consistent. + /// + void end_write(counter_type seq) { seq_.store(seq + 2, std::memory_order_release); } + + /// + /// @brief Begin a read operation. + /// + /// @returns The current sequence value if it is even (no write in progress), + /// or ``std::nullopt`` if a write is in progress. + /// + /// The returned value (if present) must be passed to ``read_validate`` after the + /// read is complete. + /// + std::optional read_begin() const { + auto seq = seq_.load(std::memory_order_acquire); + if (seq % 2 > 0) { + return std::nullopt; + } + return seq; + } + + /// + /// @brief Validate that no write occurred during the read. + /// + /// @param seq The value returned by ``read_begin``. + /// + /// @returns ``true`` if the data read between ``read_begin`` and ``read_validate`` + /// is consistent (no concurrent write occurred). + /// + bool read_validate(counter_type seq) const { + std::atomic_thread_fence(std::memory_order_acquire); + return seq_.load(std::memory_order_relaxed) == seq; + } + + private: + std::atomic seq_{0}; +}; + +/// +/// @brief Array of SeqLock counters, one per element (e.g., one per graph node). +/// +class SeqLockArray { + public: + SeqLockArray() = default; + explicit SeqLockArray(size_t n) + : counters_(n) {} + + SeqLockCounter& operator[](size_t i) { return counters_[i]; } + const SeqLockCounter& operator[](size_t i) const { return counters_[i]; } + + void resize(size_t n) { counters_.resize(n); } + size_t size() const { return counters_.size(); } + size_t capacity() const { return counters_.capacity(); } + + private: + // Grow-stable storage: appending counters never relocates existing ones, so a + // concurrent lock-free reader (greedy_search reading seq_counters_[i]) is safe + // against a writer's grow. See svs/lib/segmented_vector.h. + lib::SegmentedVector counters_; +}; + +} // namespace svs diff --git a/include/svs/lib/concurrency/writer_priority_mutex.h b/include/svs/lib/concurrency/writer_priority_mutex.h new file mode 100644 index 000000000..d98b65c40 --- /dev/null +++ b/include/svs/lib/concurrency/writer_priority_mutex.h @@ -0,0 +1,174 @@ +/* + * 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 "svs/lib/exception.h" + +#include +#include +#include + +#if defined(__GLIBC__) +#define SVS_WRITER_PRIORITY_MUTEX_PTHREAD 1 +#include +#else +#define SVS_WRITER_PRIORITY_MUTEX_PTHREAD 0 +#endif + +namespace svs { + +/// +/// @brief A shared mutex that will not starve its writer. +/// +/// Drop-in for ``std::shared_mutex`` (same member names, so ``std::shared_lock`` and +/// ``std::unique_lock`` work unchanged), but writer-preferring: once a thread is waiting in +/// ``lock()``, newly-arriving ``lock_shared()`` callers queue behind it instead of jumping +/// ahead. +/// +/// For ``svs::concurrent::MutableVamanaIndex`` this is a correctness requirement, not a +/// micro-optimization. That index holds this lock shared for the whole duration of every +/// search and exclusively for the brief windows where the dataset's capacity grows. Under +/// continuous query load from N threads there is essentially always at least one reader +/// inside the lock, so with libstdc++'s default reader-preferring ``std::shared_mutex`` +/// (``PTHREAD_RWLOCK_PREFER_READER_NP``) the writer never acquires it and insertion hangs +/// forever. That was observed, not theorized: 8 query threads spinning at 100% CPU with the +/// inserting thread parked on a futex indefinitely. +/// +/// It is worth being explicit about what this buys and what it costs. It buys liveness for +/// the writer. It costs a full query stall on every capacity growth, because the writer +/// must wait for all in-flight searches to drain. That stall is the price of leaving +/// ``svs::data::SimpleData<..., Blocked<...>>`` untouched -- see the discussion in +/// ``svs/concurrent/mutable_vamana_index.h``. +/// +/// Two implementations, selected on whether glibc's non-portable rwlock attribute is +/// available: +/// +/// * **glibc:** set ``PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP`` on the rwlock, so +/// ``lock_shared`` keeps the same atomic fast path ``std::shared_mutex`` has. +/// * **Everything else** (notably macOS, which has no ``pthread_rwlockattr_setkind_np``): +/// a condition-variable implementation. Correct, but it takes an uncontended +/// ``std::mutex`` on every ``lock_shared``, which is measurably more expensive on the +/// search path. +/// +class WriterPriorityMutex { + public: +#if SVS_WRITER_PRIORITY_MUTEX_PTHREAD + WriterPriorityMutex() { + pthread_rwlockattr_t attr; + if (pthread_rwlockattr_init(&attr) != 0) { + throw ANNEXCEPTION("Failed to initialize rwlock attributes!"); + } + // The whole point of this class. Note glibc spells the writer-preferring policy + // "nonrecursive": a thread holding the lock shared must not try to upgrade. + pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP); + const int rc = pthread_rwlock_init(&lock_, &attr); + pthread_rwlockattr_destroy(&attr); + if (rc != 0) { + throw ANNEXCEPTION("Failed to initialize rwlock!"); + } + } + + ~WriterPriorityMutex() { pthread_rwlock_destroy(&lock_); } +#else + WriterPriorityMutex() = default; + ~WriterPriorityMutex() = default; +#endif + + // Neither copyable nor movable, matching ``std::shared_mutex``. + WriterPriorityMutex(const WriterPriorityMutex&) = delete; + WriterPriorityMutex& operator=(const WriterPriorityMutex&) = delete; + WriterPriorityMutex(WriterPriorityMutex&&) = delete; + WriterPriorityMutex& operator=(WriterPriorityMutex&&) = delete; + +#if SVS_WRITER_PRIORITY_MUTEX_PTHREAD + + void lock() { pthread_rwlock_wrlock(&lock_); } + bool try_lock() { return pthread_rwlock_trywrlock(&lock_) == 0; } + void unlock() { pthread_rwlock_unlock(&lock_); } + + void lock_shared() { pthread_rwlock_rdlock(&lock_); } + bool try_lock_shared() { return pthread_rwlock_tryrdlock(&lock_) == 0; } + void unlock_shared() { pthread_rwlock_unlock(&lock_); } + + private: + pthread_rwlock_t lock_; + +#else + + void lock() { + std::unique_lock guard{mutex_}; + // Register before waiting: this is what makes arriving readers queue behind us. + ++waiting_writers_; + condition_.wait(guard, [this] { return !writer_active_ && readers_ == 0; }); + --waiting_writers_; + writer_active_ = true; + } + + bool try_lock() { + std::lock_guard guard{mutex_}; + if (writer_active_ || readers_ != 0) { + return false; + } + writer_active_ = true; + return true; + } + + void unlock() { + { + std::lock_guard guard{mutex_}; + writer_active_ = false; + } + condition_.notify_all(); + } + + void lock_shared() { + std::unique_lock guard{mutex_}; + condition_.wait(guard, [this] { return !writer_active_ && waiting_writers_ == 0; }); + ++readers_; + } + + bool try_lock_shared() { + std::lock_guard guard{mutex_}; + if (writer_active_ || waiting_writers_ != 0) { + return false; + } + ++readers_; + return true; + } + + void unlock_shared() { + bool last_reader = false; + { + std::lock_guard guard{mutex_}; + last_reader = (--readers_ == 0); + } + if (last_reader) { + condition_.notify_all(); + } + } + + private: + std::mutex mutex_{}; + std::condition_variable condition_{}; + std::size_t readers_{0}; + std::size_t waiting_writers_{0}; + bool writer_active_{false}; + +#endif +}; + +} // namespace svs diff --git a/include/svs/lib/segmented_vector.h b/include/svs/lib/segmented_vector.h new file mode 100644 index 000000000..7f642ae9c --- /dev/null +++ b/include/svs/lib/segmented_vector.h @@ -0,0 +1,330 @@ +/* + * 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 "svs/lib/boundscheck.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace svs::lib { + +/// +/// @brief An unbounded, grow-stable vector for single-writer/many-reader use. +/// +/// Two-level "lock-free dynamic array" (Dechev et al.). A fixed top-level directory of +/// ``kDirBuckets`` bucket pointers; directory bucket ``k`` is a single contiguous +/// heap array of ``kFirstBucket << k`` elements (the first bucket holds ``kFirstBucket``, +/// each subsequent bucket doubles). Grouping elements into chunks of ``kFirstBucket`` and +/// applying the power-of-two layout to the chunk index gives: ``q = i / kFirstBucket``, +/// ``bucket = floor(log2(q + 1))``, with the bucket's first global index +/// ``kFirstBucket * (2^bucket - 1)``. Sixty-four buckets address far more than any real +/// dataset, so there is no practical size cap. +/// +/// The directory array is a fixed member (never relocates), and each bucket array is +/// allocated once and never moved or reallocated. Therefore the address of any element +/// ``i < size()`` is stable for the lifetime of that element — a concurrent reader +/// indexing element ``i`` is unaffected by appends that grow the structure past ``i``. +/// +/// Concurrency contract: +/// * **One writer at a time.** ``resize`` / ``push_back`` / ``pop_back`` / ``shrink_to`` +/// (the only operations that change the structure) must be serialized by the caller +/// (e.g. under a mutex). +/// * **Many concurrent readers.** ``operator[]`` and ``size`` may run concurrently with a +/// writer's *grow*: a new bucket is allocated and its elements constructed, the bucket +/// pointer is published with a release store, and ``size_`` is bumped last (release). A +/// reader does an acquire load of ``size_`` then an acquire load of the bucket pointer, +/// so for any ``i < size()`` it observes, the bucket and element are fully published. +/// * **Shrink frees storage.** ``shrink_to`` destroys trailing elements and frees buckets +/// that lie entirely above the new size; a reader holding a reference to a freed element +/// would dangle, so the caller must drain readers (e.g. via an exclusive lock) first. +/// +/// This mirrors the std::vector subset used by the dynamic Vamana index: ``operator[]``, +/// ``at``, ``size``, ``empty``, ``capacity``, ``resize(n)``, ``resize(n, fill)``, +/// ``push_back``, ``pop_back``, ``shrink_to(n)``. +/// +template class SegmentedVector { + static constexpr std::size_t kDirBuckets = 64; + // Number of elements in the first directory bucket. Bucket ``k`` then holds + // ``kFirstBucket << k`` elements, so a small first bucket means many tiny + // allocations near the start while a large one front-loads capacity. + static constexpr std::size_t kFirstBucket = 1; + static_assert( + (kFirstBucket & (kFirstBucket - 1)) == 0, "kFirstBucket must be a power of two" + ); + + public: + using value_type = T; + using size_type = std::size_t; + using reference = T&; + using const_reference = const T&; + + SegmentedVector() = default; + explicit SegmentedVector(size_type n) { resize(n); } + SegmentedVector(size_type n, const T& fill) { resize(n, fill); } + + SegmentedVector(const SegmentedVector& other) { copy_from_(other); } + SegmentedVector& operator=(const SegmentedVector& other) { + if (this != &other) { + destroy_all_(); + copy_from_(other); + } + return *this; + } + + SegmentedVector(SegmentedVector&& other) noexcept { steal_from_(other); } + SegmentedVector& operator=(SegmentedVector&& other) noexcept { + if (this != &other) { + destroy_all_(); + steal_from_(other); + } + return *this; + } + + ~SegmentedVector() { destroy_all_(); } + + /// + /// @brief Access element ``i``. Precondition: ``i < size()``. + /// + /// Safe to call concurrently with a writer's grow ``resize``/``push_back``, provided + /// ``i`` was ``< size()`` as observed by the reader. + /// + const_reference operator[](size_type i) const noexcept { + auto [b, off] = locate_(i); + return dir_[b].load(std::memory_order_acquire)[off]; + } + reference operator[](size_type i) noexcept { + const auto& self = *this; + return const_cast(self[i]); + } + + /// @brief Bounds-checked access (used by ``svs::getindex`` when bounds checking is on). + reference at(size_type i) { + if (i >= size()) { + throw std::out_of_range("SegmentedVector::at index out of range"); + } + return (*this)[i]; + } + const_reference at(size_type i) const { + if (i >= size()) { + throw std::out_of_range("SegmentedVector::at index out of range"); + } + return (*this)[i]; + } + + size_type size() const noexcept { return size_.load(std::memory_order_acquire); } + bool empty() const noexcept { return size() == 0; } + + /// @brief Logical capacity: number of elements addressable without allocating a new + /// bucket. With ``m`` buckets this is ``kFirstBucket * (2^m - 1)``. + size_type capacity() const noexcept { return bucket_first_index_(allocated_buckets_); } + + /// @brief Grow or shrink the logical size. New elements are default-constructed. + /// Single-writer; concurrent readers safe on grow (see class contract). + void resize(size_type n) { resize_impl_(n, nullptr); } + + /// @brief Grow or shrink the logical size, filling new elements with ``fill``. + void resize(size_type n, const T& fill) { resize_impl_(n, &fill); } + + /// @brief Append one element, move-*constructing* it into the new slot. + /// + /// Grows logical size by one, allocating a new bucket if needed. The element is + /// constructed in place via T's move constructor, so types whose move-*assignment* is + /// unavailable or expensive (e.g. DenseArray, whose move-assign compares allocators) + /// still work. The element is published (bucket pointer first, then ``size_``) so a + /// concurrent reader that observes the new ``size()`` sees the constructed value. + /// Single-writer. + void push_back(T&& value) { + size_type i = size_.load(std::memory_order_relaxed); + auto [b, off] = locate_(i); + T* bucket = ensure_bucket_(b); + new (&bucket[off]) T(std::move(value)); + size_.store(i + 1, std::memory_order_release); + } + + /// @brief Drop the last element, destroying it (logical only; does not free the + /// bucket). Single-writer; caller must have drained readers if a bucket is later freed. + void pop_back() { + size_type i = size_.load(std::memory_order_relaxed); + if (i > 0) { + auto [b, off] = locate_(i - 1); + dir_[b].load(std::memory_order_relaxed)[off].~T(); + size_.store(i - 1, std::memory_order_release); + } + } + + /// @brief Shrink to ``n`` elements, destroying the dropped elements and freeing buckets + /// that lie entirely above ``n``. Single-writer; caller must have drained readers. + void shrink_to(size_type n) { + size_type old = size_.load(std::memory_order_relaxed); + if (n >= old) { + return; + } + // Stop readers from seeing the elements about to be destroyed. + size_.store(n, std::memory_order_release); + destroy_range_(n, old); + free_buckets_above_(n); + } + + private: + // Fixed top-level directory. Bucket k (when non-null) is a contiguous heap array of + // (kFirstBucket << k) elements; the elements with global index < size_ are constructed. + std::atomic dir_[kDirBuckets] = {}; + std::atomic size_{0}; + size_type allocated_buckets_{0}; + + // Number of elements held by bucket ``b`` (= kFirstBucket << b). + static constexpr size_type bucket_size_(size_type b) noexcept { + return kFirstBucket << b; + } + + // First global element index held by bucket ``b`` (= kFirstBucket * (2^b - 1)). + static constexpr size_type bucket_first_index_(size_type b) noexcept { + return kFirstBucket * ((size_type{1} << b) - 1); + } + + // Map element index ``i`` to (bucket, offset-within-bucket). Group elements into + // chunks of kFirstBucket, then apply the power-of-two bucket layout to the chunk + // index: q = i / kFirstBucket; bucket = floor(log2(q+1)); the bucket's first global + // index is kFirstBucket * (2^bucket - 1). + static constexpr std::pair locate_(size_type i) noexcept { + size_type q = i / kFirstBucket; + size_type bucket = static_cast(std::bit_width(q + 1)) - 1; + return {bucket, i - bucket_first_index_(bucket)}; + } + + // Ensure bucket ``b`` is allocated (single-writer) and return its base pointer. The + // bucket's elements are raw storage until constructed by the caller; the pointer is + // published with release so readers that later observe a matching size see it. + T* ensure_bucket_(size_type b) { + T* bucket = dir_[b].load(std::memory_order_relaxed); + if (bucket == nullptr) { + bucket = static_cast(::operator new[](bucket_size_(b) * sizeof(T))); + dir_[b].store(bucket, std::memory_order_release); + if (b + 1 > allocated_buckets_) { + allocated_buckets_ = b + 1; + } + } + return bucket; + } + + // Construct elements [from, to) in place (single-writer). ``fill`` is nullptr for + // default-construction. Allocates buckets as needed. + void construct_range_(size_type from, size_type to, const T* fill) { + for (size_type i = from; i < to; ++i) { + auto [b, off] = locate_(i); + T* bucket = ensure_bucket_(b); + if (fill == nullptr) { + new (&bucket[off]) T(); + } else { + new (&bucket[off]) T(*fill); + } + } + } + + // Destroy elements [from, to) (single-writer). Does not free buckets. + void destroy_range_(size_type from, size_type to) { + for (size_type i = from; i < to; ++i) { + auto [b, off] = locate_(i); + dir_[b].load(std::memory_order_relaxed)[off].~T(); + } + } + + void resize_impl_(size_type n, const T* fill) { + size_type old = size_.load(std::memory_order_relaxed); + if (n == old) { + return; + } + if (n < old) { + // Logical-only shrink (no bucket freeing — use shrink_to for reclamation), + // but still destroy the dropped elements to run their destructors. + size_.store(n, std::memory_order_release); + destroy_range_(n, old); + return; + } + // Grow: construct the new elements, then publish the new size last so a reader + // that observes it sees fully-constructed elements in published buckets. + construct_range_(old, n, fill); + size_.store(n, std::memory_order_release); + } + + // Free every bucket whose entire index range lies at or above ``n`` (single-writer; + // readers drained). A bucket straddling ``n`` keeps its allocation. + void free_buckets_above_(size_type n) { + for (size_type b = allocated_buckets_; b-- > 0;) { + if (bucket_first_index_(b) < n) { + break; // this and all lower buckets contain live (or kept) elements + } + T* bucket = dir_[b].load(std::memory_order_relaxed); + if (bucket != nullptr) { + ::operator delete[](static_cast(bucket)); + dir_[b].store(nullptr, std::memory_order_relaxed); + } + allocated_buckets_ = b; + } + } + + void destroy_all_() { + size_type n = size_.load(std::memory_order_relaxed); + destroy_range_(0, n); + for (size_type b = 0; b < allocated_buckets_; ++b) { + T* bucket = dir_[b].load(std::memory_order_relaxed); + if (bucket != nullptr) { + ::operator delete[](static_cast(bucket)); + dir_[b].store(nullptr, std::memory_order_relaxed); + } + } + size_.store(0, std::memory_order_relaxed); + allocated_buckets_ = 0; + } + + void copy_from_(const SegmentedVector& other) { + size_type n = other.size_.load(std::memory_order_relaxed); + for (size_type i = 0; i < n; ++i) { + auto [b, off] = locate_(i); + T* bucket = ensure_bucket_(b); + new (&bucket[off]) T(other[i]); + } + size_.store(n, std::memory_order_release); + } + + void steal_from_(SegmentedVector& other) noexcept { + for (size_type b = 0; b < kDirBuckets; ++b) { + dir_[b].store( + other.dir_[b].load(std::memory_order_relaxed), std::memory_order_relaxed + ); + other.dir_[b].store(nullptr, std::memory_order_relaxed); + } + size_.store(other.size_.load(std::memory_order_relaxed), std::memory_order_relaxed); + allocated_buckets_ = other.allocated_buckets_; + other.size_.store(0, std::memory_order_relaxed); + other.allocated_buckets_ = 0; + } +}; + +} // namespace svs::lib + +namespace svs { +// Opt SegmentedVector into svs::getindex's optional bounds checking. +template +inline constexpr bool enable_boundschecking> = true; +} // namespace svs diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 8c812d35a..b70a51a95 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -150,6 +150,10 @@ set(TEST_SOURCES ${TEST_DIR}/svs/quantization/scalar/scalar.cpp ${TEST_DIR}/svs/index/vamana/dynamic_index.cpp + + # Fine-grained concurrent index + ${TEST_DIR}/svs/concurrent/graph.cpp + ${TEST_DIR}/svs/concurrent/mutable_vamana_index.cpp ) ##### @@ -237,3 +241,59 @@ list(APPEND CMAKE_MODULE_PATH ${catch2_SOURCE_DIR}/extras) include(CTest) include(Catch) catch_discover_tests(tests ADD_TAGS_AS_LABELS SKIP_IS_FAILURE) + +##### +##### ThreadSanitizer targets for the fine-grained concurrent index +##### + +# The correctness of `svs::concurrent` rests almost entirely on memory ordering, which a +# non-instrumented test can only fail to disprove. These targets are opt-in because TSan +# costs roughly an order of magnitude in both time and memory: +# +# cmake -DSVS_EXPERIMENTAL_ENABLE_CONCURRENT_TSAN=YES ... +# ctest -L tsan +# +# `svs::svs` and Catch2 are not instrumented here, only the test translation units. That is +# enough: TSan checks the accesses it can see, and the accesses under test -- the adjacency +# lists and the sequence-lock counters -- are all in the instrumented TUs. +option(SVS_EXPERIMENTAL_ENABLE_CONCURRENT_TSAN + "Build ThreadSanitizer targets for svs::concurrent (slow)" OFF +) + +if (SVS_EXPERIMENTAL_ENABLE_CONCURRENT_TSAN) + message("Enabling ThreadSanitizer targets for svs::concurrent!") + + set(CONCURRENT_TSAN_SOURCES + ${TEST_DIR}/svs/concurrent/graph.cpp + ${TEST_DIR}/svs/concurrent/mutable_vamana_index.cpp + ) + + add_executable(concurrent_tsan ${CONCURRENT_TSAN_SOURCES}) + target_compile_options(concurrent_tsan PRIVATE -fsanitize=thread -g -O1) + target_link_options(concurrent_tsan PRIVATE -fsanitize=thread) + target_compile_definitions(concurrent_tsan PRIVATE SVS_THREAD_SANITIZER) + target_link_libraries(concurrent_tsan PRIVATE svs::svs Catch2::Catch2WithMain) + target_link_libraries( + concurrent_tsan PRIVATE svs_compile_options svs_x86_options_base + ) + add_test(NAME concurrent_tsan COMMAND concurrent_tsan) + set_tests_properties(concurrent_tsan PROPERTIES LABELS "tsan") + + # Negative control. Building the graph's element accessors as plain (non-atomic) loads + # and stores must make TSan complain -- otherwise a clean run above proves nothing about + # whether TSan was watching the right memory at all. + add_executable(concurrent_tsan_negative ${TEST_DIR}/svs/concurrent/graph.cpp) + target_compile_options(concurrent_tsan_negative PRIVATE -fsanitize=thread -g -O1) + target_link_options(concurrent_tsan_negative PRIVATE -fsanitize=thread) + target_compile_definitions(concurrent_tsan_negative + PRIVATE SVS_THREAD_SANITIZER SVS_CONCURRENT_UNSAFE_PLAIN_GRAPH_ACCESS + ) + target_link_libraries( + concurrent_tsan_negative PRIVATE svs::svs Catch2::Catch2WithMain + ) + target_link_libraries( + concurrent_tsan_negative PRIVATE svs_compile_options svs_x86_options_base + ) + add_test(NAME concurrent_tsan_negative COMMAND concurrent_tsan_negative) + set_tests_properties(concurrent_tsan_negative PROPERTIES WILL_FAIL TRUE LABELS "tsan") +endif() diff --git a/tests/svs/concurrent/graph.cpp b/tests/svs/concurrent/graph.cpp new file mode 100644 index 000000000..219a3e4f4 --- /dev/null +++ b/tests/svs/concurrent/graph.cpp @@ -0,0 +1,198 @@ +/* + * 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. + */ + +// header under test +#include "svs/concurrent/graph.h" + +#include "svs/concepts/graph.h" + +// catch2 +#include "catch2/catch_test_macros.hpp" + +// stl +#include +#include +#include +#include + +namespace { + +using Graph = svs::concurrent::SeqLockGraph; + +// The headline claim of this design: the new graph satisfies the *unmodified* upstream +// concepts, so upstream VamanaBuilder / prune / consolidate work against it with no edits +// to any existing SVS header. +static_assert(svs::graphs::ImmutableMemoryGraph); +static_assert(svs::graphs::MemoryGraph); + +} // namespace + +CATCH_TEST_CASE("SeqLockGraph single-threaded semantics", "[concurrent][graph]") { + Graph g{10, 4, /*segment_size=*/3}; + + CATCH_REQUIRE(g.n_nodes() == 10); + CATCH_REQUIRE(g.max_degree() == 4); + CATCH_REQUIRE(g.get_node_degree(0) == 0); + + CATCH_SECTION("add_edge returns the new out-degree") { + CATCH_REQUIRE(g.add_edge(0, 1) == 1); + CATCH_REQUIRE(g.add_edge(0, 2) == 2); + + // Duplicate and self-loop are both rejected without changing the degree. + CATCH_REQUIRE(g.add_edge(0, 1) == 2); + CATCH_REQUIRE(g.add_edge(0, 0) == 2); + + CATCH_REQUIRE(g.get_node_degree(0) == 2); + auto neighbors = g.get_node(0); + CATCH_REQUIRE(neighbors.size() == 2); + CATCH_REQUIRE(neighbors[0] == 1); + CATCH_REQUIRE(neighbors[1] == 2); + } + + CATCH_SECTION("adjacency lists saturate rather than overflow") { + for (uint32_t dst = 1; dst <= 4; ++dst) { + g.add_edge(0, dst); + } + CATCH_REQUIRE(g.get_node_degree(0) == 4); + CATCH_REQUIRE(g.add_edge(0, 5) == 4); + CATCH_REQUIRE(g.get_node_degree(0) == 4); + } + + CATCH_SECTION("replace_node and clear_node") { + g.add_edge(0, 1); + g.add_edge(0, 2); + + std::vector replacement{7, 8}; + g.replace_node(0, replacement); + CATCH_REQUIRE(g.get_node_degree(0) == 2); + auto neighbors = g.get_node(0); + CATCH_REQUIRE(neighbors.size() == 2); + CATCH_REQUIRE(neighbors[0] == 7); + CATCH_REQUIRE(neighbors[1] == 8); + + g.clear_node(0); + CATCH_REQUIRE(g.get_node_degree(0) == 0); + } +} + +CATCH_TEST_CASE("SeqLockGraph growth is address stable", "[concurrent][graph]") { + // A small segment size forces many segments, so growth definitely crosses one. + Graph g{3, 4, /*segment_size=*/3}; + g.add_edge(0, 1); + const void* before = g.get_node(0).data(); + + g.unsafe_resize(1000); + CATCH_REQUIRE(g.n_nodes() == 1000); + + // This is the property the whole design rests on: a reader holding a pointer into + // node 0's slot is unaffected by a writer growing the graph. + CATCH_REQUIRE(g.get_node(0).data() == before); + CATCH_REQUIRE(g.get_node_degree(0) == 1); + + // Newly added nodes are usable and zero-initialized. + CATCH_REQUIRE(g.get_node_degree(999) == 0); + CATCH_REQUIRE(g.add_edge(999, 1) == 1); +} + +// A writer repeatedly rewrites one node's adjacency list while readers use the +// sequence-lock protocol. Readers assert that every list they *accept* is internally +// consistent, i.e. they never observe a torn mixture of two generations. A plain +// std::vector-backed graph fails this test. +CATCH_TEST_CASE("SeqLockGraph rejects torn concurrent reads", "[concurrent][graph]") { + constexpr uint32_t kMaxDegree = 32; +#if defined(__SANITIZE_THREAD__) || defined(SVS_THREAD_SANITIZER) + // ThreadSanitizer costs roughly an order of magnitude in time, and it is looking for + // *races*, which show up just as readily in a shorter run. + constexpr size_t kIterations = 20000; +#else + constexpr size_t kIterations = 200000; +#endif + + Graph g{4, kMaxDegree, /*segment_size=*/2}; + + std::atomic stop{false}; + std::atomic torn_reads{0}; + std::atomic accepted_reads{0}; + std::atomic retries{0}; + + // The writer alternates between two generations. Generation k writes a list of length + // `len` where every element equals `k`, so any mixture of generations is detectable, + // and the two generations have different lengths so a torn *degree* is detectable too. + std::thread writer{[&] { + std::vector buffer; + for (size_t i = 0; i < kIterations; ++i) { + const uint32_t generation = (i % 2 == 0) ? 1u : 2u; + const size_t len = (i % 2 == 0) ? kMaxDegree : 4; + buffer.assign(len, generation); + g.replace_node(0, buffer); + } + stop.store(true); + }}; + + auto reader_body = [&] { + while (!stop.load(std::memory_order_relaxed)) { + for (;;) { + auto maybe_seq = g.seqlock(0).read_begin(); + if (!maybe_seq) { + retries.fetch_add(1, std::memory_order_relaxed); + continue; + } + auto neighbors = g.get_node_atomic(0); + bool uniform = true; + const uint32_t first = neighbors.empty() ? 0u : neighbors[0]; + for (size_t k = 0; k < neighbors.size(); ++k) { + if (neighbors[k] != first) { + uniform = false; + } + // A degree covering an unwritten slot would show up as 0 here, since + // segments are zero-initialized and no generation writes a 0. + if (neighbors[k] == 0) { + uniform = false; + } + } + if (!g.seqlock(0).read_validate(*maybe_seq)) { + retries.fetch_add(1, std::memory_order_relaxed); + continue; // Inconsistent read, correctly rejected. + } + // The sequence lock accepted this read, so it must be consistent. + if (!uniform && !neighbors.empty()) { + torn_reads.fetch_add(1, std::memory_order_relaxed); + } + accepted_reads.fetch_add(1, std::memory_order_relaxed); + break; + } + } + }; + + std::vector readers; + for (int i = 0; i < 8; ++i) { + readers.emplace_back(reader_body); + } + writer.join(); + for (auto& t : readers) { + t.join(); + } + + CATCH_INFO( + "accepted reads: " << accepted_reads.load() + << ", rejected/retried: " << retries.load() + << ", torn accepted reads: " << torn_reads.load() + ); + CATCH_REQUIRE(torn_reads.load() == 0); + CATCH_REQUIRE(accepted_reads.load() > 0); + // If this is zero then the test proved nothing: the writer never overlapped a reader. + CATCH_REQUIRE(retries.load() > 0); +} diff --git a/tests/svs/concurrent/mutable_vamana_index.cpp b/tests/svs/concurrent/mutable_vamana_index.cpp new file mode 100644 index 000000000..bfb666844 --- /dev/null +++ b/tests/svs/concurrent/mutable_vamana_index.cpp @@ -0,0 +1,493 @@ +/* + * 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. + */ + +// Concurrency tests for ``svs::concurrent::MutableVamanaIndex``: correctness of searches +// issued while another thread inserts and deletes. +// +// Recall thresholds here are deliberately coarse. They exist to catch a graph that has been +// corrupted into uselessness, not to track search quality -- that is the job of the +// benchmark suite. +// +// Assertions inside the hot loops accumulate into counters and are checked once at the end. +// Catch2's assertion bookkeeping is not free, and these loops run for millions of +// iterations. + +// header under test +#include "svs/concurrent/mutable_vamana_index.h" + +#include "svs/core/data.h" +#include "svs/core/distance.h" +#include "svs/index/vamana/vamana_build.h" +#include "svs/lib/threads.h" + +// catch2 +#include "catch2/catch_test_macros.hpp" + +// stl +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +// ThreadSanitizer costs roughly an order of magnitude in both time and memory, and it is +// looking for *races*, which show up just as readily in a small index. Shrink the problem +// rather than skipping the run. +#if defined(__SANITIZE_THREAD__) || defined(SVS_THREAD_SANITIZER) +constexpr size_t kInitialPoints = 4000; +constexpr size_t kIncrementalPoints = 1000; +constexpr size_t kNumQueries = 50; +#else +constexpr size_t kInitialPoints = 20000; +constexpr size_t kIncrementalPoints = 5000; +constexpr size_t kNumQueries = 200; +#endif + +constexpr size_t kDim = 32; +constexpr size_t kMaxDegree = 32; +constexpr size_t kNumNeighbors = 10; +constexpr size_t kBuildThreads = 8; + +using Idx = uint32_t; +using Alloc = svs::lib::Allocator; +using BlockedData = svs::data::BlockedData; +using Distance = svs::distance::DistanceL2; +using ConcurrentIndex = svs::concurrent::MutableVamanaIndex; + +std::vector random_vectors(size_t n, size_t dim, uint32_t seed) { + std::mt19937 rng{seed}; + std::normal_distribution dist{0.0f, 1.0f}; + std::vector out(n * dim); + for (auto& v : out) { + v = dist(rng); + } + return out; +} + +svs::data::SimpleData make_dataset(const std::vector& raw, size_t dim) { + const size_t n = raw.size() / dim; + auto data = svs::data::SimpleData(n, dim); + for (size_t i = 0; i < n; ++i) { + data.set_datum(i, std::span(raw.data() + i * dim, dim)); + } + return data; +} + +// Build a concurrent index over ``raw``, using the *unmodified* upstream builder. That this +// compiles at all is a load-bearing part of the design: ``SeqLockGraph`` satisfies +// ``svs::graphs::MemoryGraph`` as published, so ``VamanaBuilder`` needs no changes. +std::unique_ptr build_index( + const std::vector& raw, size_t dim, std::span ids, size_t threads +) { + const size_t n = raw.size() / dim; + auto data = BlockedData(n, dim); + for (size_t i = 0; i < n; ++i) { + data.set_datum(i, std::span(raw.data() + i * dim, dim)); + } + + auto threadpool = + svs::threads::ThreadPoolHandle{svs::threads::DefaultThreadPool{threads}}; + auto distance = Distance{}; + auto entry_point = + svs::index::vamana::extensions::compute_entry_point(data, threadpool); + + auto graph = svs::concurrent::SeqLockGraph{n, kMaxDegree}; + auto parameters = svs::index::vamana::VamanaBuildParameters{ + 1.2f, kMaxDegree, 2 * kMaxDegree, 750, kMaxDegree, true}; + auto prefetch = svs::index::vamana::extensions::estimate_prefetch_parameters(data); + auto builder = svs::index::vamana::VamanaBuilder{ + graph, data, distance, parameters, threadpool, prefetch}; + builder.construct(1.2f, static_cast(entry_point)); + + return std::make_unique( + std::move(graph), + std::move(data), + static_cast(entry_point), + distance, + ids, + svs::threads::DefaultThreadPool{threads} + ); +} + +// Brute-force ground truth over the given set of live external IDs. +std::vector> ground_truth( + const std::vector& base, + const std::unordered_set& live, + const std::vector& queries, + size_t dim, + size_t k +) { + const size_t nq = queries.size() / dim; + std::vector> result(nq); + for (size_t q = 0; q < nq; ++q) { + std::vector> scored; + scored.reserve(live.size()); + for (size_t id : live) { + float d = 0; + for (size_t j = 0; j < dim; ++j) { + float diff = queries[q * dim + j] - base[id * dim + j]; + d += diff * diff; + } + scored.emplace_back(d, id); + } + std::partial_sort( + scored.begin(), + scored.begin() + static_cast(std::min(k, scored.size())), + scored.end() + ); + for (size_t i = 0; i < std::min(k, scored.size()); ++i) { + result[q].push_back(scored[i].second); + } + } + return result; +} + +double recall_at_k( + const svs::QueryResult& got, const std::vector>& expected +) { + size_t hits = 0, total = 0; + for (size_t q = 0; q < expected.size(); ++q) { + std::unordered_set truth{expected[q].begin(), expected[q].end()}; + for (size_t j = 0; j < got.n_neighbors(); ++j) { + if (truth.count(got.index(q, j))) { + ++hits; + } + } + total += truth.size(); + } + return total == 0 ? 1.0 : static_cast(hits) / static_cast(total); +} + +// Insert ``[first, first + n)`` of ``base`` as a single batch. +void add_batch( + ConcurrentIndex& index, const std::vector& base, size_t first, size_t n +) { + auto batch = svs::data::SimpleData(n, kDim); + std::vector batch_ids(n); + for (size_t i = 0; i < n; ++i) { + batch.set_datum(i, std::span(base.data() + (first + i) * kDim, kDim)); + batch_ids[i] = first + i; + } + index.add_points(batch, batch_ids); +} + +} // namespace + +CATCH_TEST_CASE("Concurrent MutableVamanaIndex quiescent recall", "[concurrent][index]") { + auto base = random_vectors(kInitialPoints, kDim, 1234); + std::vector ids(kInitialPoints); + std::iota(ids.begin(), ids.end(), 0); + + auto index = build_index(base, kDim, ids, kBuildThreads); + CATCH_REQUIRE(index->size() == kInitialPoints); + + auto queries_raw = random_vectors(kNumQueries, kDim, 999); + auto queries = make_dataset(queries_raw, kDim); + + auto sp = index->get_search_parameters(); + sp.buffer_config({100}); + index->set_search_parameters(sp); + + auto results = svs::QueryResult{kNumQueries, kNumNeighbors}; + index->search(results.view(), queries, index->get_search_parameters()); + + std::unordered_set live{ids.begin(), ids.end()}; + auto truth = ground_truth(base, live, queries_raw, kDim, kNumNeighbors); + const double recall = recall_at_k(results, truth); + + // A correctly built Vamana graph at this window size should be well above 0.9. + CATCH_INFO("quiescent recall@" << kNumNeighbors << " = " << recall); + CATCH_REQUIRE(recall > 0.90); +} + +// The core test: searches run continuously while a writer inserts new vectors and deletes +// existing ones. Any torn adjacency read, use-after-free from a resize, or missing ID +// translation surfaces as a crash, an exception, or an invalid ID. +CATCH_TEST_CASE( + "Concurrent MutableVamanaIndex search during mutation", "[concurrent][index]" +) { + const size_t total = kInitialPoints + kIncrementalPoints; + auto base = random_vectors(total, kDim, 4321); + + std::vector initial_ids(kInitialPoints); + std::iota(initial_ids.begin(), initial_ids.end(), 0); + // Build over the first ``kInitialPoints`` only; the tail is inserted concurrently + // below. + auto initial_slice = std::vector( + base.begin(), base.begin() + static_cast(kInitialPoints * kDim) + ); + auto index = build_index(initial_slice, kDim, initial_ids, kBuildThreads); + + auto sp = index->get_search_parameters(); + sp.buffer_config({100}); + index->set_search_parameters(sp); + + auto queries_raw = random_vectors(kNumQueries, kDim, 777); + auto queries = make_dataset(queries_raw, kDim); + + std::atomic writer_done{false}; + std::atomic searches_completed{0}; + std::atomic invalid_ids{0}; + std::atomic deleted_mid_search{0}; + std::atomic duplicate_ids{0}; + std::atomic exceptions{0}; + + // Writer: insert the incremental points in batches, deleting some older ones as it goes + // so that the deletion path is exercised too. + std::thread writer{[&] { + try { + constexpr size_t kBatch = 500; + for (size_t offset = 0; offset < kIncrementalPoints; offset += kBatch) { + const size_t n = std::min(kBatch, kIncrementalPoints - offset); + add_batch(*index, base, kInitialPoints + offset, n); + + // Delete a fresh, disjoint set of the original IDs each round, *spread* + // across the whole ID range with a stride. A contiguous low-ID slice would + // almost never intersect a query's top-k, so the interesting race -- a + // result slot retired between selection and ID translation -- would go + // unexercised. Round r takes every 40th ID starting at r, so the rounds are + // disjoint and together retire 25% of the original vectors. + const size_t round = offset / kBatch; + std::vector to_delete; + for (size_t id = round; id < kInitialPoints; id += 40) { + to_delete.push_back(id); + } + index->delete_entries(to_delete); + } + } catch (const std::exception& e) { + CATCH_WARN("writer threw: " << e.what()); + exceptions.fetch_add(1); + } + writer_done.store(true); + }}; + + // Searchers: hammer the index with single-query searches throughout. + auto searcher = [&] { + try { + auto scratch = index->scratchspace(); + std::unordered_set seen_ids; + while (!writer_done.load(std::memory_order_relaxed)) { + for (size_t q = 0; q < kNumQueries; ++q) { + auto query = + std::span(queries_raw.data() + q * kDim, kDim); + index->search(query, scratch); + // ``[0, valid())`` is the region a caller is allowed to read: skipped + // (deleted) candidates have been compacted out by this point. + const size_t n = + std::min(kNumNeighbors, scratch.buffer.valid()); + seen_ids.clear(); + for (size_t j = 0; j < n; ++j) { + auto internal = scratch.buffer[j].id(); + // A ranked result must never list the same vector twice. Torn + // adjacency reads or a botched retry in the seqlock section would + // show up here, because the search buffer dedupes by ID and can + // only be fooled by inconsistent input. + if (!seen_ids.insert(internal).second) { + duplicate_ids.fetch_add(1, std::memory_order_relaxed); + } + // Every ID a search hands back must either translate to a live + // external ID, or be a slot retired by the concurrent deleter. + // Anything else means the graph led us to a node the translator + // never knew about -- e.g. a slot published before its ID mapping. + auto external = index->try_translate_internal_id(internal); + if (external == ConcurrentIndex::invalid_external_id) { + deleted_mid_search.fetch_add(1, std::memory_order_relaxed); + } else if (!index->has_id(external)) { + invalid_ids.fetch_add(1, std::memory_order_relaxed); + } + } + searches_completed.fetch_add(1, std::memory_order_relaxed); + } + } + } catch (const std::exception& e) { + CATCH_WARN("searcher threw: " << e.what()); + exceptions.fetch_add(1); + } + }; + + std::vector searchers; + for (int i = 0; i < 8; ++i) { + searchers.emplace_back(searcher); + } + writer.join(); + for (auto& t : searchers) { + t.join(); + } + + CATCH_INFO( + "searches completed: " << searches_completed.load() + << ", neighbors retired mid-search (expected, dropped): " + << deleted_mid_search.load() + ); + CATCH_REQUIRE(exceptions.load() == 0); + CATCH_REQUIRE(invalid_ids.load() == 0); + CATCH_REQUIRE(duplicate_ids.load() == 0); + CATCH_REQUIRE(searches_completed.load() > 0); + + // Post-mutation the index must still be a correct index. + std::unordered_set live; + index->on_ids([&live](size_t id) { live.insert(id); }); + CATCH_REQUIRE(live.size() == index->size()); + + auto truth = ground_truth(base, live, queries_raw, kDim, kNumNeighbors); + auto results = svs::QueryResult{kNumQueries, kNumNeighbors}; + index->search(results.view(), queries, index->get_search_parameters()); + const double recall = recall_at_k(results, truth); + CATCH_INFO("post-mutation recall@" << kNumNeighbors << " = " << recall); + CATCH_REQUIRE(recall > 0.85); + + // And consolidation/compaction must leave it correct too. + index->consolidate(); + index->compact(); + auto results2 = svs::QueryResult{kNumQueries, kNumNeighbors}; + index->search(results2.view(), queries, index->get_search_parameters()); + const double recall2 = recall_at_k(results2, truth); + CATCH_INFO("post-consolidate/compact recall@" << kNumNeighbors << " = " << recall2); + CATCH_REQUIRE(recall2 > 0.85); + CATCH_REQUIRE(index->size() == live.size()); +} + +// The batch iterator runs upstream's *unmodified* ``BatchIterator`` over a +// ``SeqLockGraphView``. This checks it works at all, and that it keeps working while a +// writer mutates the index. +CATCH_TEST_CASE("Concurrent MutableVamanaIndex batch iterator", "[concurrent][index]") { + const size_t total = kInitialPoints + kIncrementalPoints; + auto base = random_vectors(total, kDim, 24680); + + std::vector initial_ids(kInitialPoints); + std::iota(initial_ids.begin(), initial_ids.end(), 0); + auto initial_slice = std::vector( + base.begin(), base.begin() + static_cast(kInitialPoints * kDim) + ); + auto index = build_index(initial_slice, kDim, initial_ids, kBuildThreads); + + auto queries_raw = random_vectors(kNumQueries, kDim, 13579); + + CATCH_SECTION("quiescent") { + // Batches must be non-overlapping, and each batch must be sorted. + // + // Batches are *not* globally monotonic and it would be wrong to assert that: the + // iterator is approximate, so a later batch can surface a vector closer than one + // the earlier batch's window had already returned. Count those inversions and + // report them as a quality signal rather than a correctness one -- this is upstream + // behaviour and has nothing to do with concurrency. + auto query = std::span(queries_raw.data(), kDim); + auto it = index->make_batch_iterator(query); + std::unordered_set all; + float previous_worst = -1.0f; + size_t batches = 0; + size_t inversions = 0; + size_t repeats = 0; + size_t unsorted = 0; + for (; batches < 5 && !it.done(); ++batches) { + it.next(10); + float last = -1.0f; + for (const auto& n : it) { + if (!all.insert(n.id()).second) { + ++repeats; + } + if (n.distance() < last) { + ++unsorted; + } + last = n.distance(); + if (n.distance() < previous_worst) { + ++inversions; + } + } + if (it.size() != 0) { + previous_worst = (it.end() - 1)->distance(); + } + } + CATCH_INFO( + batches << " batches, " << all.size() << " distinct vectors, " << inversions + << " cross-batch inversions" + ); + CATCH_REQUIRE(repeats == 0); + CATCH_REQUIRE(unsorted == 0); + CATCH_REQUIRE(all.size() >= 40); + } + + CATCH_SECTION("during mutation") { + // The claim under test is memory safety and per-batch consistency, *not* that a + // long-lived cursor sees a stable snapshot. + std::atomic writer_done{false}; + std::atomic batches_completed{0}; + std::atomic exceptions{0}; + std::atomic bad_ids{0}; + + std::thread writer{[&] { + try { + constexpr size_t kBatch = 500; + for (size_t offset = 0; offset < kIncrementalPoints; offset += kBatch) { + const size_t n = std::min(kBatch, kIncrementalPoints - offset); + add_batch(*index, base, kInitialPoints + offset, n); + } + } catch (const std::exception& e) { + CATCH_WARN("writer threw: " << e.what()); + exceptions.fetch_add(1); + } + writer_done.store(true); + }}; + + auto reader = [&] { + try { + while (!writer_done.load(std::memory_order_relaxed)) { + for (size_t q = 0; q < kNumQueries; ++q) { + auto query = + std::span(queries_raw.data() + q * kDim, kDim); + auto it = index->make_batch_iterator(query); + for (size_t b = 0; b < 3 && !it.done(); ++b) { + it.next(10); + for (const auto& n : it) { + // Any ID the iterator yields must be a real external ID (or + // one retired mid-flight); a bad graph read shows up here. + if (n.id() != ConcurrentIndex::invalid_external_id && + !index->has_id(n.id())) { + bad_ids.fetch_add(1, std::memory_order_relaxed); + } + } + batches_completed.fetch_add(1, std::memory_order_relaxed); + } + } + } + } catch (const std::exception& e) { + CATCH_WARN("reader threw: " << e.what()); + exceptions.fetch_add(1); + } + }; + + std::vector readers; + for (int i = 0; i < 4; ++i) { + readers.emplace_back(reader); + } + writer.join(); + for (auto& t : readers) { + t.join(); + } + + CATCH_INFO("batches completed during mutation: " << batches_completed.load()); + CATCH_REQUIRE(exceptions.load() == 0); + CATCH_REQUIRE(bad_ids.load() == 0); + CATCH_REQUIRE(batches_completed.load() > 0); + } +}