attempt to fix syncing mechanism

cpp/bench/ann/src/common/benchmark.hpp

@@ -39,6 +39,7 @@ namespace raft::bench::ann {
 
 std::mutex init_mutex;
 std::condition_variable cond_var;
+bool processed = false;
 
 static inline std::unique_ptr<AnnBase> current_algo{nullptr};
 static inline std::shared_ptr<AlgoProperty> current_algo_props{nullptr};
@@ -199,6 +200,9 @@ void bench_search(::benchmark::State& state,
    */
   if (state.thread_index() == 0) {
     std::unique_lock lk(init_mutex);
+    std::cout << "Thread 0 acquired on lock" << std::endl;
+    cond_var.wait(lk, [] { return !processed; });
+    std::cout << "Thread 0 starting to process" << std::endl;
     // algo is static to cache it between close search runs to save time on index loading
     static std::string index_file = "";
     if (index.file != index_file) {
@@ -247,13 +251,16 @@ void bench_search(::benchmark::State& state,
     }
 
     query_set = dataset->query_set(current_algo_props->query_memory_type);
+    processed = true;
+    std::cout << "Thread 0 finishing process, about to notify waiting threads" << std::endl;
     cond_var.notify_all();
   } else {
-    std::unique_lock lk(init_mutex);
+    std::unique_lock lk(init_mutex, std::defer_lock);
     // All other threads will wait for the first thread to initialize the algo.
-
+    std::cout << "Other threads waiting on state to become true" << std::endl;
     cond_var.wait(
-      lk, [] { return current_algo_props.get() != nullptr && current_algo.get() != nullptr; });
+      lk, [] { return processed; });
+      std::cout << "Other threads are alive, state is true" << std::endl;
     // gbench ensures that all threads are synchronized at the start of the benchmark loop.
     // We are accessing shared variables (like current_algo, current_algo_probs) before the
     // benchmark loop, therefore the synchronization here is necessary.
@@ -340,6 +347,11 @@ void bench_search(::benchmark::State& state,
       double actual_recall = static_cast<double>(match_count) / static_cast<double>(total_count);
       state.counters.insert({{"Recall", actual_recall}});
     }
+    std::cout << "Last thread about to acquire lock" << std::endl;
+    std::unique_lock lk(init_mutex);
+    processed = false;
+    std::cout << "Last thread reset sync, about to notify" << std::endl;
+    cond_var.notify_all();
   }
 }