Skip to content

[8.4] SVS shared thread pool with rental model (MOD-9881, MOD-17090) #1003

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dor-forer merged 4 commits into from
Jul 23, 2026
Merged

[8.4] SVS shared thread pool with rental model (MOD-9881, MOD-17090) #1003

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7102be2
refactor: SVS shared thread pool with rental model (MOD-9881) (#925)
meiravgri Apr 15, 2026
bbaa34a
MOD-15578 Track shared SVS thread pool memory & expose it through pub…
dor-forer Jun 4, 2026
947a37b
MOD-15579 Lazy initialization of the shared SVS thread pool (#971)
dor-forer Jun 9, 2026
0b975a1
MOD-15579 Report 0 shared SVS memory until first index attaches (#979)
dor-forer Jun 11, 2026
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
55 changes: 33 additions & 22 deletions src/VecSim/algorithms/svs/svs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,14 @@ struct SVSIndexBase
{
SVSIndexBase() : num_marked_deleted{0} {};
virtual ~SVSIndexBase() = default;

virtual int addVectors(const void *vectors_data, const labelType *labels, size_t n) = 0;
virtual int deleteVectors(const labelType *labels, size_t n) = 0;
virtual bool isLabelExists(labelType label) const = 0;
virtual size_t indexStorageSize() const = 0;
virtual size_t getNumThreads() const = 0;
virtual void setNumThreads(size_t numThreads) = 0;
virtual size_t getThreadPoolCapacity() const = 0;
virtual size_t getParallelism() const = 0;
virtual void setParallelism(size_t parallelism) = 0;
virtual size_t getPoolSize() const = 0;
virtual bool isCompressed() const = 0;
size_t getNumMarkedDeleted() const { return num_marked_deleted; }

Expand All @@ -66,9 +67,9 @@ struct SVSIndexBase
};

/** Thread Management Strategy:
* - addVector(): Requires numThreads == 1
* - addVectors(): Allows any numThreads value, but prohibits n=1 with numThreads>1
* - Callers are responsible for setting appropriate thread counts
* - addVector(): Requires parallelism == 1
* - addVectors(): Allows any parallelism value, but prohibits n=1 with parallelism>1
* - Callers are responsible for setting appropriate parallelism
**/
template <typename MetricType, typename DataType, bool isMulti, size_t QuantBits,
size_t ResidualBits, bool IsLeanVec>
Expand Down Expand Up @@ -251,12 +252,12 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
this->impl_ = std::move(svs_handler->impl);
}

// Assuming numThreads was updated to reflect the number of available threads before this
// Assuming parallelism was updated to reflect the number of available threads before this
// function was called.
// This function assumes that the caller has already set numThreads to the appropriate value
// This function assumes that the caller has already set parallelism to the appropriate value
// for the operation.
// Important NOTE: For single vector operations (n=1), numThreads should be 1.
// For bulk operations (n>1), numThreads should reflect the number of available threads.
// Important NOTE: For single vector operations (n=1), parallelism should be 1.
// For bulk operations (n>1), parallelism should reflect the number of available threads.
int addVectorsImpl(const void *vectors_data, const labelType *labels, size_t n) {
if (n == 0) {
return 0;
Expand Down Expand Up @@ -361,9 +362,13 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
svs_details::getOrDefault(params.leanvec_dim, SVS_VAMANA_DEFAULT_LEANVEC_DIM)},
epsilon{svs_details::getOrDefault(params.epsilon, SVS_VAMANA_DEFAULT_EPSILON)},
is_two_level_lvq{isTwoLevelLVQ(params.quantBits)},
threadpool_{std::max(size_t{SVS_VAMANA_DEFAULT_NUM_THREADS}, params.num_threads)},
impl_{nullptr} {
threadpool_{this->allocator, this->logCallbackCtx}, impl_{nullptr} {
logger_ = makeLogger();
if (params.num_threads != 0) {
this->log(VecSimCommonStrings::LOG_WARNING_STRING,
"SVSParams.num_threads is deprecated and ignored. "
"Thread pool size is controlled globally via VecSim_UpdateThreadPoolSize().");
}
}

~SVSIndex() = default;
Expand Down Expand Up @@ -412,8 +417,8 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
.maxCandidatePoolSize = this->buildParams.max_candidate_pool_size,
.pruneTo = this->buildParams.prune_to,
.useSearchHistory = this->buildParams.use_full_search_history,
.numThreads = this->getThreadPoolCapacity(),
.lastReservedThreads = this->getNumThreads(),
.numThreads = this->getPoolSize(),
.lastReservedThreads = this->getParallelism(),
.numberOfMarkedDeletedNodes = this->num_marked_deleted,
.searchWindowSize = this->search_window_size,
.searchBufferCapacity = this->search_buffer_capacity,
Expand All @@ -424,8 +429,14 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl

VecSimDebugInfoIterator *debugInfoIterator() const override {
VecSimIndexDebugInfo info = this->debugInfo();
// For readability. Update this number when needed.
size_t numberOfInfoFields = 23;
// Capacity hint: 26 = 25 fields added below (1 ALGORITHM + 9 from
// addCommonInfoToIterator + 15 SVS-specific) + 1 for the SHARED_MEMORY field
// that the C API wrapper VecSimIndex_DebugInfoIterator appends after this
// method returns. Reserving the extra slot avoids a reallocation on the
// top-level path; when nested in a tiered BACKEND_INDEX the C API does not
// append it, so the hint over-reserves by one (harmless).
// Update this number when fields are added or removed.
size_t numberOfInfoFields = 26;
VecSimDebugInfoIterator *infoIterator =
new VecSimDebugInfoIterator(numberOfInfoFields, this->allocator);

Expand Down Expand Up @@ -517,16 +528,16 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl

int addVector(const void *vector_data, labelType label) override {
// Enforce single-threaded execution for single vector operations to ensure optimal
// performance and consistent behavior. Callers must set numThreads=1 before calling this
// performance and consistent behavior. Callers must set parallelism=1 before calling this
// method.
assert(getNumThreads() == 1 && "Can't use more than one thread to insert a single vector");
assert(getParallelism() == 1 && "Can't use more than one thread to insert a single vector");
return addVectorsImpl(vector_data, &label, 1);
}

int addVectors(const void *vectors_data, const labelType *labels, size_t n) override {
// Prevent misuse: single vector operations should use addVector(), not addVectors() with
// n=1 This ensures proper thread management and API contract enforcement.
assert(!(n == 1 && getNumThreads() > 1) &&
assert(!(n == 1 && getParallelism() > 1) &&
"Can't use more than one thread to insert a single vector");
return addVectorsImpl(vectors_data, labels, n);
}
Expand All @@ -541,10 +552,10 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
return impl_ ? impl_->has_id(label) : false;
}

size_t getNumThreads() const override { return threadpool_.size(); }
void setNumThreads(size_t numThreads) override { threadpool_.resize(numThreads); }
size_t getParallelism() const override { return threadpool_.getParallelism(); }
void setParallelism(size_t parallelism) override { threadpool_.setParallelism(parallelism); }

size_t getThreadPoolCapacity() const override { return threadpool_.capacity(); }
size_t getPoolSize() const override { return threadpool_.poolSize(); }

bool isCompressed() const override { return storage_traits_t::is_compressed(); }

Expand Down
59 changes: 38 additions & 21 deletions src/VecSim/algorithms/svs/svs_tiered.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ class SVSMultiThreadJob : public AsyncJob {
task_type task;
std::shared_ptr<ControlBlock> controlBlock;
JobsRegistry *jobsRegistry;
bool isScheduled; // true if this job holds a pending-job reservation on the thread pool

@cursor cursor Bot Jul 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Task exception permanently leaks pending_jobs counter

Medium Severity

If jobPtr->task() throws in ExecuteMultiThreadJobImpl, delete_job(job) is never reached, so the ~SVSMultiThreadJob destructor never runs and endScheduledJob() is never called. This permanently increments pending_jobs_, causing all future pool shrink requests to be deferred indefinitely and never applied. The isScheduled / endScheduledJob() mechanism is new in this PR, introducing this failure mode where the deferred-resize protocol becomes permanently stuck.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 0b975a1. Configure here.

@dor-forer dor-forer Jul 23, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed — real bug, and worse than described: ExecuteMultiThreadJobImpl is invoked from RediSearch's plain-C pthread worker pool (no C++ handler anywhere on that call stack), so an uncaught exception here calls std::terminate() — a process crash, not just a stuck pending_jobs_ counter. task() realistically can throw: updateSVSIndexWrapper/SVSIndexGCWrapper → updateSVSIndex() → createImpl/setImpl, which already contain explicit throw std::logic_error(...) (non-empty impl_, failed dynamic_cast) plus ordinary bad_alloc risk.

Important: this isn't introduced by this backport. isScheduled/endScheduledJob() are verbatim from #925, already merged and live on main, 8.6, and 8.6-rse. Bugbot flags it as new only because it's diffing against 8.4's pre-#925 baseline.

Not fixing this in the backport PR — patching only here would diverge from upstream and this needs a real fix (wrap the task() call, decide failure semantics) on main first, then flow down through the other branches like everything else. Tracking separately.


static void ExecuteMultiThreadJobImpl(AsyncJob *job) {
auto *jobPtr = static_cast<SVSMultiThreadJob *>(job);
Expand All @@ -162,26 +163,44 @@ class SVSMultiThreadJob : public AsyncJob {

SVSMultiThreadJob(std::shared_ptr<VecSimAllocator> allocator, JobType jobType,
task_type callback, VecSimIndex *index,
std::shared_ptr<ControlBlock> controlBlock, JobsRegistry *registry)
std::shared_ptr<ControlBlock> controlBlock, JobsRegistry *registry,
bool scheduled = false)
: AsyncJob(std::move(allocator), jobType, ExecuteMultiThreadJobImpl, index),
task(std::move(callback)), controlBlock(std::move(controlBlock)), jobsRegistry(registry) {
task(std::move(callback)), controlBlock(std::move(controlBlock)), jobsRegistry(registry),
isScheduled(scheduled) {}

~SVSMultiThreadJob() {
if (isScheduled) {
VecSimSVSThreadPoolImpl::instance()->endScheduledJob();
}
}

public:
template <typename Rep, typename Period>
static vecsim_stl::vector<AsyncJob *>
createScheduledJobs(const std::shared_ptr<VecSimAllocator> &allocator, JobType jobType,
std::function<void(VecSimIndex *, size_t)> callback, VecSimIndex *index,
std::chrono::duration<Rep, Period> threads_wait_timeout,
JobsRegistry *registry) {
size_t num_threads = VecSimSVSThreadPoolImpl::instance()->beginScheduledJob();
return createJobs(allocator, jobType, callback, index, num_threads, threads_wait_timeout,
registry, /*scheduled=*/true);
}

template <typename Rep, typename Period>
static vecsim_stl::vector<AsyncJob *>
createJobs(const std::shared_ptr<VecSimAllocator> &allocator, JobType jobType,
std::function<void(VecSimIndex *, size_t)> callback, VecSimIndex *index,
size_t num_threads, std::chrono::duration<Rep, Period> threads_wait_timeout,
JobsRegistry *registry) {
JobsRegistry *registry, bool scheduled = false) {
assert(num_threads > 0);
std::shared_ptr<ControlBlock> controlBlock =
num_threads == 1 ? nullptr
: std::make_shared<ControlBlock>(num_threads, threads_wait_timeout);

vecsim_stl::vector<AsyncJob *> jobs(num_threads, allocator);
jobs[0] = new (allocator)
SVSMultiThreadJob(allocator, jobType, callback, index, controlBlock, registry);
jobs[0] = new (allocator) SVSMultiThreadJob(allocator, jobType, callback, index,
controlBlock, registry, scheduled);
for (size_t i = 1; i < num_threads; ++i) {
jobs[i] =
new (allocator) ReserveThreadJob(allocator, jobType, index, controlBlock, registry);
Expand Down Expand Up @@ -587,7 +606,8 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
// No need to run GC on an empty index.
return;
}
svs_index->setNumThreads(std::min(availableThreads, index->backendIndex->indexSize()));
index->executeTracingCallback("GCJob::before_run_gc");
svs_index->setParallelism(std::min(availableThreads, index->backendIndex->indexSize()));
// VecSimIndexAbstract::runGC() is protected
static_cast<VecSimIndexInterface *>(index->backendIndex)->runGC();
}
Expand All @@ -601,9 +621,8 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
return;
}

auto total_threads = this->GetSVSIndex()->getThreadPoolCapacity();
auto jobs = SVSMultiThreadJob::createJobs(
this->allocator, SVS_BATCH_UPDATE_JOB, updateSVSIndexWrapper, this, total_threads,
auto jobs = SVSMultiThreadJob::createScheduledJobs(
this->allocator, SVS_BATCH_UPDATE_JOB, updateSVSIndexWrapper, this,
std::chrono::microseconds(updateJobWaitTime), &uncompletedJobs);
this->submitJobs(jobs);
}
Expand All @@ -614,9 +633,8 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
return;
}

auto total_threads = this->GetSVSIndex()->getThreadPoolCapacity();
auto jobs = SVSMultiThreadJob::createJobs(
this->allocator, SVS_GC_JOB, SVSIndexGCWrapper, this, total_threads,
auto jobs = SVSMultiThreadJob::createScheduledJobs(
this->allocator, SVS_GC_JOB, SVSIndexGCWrapper, this,
std::chrono::microseconds(updateJobWaitTime), &uncompletedJobs);
this->submitJobs(jobs);
}
Expand Down Expand Up @@ -677,13 +695,14 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
} // release frontend index

executeTracingCallback("UpdateJob::before_add_to_svs");
{ // lock backend index for writing and add vectors there
if (!labels_to_move.empty()) {
// lock backend index for writing and add vectors there
std::shared_lock main_shared_lock(this->mainIndexGuard);
auto svs_index = GetSVSIndex();
assert(labels_to_move.size() == vectors_to_move.size() / this->frontendIndex->getDim());
if (this->backendIndex->indexSize() == 0) {
// If backend index is empty, we need to initialize it first.
svs_index->setNumThreads(std::min(availableThreads, labels_to_move.size()));
svs_index->setParallelism(std::min(availableThreads, labels_to_move.size()));
auto impl = svs_index->createImpl(vectors_to_move.data(), labels_to_move.data(),
labels_to_move.size());

Expand All @@ -696,7 +715,7 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
main_shared_lock.unlock();
std::lock_guard lock(this->mainIndexGuard);
// Upgrade to unique lock to add vectors
svs_index->setNumThreads(std::min(availableThreads, labels_to_move.size()));
svs_index->setParallelism(std::min(availableThreads, labels_to_move.size()));
svs_index->addVectors(vectors_to_move.data(), labels_to_move.data(),
labels_to_move.size());
}
Expand Down Expand Up @@ -819,11 +838,9 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
// prevent update job from running in parallel and lock any access to the backend
// index
std::scoped_lock lock(this->updateJobMutex, this->mainIndexGuard);
// Set available thread count to 1 for single vector write-in-place operation.
// This maintains the contract that single vector operations use exactly one thread.
// TODO: Replace this setNumThreads(1) call with an assertion once we establish
// a contract that write-in-place mode guarantees numThreads == 1.
svs_index->setNumThreads(1);
// Defensive: ensure single-threaded operation for write-in-place mode.
// parallelism_ defaults to 1, so this is a no-op in the normal case.
svs_index->setParallelism(1);
return this->backendIndex->addVector(storage_blob.get(), label);
}
}
Expand Down Expand Up @@ -1087,7 +1104,7 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
return;
}
// Force single thread for write-in-place mode.
this->GetSVSIndex()->setNumThreads(1);
this->GetSVSIndex()->setParallelism(1);
// VecSimIndexAbstract::runGC() is protected
static_cast<VecSimIndexInterface *>(this->backendIndex)->runGC();
return;
Expand Down
Loading
Loading