Fix/improve threadpooling

Author: rfsaliev

src/VecSim/algorithms/svs/svs.h

@@ -20,6 +20,7 @@ struct SVSIndexBase {
     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 size_t indexStorageSize() const = 0;
+    virtual size_t getNumThreads() const = 0;
     virtual void setNumThreads(size_t numThreads) = 0;
     virtual size_t getThreadPoolCapacity() const = 0;
 };
@@ -180,14 +181,25 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
         // Wrap data into SVS SimpleDataView for SVS API
         auto points = svs::data::SimpleDataView<DataType>{typed_vectors_data, n, this->dim};
 
-        // SVS index instance cannot be empty, so we have to construct it at first rows
+        // If n == 1, we should ensure single-threading
+        const size_t num_threads = (n == 1) ? getNumThreads() : 1;
+        if (num_threads > 1) {
+            setNumThreads(1);
+        }
+
         if (!impl_) {
+            // SVS index instance cannot be empty, so we have to construct it at first rows
             initImpl(points, ids);
-            return n - deleted_num;
+        } else {
+            // Add new points to existing SVS index
+            impl_->add_points(points, ids);
+        }
+
+        // Restore multi-threading if needed
+        if (num_threads > 1) {
+            setNumThreads(num_threads);
         }
 
-        // Add new points to existing SVS index
-        impl_->add_points(points, ids);
         return n - deleted_num;
     }
 
@@ -209,7 +221,19 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
             return 0;
         }
 
+        // If entries_to_delete.size() == 1, we should ensure single-threading
+        const size_t num_threads = (entries_to_delete.size() == 1) ? getNumThreads() : 1;
+        if (num_threads > 1) {
+            setNumThreads(1);
+        }
+
         impl_->delete_entries(entries_to_delete);
+
+        // Restore multi-threading if needed
+        if (num_threads > 1) {
+            setNumThreads(num_threads);
+        }
+
         this->markIndexUpdate(entries_to_delete.size());
         return entries_to_delete.size();
     }
@@ -307,6 +331,7 @@ class SVSIndex : public VecSimIndexAbstract<svs_details::vecsim_dt<DataType>, fl
         return deleteVectorsImpl(labels, n);
     }
 
+    size_t getNumThreads() const override { return threadpool_.size(); }
     void setNumThreads(size_t numThreads) override { threadpool_.resize(numThreads); }
 
     size_t getThreadPoolCapacity() const override { return threadpool_.capacity(); }

src/VecSim/algorithms/svs/svs_tiered.h

@@ -9,6 +9,25 @@
 #include <memory>
 #include <mutex>
 
+/**
+ * @class SVSMultiThreadJob
+ * @brief Represents a multi-threaded asynchronous job for the SVS algorithm.
+ *
+ * This class is responsible for managing multi-threaded jobs, including thread reservation,
+ * synchronization, and execution of tasks. It uses a control block to coordinate threads
+ * and ensure proper execution of the job.
+ *
+ * @details
+ * The SVSMultiThreadJob class supports creating multiple threads for a task and ensures
+ * synchronization between them. It uses a nested ControlBlock class to manage thread
+ * reservations and job completion. Additionally, it includes a nested ReserveThreadJob
+ * class to handle individual thread reservations.
+ *
+ * The main job executes a user-defined task with the number of reserved threads, while
+ * additional threads wait for the main job to complete.
+ *
+ * @note This class is designed to work with the AsyncJob framework.
+ */
 class SVSMultiThreadJob : public AsyncJob {
 
     // Thread reservation control block shared between all threads
@@ -32,16 +51,11 @@ class SVSMultiThreadJob : public AsyncJob {
         // reserve a thread and wait for the job to be done
         void reserveThreadAndWait() {
             // count current thread
-            {
-                std::lock_guard lock{mutex};
-                ++reservedThreads;
-            }
-            cv.notify_one();
-            // wait for the job to be done
-            {
-                std::unique_lock lock{mutex};
-                cv.wait(lock, [&] { return jobDone; });
-            }
+            std::unique_lock lock{mutex};
+            ++reservedThreads;
+            cv.notify_all();
+            // Wait until the job is marked as done, handling potential spurious wakeups.
+            cv.wait(lock, [&] { return jobDone; });
         }
 
         // wait for threads to be reserved
@@ -528,7 +542,7 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
             // Use the frontend parameters to manually prepare the blob for its transfer to the SVS
             // index.
             auto storage_blob = this->frontendIndex->preprocessForStorage(blob);
-            std::unique_lock<std::shared_mutex> svs_lock(this->mainIndexGuard);
+            std::scoped_lock lock(this->updateJobMutex, this->mainIndexGuard);
             return svs_index->addVectors(storage_blob.get(), &label, 1);
         }
         bool index_update_needed = false;
@@ -557,7 +571,7 @@ class TieredSVSIndex : public VecSimTieredIndex<DataType, float> {
                 return !this->frontendIndex->isLabelExists(label);
             }());
 
-            std::unique_lock<std::shared_mutex> svs_lock(this->mainIndexGuard);
+            std::scoped_lock lock(this->updateJobMutex, this->mainIndexGuard);
             return svs_index->deleteVectors(&label, 1);
         }