Merge pull request #1062 from igchor/fix_parallel_xexec

Make syncthreads() in parallel_xexec multi-use

Author: igchor

tests/CMakeLists.txt

@@ -149,6 +149,9 @@ else()
 	skip_test("aggregate_initialization" "SKIPPED_BECAUSE_OF_NO_COMPILER_SUPPORT")
 endif()
 
+build_test(helpers_test common/helpers_test.cpp)
+add_test_generic(NAME helpers_test TRACERS none memcheck)
+
 build_test(allocator allocator/allocator.cpp)
 add_test_generic(NAME allocator TRACERS none memcheck pmemcheck)
 

tests/common/helpers_test.cpp

@@ -0,0 +1,44 @@
+// SPDX-License-Identifier: BSD-3-Clause
+/* Copyright 2021, Intel Corporation */
+
+/*
+ * helpers_test - test for all helper classes/functions used in testing
+ * framework
+ */
+
+#include "thread_helpers.hpp"
+#include "unittest.hpp"
+
+#include <atomic>
+
+static size_t const concurrency = 4;
+
+/* Verify if syncthreads is working correctly */
+static void
+test()
+{
+	std::atomic<int> counter;
+	counter = 0;
+
+	parallel_xexec(concurrency,
+		       [&](size_t id, std::function<void()> syncthreads) {
+			       counter++;
+
+			       syncthreads();
+			       UT_ASSERTeq(counter.load(), concurrency);
+			       syncthreads();
+
+			       counter++;
+
+			       syncthreads();
+			       UT_ASSERTeq(counter.load(), concurrency * 2);
+			       syncthreads();
+			       UT_ASSERTeq(counter.load(), concurrency * 2);
+		       });
+}
+
+int
+main()
+{
+	return run_test(test);
+}

tests/common/thread_helpers.hpp

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: BSD-3-Clause
-/* Copyright 2020, Intel Corporation */
+/* Copyright 2020-2021, Intel Corporation */
 #ifndef THREAD_HELPERS_COMMON_HPP
 #define THREAD_HELPERS_COMMON_HPP
 
@@ -25,31 +25,58 @@ parallel_exec(size_t concurrency, Function f)
 	}
 }
 
+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;
+};
+
 /*
  * This function executes 'concurrency' threads and provides
- * 'syncthreads' method (synchronization barrier) for f()
+ * 'syncthreads' method (multi-use synchronization barrier) for f()
  */
 template <typename Function>
 void
 parallel_xexec(size_t concurrency, Function f)
 {
-	std::condition_variable cv;
 	std::mutex m;
-	std::unique_ptr<size_t> counter =
-		std::unique_ptr<size_t>(new size_t(0));
+	std::shared_ptr<latch> current_latch =
+		std::shared_ptr<latch>(new latch(concurrency));
 
+	/* 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. */
 	auto syncthreads = [&] {
 		std::unique_lock<std::mutex> lock(m);
-		(*counter)++;
-		if (*counter < concurrency)
-			cv.wait(lock, [&] { return *counter >= concurrency; });
-		else
-			/*
-			 * notify_call could be called outside of a lock
-			 * (it would perform better) but drd complains
-			 * in that case
-			 */
-			cv.notify_all();
+		auto l = current_latch;
+		if (l->wait(lock))
+			current_latch =
+				std::shared_ptr<latch>(new latch(concurrency));
 	};
 
 	parallel_exec(concurrency, [&](size_t tid) { f(tid, syncthreads); });

tests/drd.supp

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

tests/helpers.cmake

@@ -110,7 +110,7 @@ function(execute_common expect_success output_file name)
         set(TRACE valgrind --error-exitcode=99 --tool=helgrind)
         set(ENV{LIBPMEMOBJ_CPP_TRACER_HELGRIND} 1)
     elseif(${TRACER} STREQUAL drd)
-        set(TRACE valgrind --error-exitcode=99 --tool=drd)
+        set(TRACE valgrind --error-exitcode=99 --tool=drd --suppressions=${TEST_ROOT_DIR}/drd.supp)
         set(ENV{LIBPMEMOBJ_CPP_TRACER_DRD} 1)
     elseif(${TRACER} STREQUAL gdb)
         set(TRACE gdb --batch --command=${GDB_BATCH_FILE} --args)