Merge pull request #428 from vinser52/svinogra_refactor_mt_tests

Share multithread utils between tests and benchmarks

Author: vinser52

benchmark/CMakeLists.txt

@@ -42,8 +42,10 @@ function(add_umf_benchmark)
         LIBS ${BENCH_LIBS})
 
     target_include_directories(
-        ${BENCH_NAME} PRIVATE ${UMF_CMAKE_SOURCE_DIR}/include
-                              ${UMF_CMAKE_SOURCE_DIR}/src/utils)
+        ${BENCH_NAME}
+        PRIVATE ${UMF_CMAKE_SOURCE_DIR}/include
+                ${UMF_CMAKE_SOURCE_DIR}/src/utils
+                ${UMF_CMAKE_SOURCE_DIR}/test/common)
 
     target_link_directories(${BENCH_NAME} PRIVATE ${ARG_LIBDIRS})
 

benchmark/multithread.hpp

@@ -17,75 +17,9 @@
 #include <thread>
 #include <vector>
 
-namespace umf_bench {
-
-template <typename Function>
-void parallel_exec(size_t threads_number, Function &&f) {
-    std::vector<std::thread> threads;
-    threads.reserve(threads_number);
-
-    for (size_t i = 0; i < threads_number; ++i) {
-        threads.emplace_back([&](size_t id) { f(id); }, i);
-    }
-
-    for (auto &t : threads) {
-        t.join();
-    }
-}
-
-class latch {
-  public:
-    latch(size_t desired) : counter(desired) {}
-
-    /* Returns true for the last thread arriving at the latch, false for all
-     * other threads. */
-    bool wait(std::unique_lock<std::mutex> &lock) {
-        counter--;
-        if (counter > 0) {
-            cv.wait(lock, [&] { return counter == 0; });
-            return false;
-        } else {
-            /*
-             * notify_call could be called outside of a lock
-             * (it would perform better) but drd complains
-             * in that case
-             */
-            cv.notify_all();
-            return true;
-        }
-    }
+#include "multithread_helpers.hpp"
 
-  private:
-    std::condition_variable cv;
-    size_t counter = 0;
-};
-
-/* Implements multi-use barrier (latch). Once all threads arrive at the
- * latch, a new latch is allocated and used by all subsequent calls to
- * syncthreads. */
-struct syncthreads_barrier {
-    syncthreads_barrier(size_t num_threads) : num_threads(num_threads) {
-        mutex = std::shared_ptr<std::mutex>(new std::mutex);
-        current_latch = std::shared_ptr<latch>(new latch(num_threads));
-    }
-
-    syncthreads_barrier(const syncthreads_barrier &) = delete;
-    syncthreads_barrier &operator=(const syncthreads_barrier &) = delete;
-    syncthreads_barrier(syncthreads_barrier &&) = default;
-
-    void operator()() {
-        std::unique_lock<std::mutex> lock(*mutex);
-        auto l = current_latch;
-        if (l->wait(lock)) {
-            current_latch = std::shared_ptr<latch>(new latch(num_threads));
-        }
-    }
-
-  private:
-    size_t num_threads;
-    std::shared_ptr<std::mutex> mutex;
-    std::shared_ptr<latch> current_latch;
-};
+namespace umf_bench {
 
 template <typename TimeUnit, typename F>
 typename TimeUnit::rep measure(F &&func) {
@@ -110,14 +44,12 @@ auto measure(size_t iterations, size_t concurrency, F &&run_workload) {
 
     for (size_t i = 0; i < iterations; i++) {
         std::vector<ResultsType> iteration_results(concurrency);
-        syncthreads_barrier syncthreads(concurrency);
-        parallel_exec(concurrency, [&](size_t id) {
+        umf_test::syncthreads_barrier syncthreads(concurrency);
+        umf_test::parallel_exec(concurrency, [&](size_t id) {
             syncthreads();
 
             iteration_results[id] =
                 measure<TimeUnit>([&]() { run_workload(id); });
-
-            syncthreads();
         });
 
         // skip the first 'warmup' iteration

src/critnib/critnib.c

@@ -343,7 +343,7 @@ int critnib_insert(struct critnib *c, word key, void *value, int update) {
 
     struct critnib_node *n = c->root;
     if (!n) {
-        c->root = kn;
+        store(&c->root, kn);
 
         util_mutex_unlock(&c->mutex);
 

test/common/multithread_helpers.hpp

@@ -0,0 +1,86 @@
+/*
+ *
+ * Copyright (C) 2024 Intel Corporation
+ *
+ * Under the Apache License v2.0 with LLVM Exceptions. See LICENSE.TXT.
+ * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+ *
+ */
+
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace umf_test {
+
+template <typename Function>
+void parallel_exec(size_t threads_number, Function &&f) {
+    std::vector<std::thread> threads;
+    threads.reserve(threads_number);
+
+    for (size_t i = 0; i < threads_number; ++i) {
+        threads.emplace_back([&](size_t id) { f(id); }, i);
+    }
+
+    for (auto &t : threads) {
+        t.join();
+    }
+}
+
+class latch {
+  public:
+    latch(size_t desired) : counter(desired) {}
+
+    /* Returns true for the last thread arriving at the latch, false for all
+     * other threads. */
+    bool wait(std::unique_lock<std::mutex> &lock) {
+        counter--;
+        if (counter > 0) {
+            cv.wait(lock, [&] { return counter == 0; });
+            return false;
+        } else {
+            /*
+             * notify_call could be called outside of a lock
+             * (it would perform better) but drd complains
+             * in that case
+             */
+            cv.notify_all();
+            return true;
+        }
+    }
+
+  private:
+    std::condition_variable cv;
+    size_t counter = 0;
+};
+
+/* Implements multi-use barrier (latch). Once all threads arrive at the
+ * latch, a new latch is allocated and used by all subsequent calls to
+ * syncthreads. */
+struct syncthreads_barrier {
+    syncthreads_barrier(size_t num_threads) : num_threads(num_threads) {
+        mutex = std::shared_ptr<std::mutex>(new std::mutex);
+        current_latch = std::shared_ptr<latch>(new latch(num_threads));
+    }
+
+    syncthreads_barrier(const syncthreads_barrier &) = delete;
+    syncthreads_barrier &operator=(const syncthreads_barrier &) = delete;
+    syncthreads_barrier(syncthreads_barrier &&) = default;
+
+    void operator()() {
+        std::unique_lock<std::mutex> lock(*mutex);
+        auto l = current_latch;
+        if (l->wait(lock)) {
+            current_latch = std::shared_ptr<latch>(new latch(num_threads));
+        }
+    }
+
+  private:
+    size_t num_threads;
+    std::shared_ptr<std::mutex> mutex;
+    std::shared_ptr<latch> current_latch;
+};
+
+} // namespace umf_test

test/ipcAPI.cpp

@@ -3,6 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 // This file contains tests for UMF pool API
 
+#include "multithread_helpers.hpp"
 #include "pool.hpp"
 #include "provider.hpp"
 
@@ -217,10 +218,10 @@ TEST_F(umfIpcTest, ConcurrentGetPutHandles) {
 
     std::array<std::vector<umf_ipc_handle_t>, NTHREADS> ipcHandles;
 
-    auto getHandlesFn = [&ipcHandles, &ptrs](size_t tid) {
-        // TODO: better to wait on the barrier here so that every thread
-        // starts at the same point. But std::barrier is available only
-        // starting from C++20
+    umf_test::syncthreads_barrier syncthreads(NTHREADS);
+
+    auto getHandlesFn = [&ipcHandles, &ptrs, &syncthreads](size_t tid) {
+        syncthreads();
         for (void *ptr : ptrs) {
             umf_ipc_handle_t ipcHandle;
             size_t handleSize;
@@ -230,31 +231,17 @@ TEST_F(umfIpcTest, ConcurrentGetPutHandles) {
         }
     };
 
-    std::vector<std::thread> threads;
-    for (int i = 0; i < NTHREADS; i++) {
-        threads.emplace_back(getHandlesFn, i);
-    }
-
-    for (auto &thread : threads) {
-        thread.join();
-    }
-    threads.clear();
+    umf_test::parallel_exec(NTHREADS, getHandlesFn);
 
-    auto putHandlesFn = [&ipcHandles](size_t tid) {
+    auto putHandlesFn = [&ipcHandles, &syncthreads](size_t tid) {
+        syncthreads();
         for (umf_ipc_handle_t ipcHandle : ipcHandles[tid]) {
             umf_result_t ret = umfPutIPCHandle(ipcHandle);
             EXPECT_EQ(ret, UMF_RESULT_SUCCESS);
         }
     };
 
-    for (int i = 0; i < NTHREADS; i++) {
-        threads.emplace_back(putHandlesFn, i);
-    }
-
-    for (auto &thread : threads) {
-        thread.join();
-    }
-    threads.clear();
+    umf_test::parallel_exec(NTHREADS, putHandlesFn);
 
     for (void *ptr : ptrs) {
         umf_result_t ret = umfPoolFree(pool.get(), ptr);
@@ -287,10 +274,11 @@ TEST_F(umfIpcTest, ConcurrentOpenCloseHandles) {
 
     std::array<std::vector<void *>, NTHREADS> openedIpcHandles;
 
-    auto openHandlesFn = [this, &ipcHandles, &openedIpcHandles](size_t tid) {
-        // TODO: better to wait on the barrier here so that every thread
-        // starts at the same point. But std::barrier is available only
-        // starting from C++20
+    umf_test::syncthreads_barrier syncthreads(NTHREADS);
+
+    auto openHandlesFn = [this, &ipcHandles, &openedIpcHandles,
+                          &syncthreads](size_t tid) {
+        syncthreads();
         for (auto ipcHandle : ipcHandles) {
             void *ptr;
             umf_result_t ret = umfOpenIPCHandle(pool.get(), ipcHandle, &ptr);
@@ -299,31 +287,17 @@ TEST_F(umfIpcTest, ConcurrentOpenCloseHandles) {
         }
     };
 
-    std::vector<std::thread> threads;
-    for (int i = 0; i < NTHREADS; i++) {
-        threads.emplace_back(openHandlesFn, i);
-    }
-
-    for (auto &thread : threads) {
-        thread.join();
-    }
-    threads.clear();
+    umf_test::parallel_exec(NTHREADS, openHandlesFn);
 
-    auto closeHandlesFn = [&openedIpcHandles](size_t tid) {
+    auto closeHandlesFn = [&openedIpcHandles, &syncthreads](size_t tid) {
+        syncthreads();
         for (void *ptr : openedIpcHandles[tid]) {
             umf_result_t ret = umfCloseIPCHandle(ptr);
             EXPECT_EQ(ret, UMF_RESULT_SUCCESS);
         }
     };
 
-    for (int i = 0; i < NTHREADS; i++) {
-        threads.emplace_back(closeHandlesFn, i);
-    }
-
-    for (auto &thread : threads) {
-        thread.join();
-    }
-    threads.clear();
+    umf_test::parallel_exec(NTHREADS, closeHandlesFn);
 
     for (auto ipcHandle : ipcHandles) {
         umf_result_t ret = umfPutIPCHandle(ipcHandle);

test/supp/drd-umf_test-ipc.supp

@@ -0,0 +1,7 @@
+{
+   Conditional variable destruction false-positive
+   drd:CondErr
+   ...
+   fun:pthread_cond_destroy@*
+   ...
+}