mpsc_queue: enbale drd/helgrind tests

Author: igchor

include/libpmemobj++/detail/ringbuf.hpp

@@ -115,6 +115,22 @@ struct ringbuf_t {
 		space = length;
 		end = RBUF_OFF_MAX;
 		nworkers = max_workers;
+
+		/* Helgrind/Drd does not understand std::atomic */
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+		VALGRIND_HG_DISABLE_CHECKING(&next, sizeof(next));
+		VALGRIND_HG_DISABLE_CHECKING(&end, sizeof(end));
+		VALGRIND_HG_DISABLE_CHECKING(&written, sizeof(written));
+
+		for (size_t i = 0; i < max_workers; i++) {
+			VALGRIND_HG_DISABLE_CHECKING(
+				&workers[i].seen_off,
+				sizeof(workers[i].seen_off));
+			VALGRIND_HG_DISABLE_CHECKING(
+				&workers[i].registered,
+				sizeof(workers[i].registered));
+		}
+#endif
 	}
 };
 

include/libpmemobj++/experimental/mpsc_queue.hpp

@@ -252,6 +252,11 @@ mpsc_queue::try_consume(Function &&f)
 
 	size_t offset;
 	size_t len = ringbuf_consume(ring_buffer.get(), &offset);
+
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	ANNOTATE_HAPPENS_AFTER(ring_buffer.get());
+#endif
+
 	if (len != 0) {
 		pmem->written = offset;
 		pop.persist(pmem->written);
@@ -264,6 +269,10 @@ mpsc_queue::try_consume(Function &&f)
 		if (elements_to_consume)
 			f(acc);
 
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+		ANNOTATE_HAPPENS_BEFORE(ring_buffer.get());
+#endif
+
 		ringbuf_release(ring_buffer.get(), len);
 
 		/* XXX: it would be better to call f once - hide
@@ -280,6 +289,13 @@ inline mpsc_queue::worker::worker(mpsc_queue *q)
 {
 	queue = q;
 	auto &manager = queue->get_id_manager();
+
+#if LIBPMEMOBJ_CPP_VG_DRD_ENABLED
+	ANNOTATE_BENIGN_RACE_SIZED(
+		&manager, sizeof(std::mutex),
+		"https://bugs.kde.org/show_bug.cgi?id=416286");
+#endif
+
 	id = manager.get();
 
 	assert(id < q->ring_buffer->nworkers);
@@ -330,6 +346,10 @@ mpsc_queue::worker::try_produce(size_t size, Function &&f)
 					       pmem::detail::CACHELINE_SIZE);
 	auto offset = ringbuf_acquire(queue->ring_buffer.get(), w, req_size);
 
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	ANNOTATE_HAPPENS_AFTER(queue->ring_buffer.get());
+#endif
+
 	if (offset == -1)
 		return false;
 
@@ -342,6 +362,10 @@ mpsc_queue::worker::try_produce(size_t size, Function &&f)
 	store_to_log(pmem::obj::string_view(data.get(), size),
 		     queue->buf + offset);
 
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	ANNOTATE_HAPPENS_BEFORE(queue->ring_buffer.get());
+#endif
+
 	ringbuf_produce(queue->ring_buffer.get(), w);
 
 	return true;
@@ -355,11 +379,19 @@ mpsc_queue::worker::try_produce(pmem::obj::string_view data)
 				       pmem::detail::CACHELINE_SIZE);
 	auto offset = ringbuf_acquire(queue->ring_buffer.get(), w, req_size);
 
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	ANNOTATE_HAPPENS_AFTER(queue->ring_buffer.get());
+#endif
+
 	if (offset == -1)
 		return false;
 
 	store_to_log(data, queue->buf + offset);
 
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	ANNOTATE_HAPPENS_BEFORE(queue->ring_buffer.get());
+#endif
+
 	ringbuf_produce(queue->ring_buffer.get(), w);
 
 	return true;

tests/CMakeLists.txt

@@ -958,13 +958,12 @@ endif()
 ################################################################################
 #################################### MPSC_QUEUE ################################
 if(TEST_MPSC_QUEUE)
-	#Test for internal implementation of ringbuffer
+	# Test for internal implementation of ringbuffer
 	build_test(ringbuf mpsc_queue/ringbuf.cpp)
-	add_test_generic(NAME ringbuf TRACERS none memcheck)
+	add_test_generic(NAME ringbuf TRACERS none memcheck drd helgrind)
 
 	build_test(mpsc_queue_mt mpsc_queue/mt.cpp)
-	# XXX: fix memcheck pmemcheck
-	add_test_generic(NAME mpsc_queue_mt TRACERS none)
+	add_test_generic(NAME mpsc_queue_mt TRACERS none drd helgrind memcheck pmemcheck)
 
 	build_test(mpsc_queue_basic mpsc_queue/basic.cpp)
 	add_test_generic(NAME mpsc_queue_basic SCRIPT mpsc_queue/mpsc_queue_basic_0.cmake TRACERS none memcheck pmemcheck)

tests/mpsc_queue/mt.cpp

@@ -44,6 +44,9 @@ mt_test(pmem::obj::pool<root> pop, size_t concurrency)
 	std::vector<std::string> values = {"xxx", "aaaaaaa", "bbbbb", "cccc"};
 
 	std::atomic<size_t> threads_counter(concurrency);
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+	VALGRIND_HG_DISABLE_CHECKING(&threads_counter, sizeof(threads_counter));
+#endif
 
 	std::vector<std::string> values_on_pmem;
 	parallel_exec(concurrency + 1, [&](size_t thread_id) {
@@ -111,7 +114,9 @@ test(int argc, char *argv[])
 
 	const char *path = argv[1];
 
-	constexpr size_t concurrency = 16;
+	size_t concurrency = 48;
+	if (On_valgrind)
+		concurrency = 2;
 
 	pmem::obj::pool<struct root> pop;