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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions bindings/c/docs/C_API_Design.md
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,8 @@ The main search structure providing vector similarity search operations.
**Dynamic Index Parameters:**

The `_ex` variants of the dynamic functions (`svs_index_build_dynamic_ex`,
`svs_index_load_dynamic_ex`, `svs_index_convert_dynamic_ex`,
`svs_index_load_dynamic_ex`, `svs_index_load_stream_dynamic_ex`,
`svs_index_convert_dynamic_ex`,
`svs_index_builder_estimate_memory_dynamic_ex`,
`svs_index_builder_estimate_search_memory_dynamic_ex`) take a versioned
`const svs_dynamic_index_params_t*` instead of a bare `blocksize_bytes` argument.
Expand All @@ -277,7 +278,7 @@ Invalid parameters are rejected with `SVS_ERROR_INVALID_ARGUMENT`; unlike the le
|-----------------|----------|
| `SVS_SYNC_KIND_NONE` | No internal synchronization (default); the caller must serialize access |
| `SVS_SYNC_KIND_GLOBAL` | Index-wide reader/writer lock: read-only operations (search, `has_id`, `get_distance`, reconstruct, `get_size`, `get_num_threads`, memory queries, use as a conversion source) run concurrently; mutating operations (add/delete points, consolidate, compact, save, `set_num_threads`) are exclusive |
| `SVS_SYNC_KIND_FINE_GRAIN` | Backed by the concurrent Vamana index, which synchronizes updates itself: read-only operations and add/delete points, consolidate, compact all take a shared lock and run in parallel (compact still waits inside the index). Save, `set_num_threads`, memory queries and use as a conversion source are exclusive |
| `SVS_SYNC_KIND_FINE_GRAIN` | Backed by the concurrent Vamana index, which synchronizes updates itself: read-only operations and add/delete points, consolidate, compact all take a shared lock and run in parallel (compact still waits inside the index). Save takes the shared lock too: it runs alongside searches, waits for running writers, and does not consolidate or compact. `set_num_threads`, memory queries and use as a conversion source are exclusive |

Synchronization limits:
- `svs_index_free()` must not run concurrently with any other call on the same handle.
Expand Down Expand Up @@ -654,7 +655,7 @@ for full signatures, parameters, and Doxygen documentation.
| **Search params** | `svs_search_params_create_vamana`, `svs_search_params_free` |
| **Builder** | `svs_index_builder_create`, `svs_index_builder_set_{storage,threadpool,threadpool_custom,allocator,allocator_custom}`, `svs_index_builder_free` |
| **Memory estimation** | `svs_index_builder_estimate_memory`, `svs_index_builder_estimate_memory_dynamic[_ex]`, `svs_index_builder_estimate_search_memory`, `svs_index_builder_estimate_search_memory_dynamic[_ex]`, `svs_index_builder_get_default_blocksize_bytes` |
| **Index lifecycle** | `svs_index_build`, `svs_index_build_dynamic[_ex]`, `svs_index_convert`, `svs_index_convert_dynamic[_ex]`, `svs_index_load`, `svs_index_load_dynamic[_ex]`, `svs_index_load_stream`, `svs_index_load_stream_dynamic`, `svs_index_save`, `svs_index_save_stream`, `svs_index_free` |
| **Index lifecycle** | `svs_index_build`, `svs_index_build_dynamic[_ex]`, `svs_index_convert`, `svs_index_convert_dynamic[_ex]`, `svs_index_load`, `svs_index_load_dynamic[_ex]`, `svs_index_load_stream`, `svs_index_load_stream_dynamic[_ex]`, `svs_index_save`, `svs_index_save_stream`, `svs_index_free` |
| **Dynamic ops** | `svs_index_dynamic_{add_points,delete_points,has_id,consolidate,compact,get_sync_kind}` |
| **Introspection** | `svs_index_get_num_threads` / `set_num_threads`, `svs_index_get_distance`, `svs_index_reconstruct`, `svs_index_get_memory_usage`, `svs_index_get_memory_breakdown` |
| **Search** | `svs_index_search_topk` (+ deprecated `svs_index_search`), `svs_search_results_free` |
Expand Down
25 changes: 24 additions & 1 deletion bindings/c/include/svs/c/svs_c.h
Original file line number Diff line number Diff line change
Expand Up @@ -597,7 +597,9 @@ struct svs_memory_breakdown {
/// With SVS_SYNC_KIND_FINE_GRAIN, the handle is backed by a concurrent index that
/// synchronizes updates itself: add_points, delete_points, consolidate and compact take a
/// shared lock and run alongside searches and each other (compact still waits for
/// in-flight operations inside the index). Only save, set_num_threads, memory queries
/// in-flight operations inside the index). save takes a shared lock and runs alongside
/// searches; the index blocks its own writers while saving and keeps deleted and empty
/// slots as they are (no consolidate or compact). Only set_num_threads, memory queries
/// and use as a conversion source take an exclusive lock.
///
/// @remarks Synchronization covers operations on the index only:
Expand Down Expand Up @@ -1411,6 +1413,27 @@ SVS_API svs_index_h svs_index_load_stream_dynamic(
svs_error_h out_err /*=NULL*/
);

/// @brief Load a dynamic index from a caller-supplied stream with explicit parameters
/// @param builder The index builder handle (used for configuration)
/// @param stream The stream interface to read the index from
/// @param params Pointer to the dynamic index parameters structure, or NULL for defaults
/// @param out_err An optional error handle to capture errors
/// @return A handle to the loaded dynamic index
/// @remarks The block parameters only affect the in-memory layout, not the on-disk
/// format. The sync kind is not persisted with the index and must be passed on every load.
/// @remarks The operations table is copied, but @p stream->self is retained as-is. It only
/// needs to remain valid until this function returns.
/// @remarks Accepts both the native stream encoding and a packed directory archive. The
/// native encoding carries no total length, so a load may read up to 64 KiB past the end
/// of the index; frame the payload if embedding it in a larger stream.
/// @error SVS_ERROR_INVALID_ARGUMENT if @p params is invalid (see svs_dynamic_index_params)
SVS_API svs_index_h svs_index_load_stream_dynamic_ex(
svs_index_builder_h builder,
svs_stream_i stream,
const svs_dynamic_index_params_t* params /*=NULL*/,
svs_error_h out_err /*=NULL*/
);

/// @brief Convert an index using new builder configuration
/// @param builder The index builder handle (used for configuration)
/// @param src_index The source index handle to convert from
Expand Down
60 changes: 59 additions & 1 deletion bindings/c/src/dispatcher_concurrent_vamana.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
#include <svs/orchestrators/concurrent_dynamic_vamana.h>

#include <filesystem>
#include <istream>
#include <memory>
#include <span>
#include <utility>
#include <variant>
Expand Down Expand Up @@ -196,9 +198,38 @@ svs::DynamicVamana load_concurrent_vamana_index(
);
}

template <typename DataLoader, typename Distance>
svs::DynamicVamana load_stream_concurrent_vamana_index(
const svs::index::vamana::VamanaBuildParameters& SVS_UNUSED(build_params),
std::unique_ptr<std::istream> stream,
DataLoader SVS_UNUSED(loader),
Distance D,
svs::threads::ThreadPoolHandle pool,
const AllocatorBuilder& allocator_builder,
const svs::data::BlockingParameters& block_params,
svs::logging::logger_ptr logger
) {
using allocator_type = typename DataLoader::allocator_type;
using value_type = typename allocator_type::value_type;
using data_type = typename DataLoader::data_type;
auto allocator = allocator_type{block_params, allocator_builder.build<value_type>()};
auto graph_allocator = ConcurrentGraphAllocator{
block_params, allocator_builder.build_for_graph<uint32_t>()};

return svs::ConcurrentDynamicVamana::assemble<float>(
*stream,
svs::make_stream_loader<ConcurrentGraph>(graph_allocator),
svs::make_stream_loader<data_type>(allocator),
std::move(D),
std::move(pool),
std::move(logger)
);
}

using ConcurrentVamanaSource = std::variant<
std::pair<svs::data::ConstSimpleDataView<float>, std::span<const size_t>>,
std::filesystem::path>;
std::filesystem::path,
std::unique_ptr<std::istream>>;

using BuildConcurrentIndexDispatcher = svs::lib::Dispatcher<
svs::DynamicVamana,
Expand All @@ -221,10 +252,15 @@ const BuildConcurrentIndexDispatcher& build_concurrent_vamana_index_dispatcher()
d.register_target(&load_concurrent_vamana_index<DataLoader, Distance>);
};

auto load_stream_closure = [&d]<typename DataLoader, typename Distance>() {
d.register_target(&load_stream_concurrent_vamana_index<DataLoader, Distance>);
};

constexpr auto kind = ConcurrentVamanaFlavor::block_kind;

for_all_specializations<kind>(build_closure);
for_all_specializations<kind>(load_closure);
for_all_specializations<kind>(load_stream_closure);
return d;
}();
return dispatcher;
Expand Down Expand Up @@ -293,6 +329,28 @@ svs::DynamicVamana dispatch_concurrent_vamana_index_load(
);
}

svs::DynamicVamana dispatch_concurrent_vamana_index_load_stream(
const svs::index::vamana::VamanaBuildParameters& build_params,
std::unique_ptr<std::istream> stream,
const Storage* storage,
svs::DistanceType distance_type,
svs::threads::ThreadPoolHandle pool,
const AllocatorBuilder& allocator_builder,
const svs::data::BlockingParameters& block_params,
svs::logging::logger_ptr logger
) {
return build_concurrent_vamana_index_dispatcher().invoke(
build_params,
ConcurrentVamanaSource{std::move(stream)},
storage,
distance_type,
std::move(pool),
allocator_builder,
block_params,
std::move(logger)
);
}

svs::DynamicVamana dispatch_concurrent_vamana_index_copy(
const svs::index::vamana::VamanaBuildParameters& build_params,
const svs::DynamicVamana& src_index,
Expand Down
11 changes: 11 additions & 0 deletions bindings/c/src/dispatcher_concurrent_vamana.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,17 @@ svs::DynamicVamana dispatch_concurrent_vamana_index_load(
svs::logging::logger_ptr logger
);

svs::DynamicVamana dispatch_concurrent_vamana_index_load_stream(
const svs::index::vamana::VamanaBuildParameters& build_params,
std::unique_ptr<std::istream> stream,
const Storage* storage,
svs::DistanceType distance_type,
svs::threads::ThreadPoolHandle pool,
const AllocatorBuilder& allocator_builder,
const svs::data::BlockingParameters& block_params,
svs::logging::logger_ptr logger
);

// Copies between a concurrent and a regular or concurrent dynamic index;
// at least one of `src_concurrent` and `dst_concurrent` must be true.
svs::DynamicVamana dispatch_concurrent_vamana_index_copy(
Expand Down
40 changes: 23 additions & 17 deletions bindings/c/src/index_builder.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -346,23 +346,29 @@ std::shared_ptr<DynamicIndex> IndexBuilder::load_stream_dynamic(
) {
if (algorithm->type == SVS_ALGORITHM_TYPE_VAMANA) {
auto vamana_algorithm = static_cast<AlgorithmVamana*>(algorithm.get());

auto index = std::make_shared<DynamicIndexVamana>(
*this,
dispatch_dynamic_vamana_index_load_stream(
vamana_algorithm->build_parameters(),
std::move(stream),
storage.get(),
to_distance_type(distance_metric),
pool_builder.build(sync_kind),
allocator_builder,
block_params,
get_logger()
),
sync_kind
);

return index;
const auto& build_params = vamana_algorithm->build_parameters();
auto index = uses_concurrent_index(sync_kind)
? dispatch_concurrent_vamana_index_load_stream(
build_params,
std::move(stream),
storage.get(),
to_distance_type(distance_metric),
pool_builder.build(sync_kind),
allocator_builder,
block_params,
get_logger()
)
: dispatch_dynamic_vamana_index_load_stream(
build_params,
std::move(stream),
storage.get(),
to_distance_type(distance_metric),
pool_builder.build(sync_kind),
allocator_builder,
block_params,
get_logger()
);
return make_dynamic_index(*this, std::move(index), sync_kind);
}
return nullptr;
}
Expand Down
15 changes: 13 additions & 2 deletions bindings/c/src/index_vamana.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -158,8 +158,9 @@ struct DynamicIndexVamana : public DynamicIndex {
/// Dynamic index backed by svs::ConcurrentDynamicVamana.
///
/// The index synchronizes add/delete/consolidate/compact itself, so they take the shared
/// lock alongside searches. Operations that are not thread-safe in the index (save,
/// set_num_threads) and whole-index reads (memory breakdown, use as a conversion source)
/// lock alongside searches. save() also takes the shared lock: the index excludes its own
/// writers while saving. Operations that are not thread-safe in the index
/// (set_num_threads) and whole-index reads (memory breakdown, use as a conversion source)
/// take the exclusive lock.
struct ConcurrentIndexVamana : public DynamicIndexVamana {
ConcurrentIndexVamana(
Expand All @@ -179,6 +180,16 @@ struct ConcurrentIndexVamana : public DynamicIndexVamana {

size_t delete_points(std::span<const size_t> ids) override;

void save(const std::filesystem::path& directory) override {
auto lock = read_lock();
index.save(directory / "config", directory / "graph", directory / "data");
}

void save(std::ostream& stream) override {
auto lock = read_lock();
index.save(stream);
}

void consolidate() override {
auto lock = read_lock();
index.consolidate();
Expand Down
34 changes: 34 additions & 0 deletions bindings/c/src/svs_c.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1245,6 +1245,40 @@ extern "C" svs_index_h svs_index_load_stream_dynamic(
);
}

extern "C" svs_index_h svs_index_load_stream_dynamic_ex(
svs_index_builder_h builder,
svs_stream_i stream,
const svs_dynamic_index_params_t* params,
svs_error_h out_err
) {
using namespace svs::c_runtime;
return wrap_exceptions(
[&]() {
EXPECT_ARG_NOT_NULL(builder);
EXPECT_ARG_NOT_NULL(stream);
NOT_IMPLEMENTED_IF(
(builder->impl->algorithm->type != SVS_ALGORITHM_TYPE_VAMANA),
"Only Vamana algorithm is currently supported for dynamic index loading"
);
StreamBuf::validate(stream, /*need_write=*/false);
auto dynamic_params = read_dynamic_index_params(params);
auto index = builder->impl->load_stream_dynamic(
std::make_unique<InputStream>(*stream->ops, stream->self),
dynamic_params.block_params,
dynamic_params.sync_kind
);
if (index == nullptr) {
SET_ERROR(out_err, SVS_ERROR_RUNTIME, "Dynamic index load failed");
return svs_index_h{nullptr};
}
auto result = new svs_index;
result->impl = index;
return result;
},
out_err
);
}

extern "C" svs_index_h
svs_index_convert(svs_index_builder_h builder, svs_index_h src_index, svs_error_h out_err) {
using namespace svs::c_runtime;
Expand Down
8 changes: 8 additions & 0 deletions data/serialization.toml
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,14 @@ schema_version = "v0.0.0"
parameters = "table"
translation = "table"

[[schemas]]
__schema__ = "vamana_dynamic_auxiliary_parameters"
__version__ = "v0.0.1"
name = "str"
parameters = "table"
translation = "table"
status = "table"

[[schemas]]
__schema__ = "vamana_build_parameters"
__version__ = "v0.0.1"
Expand Down
19 changes: 14 additions & 5 deletions include/svs/concurrent/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,12 +84,18 @@ storage smaller, so it must wait for every reader to finish first. It is the onl
caller that takes `compact_mutex_` exclusive. It runs `consolidate_locked()`
before it compacts.

### 1d. Not safe at all
### 1d. `save()`

`save()` writes the slots as they are, with deleted and empty slots included. It does
not call `consolidate()` or `compact()`. It takes `writers_gatekeeper_` and then
`writers_mutex_` exclusive, so it waits for running writers and blocks new ones. New
writers hold the gatekeeper only while acquiring `writers_mutex_`, so they cannot
starve `save()`. Searches keep running.

### 1e. Not safe at all

Call these only when no other thread uses the index:

- `save()` - it holds no lock while it writes the files. It calls `consolidate()`
and `compact()` first, but it releases both locks before writing.
- `set_alpha`, `set_prune_to`, `set_max_candidates`,
`set_construction_window_size`, `set_threadpool` - plain writes, no lock.

Expand Down Expand Up @@ -142,6 +148,8 @@ There are three groups:

| Lock | Protects | Taken exclusive by |
| --- | --- | --- |
| `writers_mutex_` | the index state stays unchanged while `save()` writes it | `save()` only |
| `writers_gatekeeper_` | new writers wait behind a pending `save()` | `save()` only |
| `compact_mutex_` | the storage stays alive | `compact()` only |
| `translator_mutex_` | the two id hash maps | `add_points`, `consolidate`, `compact`, `replace_external_id` |
| `pending_insertions_mutex_` | `pending_insertions_` | `add_points` |
Expand All @@ -159,6 +167,7 @@ so it waits for all of them.
deadlock.

```
writers_gatekeeper_ -> writers_mutex_ -> compact_gatekeeper_ -> compact_mutex_
compact_mutex_ -> slot_alloc_mutex_
compact_mutex_ -> translator_mutex_
l2e_mutex_ -> e2l_mutex_ (multi index)
Expand Down Expand Up @@ -247,8 +256,8 @@ The dataset must use a `SegmentedBlocked` allocator; this is checked at compile
time. `distance` may be a functor or a `svs::DistanceType`. `batch_iterator()`
returns this index's own `BatchIterator`, so it takes the locks described in 1a.

The rules from section 1 apply unchanged: `save()`, `set_threadpool()` and the
build-parameter setters still need exclusive access.
The rules from section 1 apply unchanged: `set_threadpool()` and the
build-parameter setters still need exclusive access; `save()` follows 1d.

```cpp
using Data = svs::index::vamana::concurrent::SegmentedBlockedData<float>;
Expand Down
Loading
Loading