fix(search): Use thread lock for hnsw multi reader/multi writer mutex (#6205)

fix(hnsw): Use thread lock for hnsw multi reader/multi writer mutex

When we expire keys in periodic heartbeat we shouldn't take any fiber
locks when removing documents from hnsw index. Use thread lock instead.

Signed-off-by: mkaruza <[email protected]>

Author: mkaruza

src/core/search/mrmw_mutex.h

@@ -2,8 +2,10 @@
 // See LICENSE for licensing terms.
 //
 
+#include <condition_variable>
+
 #include "base/logging.h"
-#include "util/fibers/synchronization.h"
+#include "base/spinlock.h"
 
 namespace dfly::search {
 
@@ -50,16 +52,18 @@ class MRMWMutex {
     return target_mode == LockMode::kReadLock ? reader_waiters_ : writer_waiters_;
   };
 
-  inline util::fb2::CondVar& GetCondVar(LockMode target_mode) {
+  inline std::condition_variable_any& GetCondVar(LockMode target_mode) {
     return target_mode == LockMode::kReadLock ? reader_cond_var_ : writer_cond_var_;
   };
 
   static inline LockMode GetInverseMode(LockMode mode) {
     return mode == LockMode::kReadLock ? LockMode::kWriteLock : LockMode::kReadLock;
   }
 
-  util::fb2::Mutex mutex_;
-  util::fb2::CondVar reader_cond_var_, writer_cond_var_;
+  // TODO: use fiber sync primitives in future
+  base::SpinLock mutex_;
+  std::condition_variable_any reader_cond_var_, writer_cond_var_;
+
   size_t writer_waiters_ = 0, reader_waiters_ = 0;
   size_t active_runners_ = 0;
   LockMode lock_mode_;

src/core/search/mrmw_mutex_test.cc

@@ -176,49 +176,79 @@ TEST_F(MRMWMutexTest, WriterAfterReaders) {
   EXPECT_EQ(write_count.load(), 0);
 }
 
-// Test 5: Mix of readers and writes
-TEST_F(MRMWMutexTest, MixWritersReaders) {
+TEST_F(MRMWMutexTest, MixWritersReadersOnDifferentFibers) {
   std::atomic<size_t> read_count(0);
   std::atomic<size_t> write_count(0);
 
   // Start multiple readers and writers
   const int num_threads = 100;
   std::vector<util::fb2::Fiber> threads;
-  threads.reserve(num_threads + 1);
+  threads.reserve(num_threads);
 
-  // Add long read task that will block all write tasks
-  threads.emplace_back(
-      pp_->at(0)->LaunchFiber([&] { ReadTask(&mutex_, std::ref(read_count), 2000); }));
-
-  // Give long writer time to acquire the lock
-  util::ThisFiber::SleepFor(std::chrono::milliseconds(100));
-
-  size_t write_threads = 0;
   for (int i = 0; i < num_threads; ++i) {
-    size_t fiber_id = rand() % 2;
     if (rand() % 3) {
-      threads.emplace_back(pp_->at(fiber_id)->LaunchFiber(util::fb2::Launch::post, [&] {
+      threads.emplace_back(pp_->at(0)->LaunchFiber(util::fb2::Launch::post, [&] {
         ReadTask(&mutex_, std::ref(read_count), kReadTaskSleepTime);
       }));
     } else {
-      write_threads++;
-      threads.emplace_back(pp_->at(fiber_id)->LaunchFiber(util::fb2::Launch::post, [&] {
+      threads.emplace_back(pp_->at(1)->LaunchFiber(util::fb2::Launch::post, [&] {
         WriteTask(&mutex_, std::ref(write_count), kWriteTaskSleepTime);
       }));
     }
   }
 
-  // All shorter threads should be done and only long one remains
-  util::ThisFiber::SleepFor(std::chrono::milliseconds(500));
-
-  EXPECT_EQ(read_count.load(), 1);
-
-  EXPECT_EQ(write_count.load(), write_threads);
-
   // Wait for all readers to acquire and release the lock
   for (auto& t : threads) {
     t.Join();
   }
 }
 
+// TODO: Once we have fiber locking we can test scenario where we write/read on same fibers
+// current implementation block thread so it is not possible to test this for now.
+
+// Test 6: Mix of readers and writes on random fibers
+// TEST_F(MRMWMutexTest, MixWritersReadersOnFibers) {
+//   std::atomic<size_t> read_count(0);
+//   std::atomic<size_t> write_count(0);
+
+//   // Start multiple readers and writers
+//   const int num_threads = 100;
+//   std::vector<util::fb2::Fiber> threads;
+//   threads.reserve(num_threads + 1);
+
+//   // Add long read task that will block all write tasks
+//   threads.emplace_back(
+//       pp_->at(0)->LaunchFiber([&] { ReadTask(&mutex_, std::ref(read_count), 2000); }));
+
+//   // Give long writer time to acquire the lock
+//   util::ThisFiber::SleepFor(std::chrono::milliseconds(100));
+
+//   size_t write_threads = 0;
+//   for (int i = 0; i < num_threads; ++i) {
+//     size_t fiber_id = rand() % 2;
+//     if (rand() % 3) {
+//       threads.emplace_back(pp_->at(fiber_id)->LaunchFiber(util::fb2::Launch::post, [&] {
+//         ReadTask(&mutex_, std::ref(read_count), kReadTaskSleepTime);
+//       }));
+//     } else {
+//       write_threads++;
+//       threads.emplace_back(pp_->at(fiber_id)->LaunchFiber(util::fb2::Launch::post, [&] {
+//         WriteTask(&mutex_, std::ref(write_count), kWriteTaskSleepTime);
+//       }));
+//     }
+//   }
+
+//   // All shorter threads should be done and only long one remains
+//   util::ThisFiber::SleepFor(std::chrono::milliseconds(500));
+
+//   EXPECT_EQ(read_count.load(), 1);
+
+//   EXPECT_EQ(write_count.load(), write_threads);
+
+//   // Wait for all readers to acquire and release the lock
+//   for (auto& t : threads) {
+//     t.Join();
+//   }
+// }
+
 }  // namespace dfly::search