Merge pull request #1125 from karczex/mpsc_queue_remove_overload_with_tests

Mpsc queue remove overload with copying data in lambda function

Author: lukaszstolarczuk

include/libpmemobj++/experimental/mpsc_queue.hpp

@@ -9,7 +9,6 @@
 #include <libpmemobj++/detail/ringbuf.hpp>
 #include <libpmemobj++/make_persistent.hpp>
 #include <libpmemobj++/persistent_ptr.hpp>
-#include <libpmemobj++/slice.hpp>
 #include <libpmemobj++/string_view.hpp>
 #include <libpmemobj++/transaction.hpp>
 
@@ -117,9 +116,6 @@ class mpsc_queue {
 		worker(worker &&other);
 		worker &operator=(worker &&other);
 
-		template <typename Function>
-		bool try_produce(size_t size, Function &&f);
-
 		template <typename Function = void (*)(pmem::obj::string_view)>
 		bool try_produce(
 			pmem::obj::string_view data,
@@ -427,46 +423,6 @@ inline mpsc_queue::worker::~worker()
 	}
 }
 
-/**
- * @param f cannot fail. Any exception thrown from f will result
- * in terminate().
- */
-template <typename Function>
-inline bool
-mpsc_queue::worker::try_produce(size_t size, Function &&f)
-{
-	auto data = std::unique_ptr<char[]>(new char[size]);
-	auto range = pmem::obj::slice<char *>(data.get(), data.get() + size);
-
-	auto req_size = pmem::detail::align_up(size + sizeof(first_block::size),
-					       pmem::detail::CACHELINE_SIZE);
-	auto offset = acquire_cachelines(req_size);
-
-#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
-	ANNOTATE_HAPPENS_AFTER(queue->ring_buffer.get());
-#endif
-
-	if (offset == -1)
-		return false;
-
-	try {
-		f(range);
-	} catch (...) {
-		std::terminate();
-	}
-
-	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
-
-	produce_cachelines();
-
-	return true;
-}
-
 template <typename Function>
 bool
 mpsc_queue::worker::try_produce(pmem::obj::string_view data,

tests/mpsc_queue/basic.cpp

@@ -49,11 +49,7 @@ basic_test(pmem::obj::pool<root> pop, bool create)
 
 		/* Insert the data */
 		for (const auto &e : values) {
-			auto ret = worker.try_produce(
-				e.size(), [&](pmem::obj::slice<char *> range) {
-					std::copy_n(e.begin(), e.size(),
-						    range.begin());
-				});
+			auto ret = worker.try_produce(e);
 			UT_ASSERT(ret);
 		}
 
@@ -77,13 +73,7 @@ basic_test(pmem::obj::pool<root> pop, bool create)
 
 		/* Insert new data, which may be recovered in next run of
 		 * application */
-		ret = worker.try_produce(
-			store_to_next_run.size(),
-			[&](pmem::obj::slice<char *> range) {
-				std::copy_n(store_to_next_run.begin(),
-					    store_to_next_run.size(),
-					    range.begin());
-			});
+		ret = worker.try_produce(store_to_next_run);
 		UT_ASSERT(ret);
 	} else {
 		std::vector<std::string> values_on_pmem;

tests/mpsc_queue/consume_interrupt.cpp

@@ -63,11 +63,7 @@ consume_interrupt(pmem::obj::pool<root> pop, bool create)
 
 		/* Insert the data */
 		for (const auto &e : values) {
-			ret = worker.try_produce(
-				e.size(), [&](pmem::obj::slice<char *> range) {
-					std::copy_n(e.begin(), e.size(),
-						    range.begin());
-				});
+			ret = worker.try_produce(e);
 			UT_ASSERT(ret);
 		}
 

tests/mpsc_queue/consume_multipass.cpp

@@ -45,10 +45,7 @@ consume_multipass(pmem::obj::pool<root> pop, size_t n_iters)
 
 	/* Insert the data */
 	for (const auto &e : values) {
-		auto ret = worker.try_produce(
-			e.size(), [&](pmem::obj::slice<char *> range) {
-				std::copy_n(e.begin(), e.size(), range.begin());
-			});
+		auto ret = worker.try_produce(e);
 		UT_ASSERT(ret);
 	}
 

tests/mpsc_queue/empty.cpp

@@ -51,15 +51,14 @@ consume_empty_after_insertion(pmem::obj::pool<root> pop)
 		[&](queue_type::batch_type rd_acc) { ASSERT_UNREACHABLE; });
 	UT_ASSERTeq(consumed, false);
 
-	std::vector<std::string> values = {"xxx", "aaaaaaa", "bbbbb"};
+	std::vector<std::string> values = {
+		"xxx", "aaaaaaa", "bbbbb",
+		std::string(QUEUE_SIZE / 2 - 1, 'a')};
 
 	auto worker = queue.register_worker();
 	/* Insert some data */
 	for (const auto &e : values) {
-		auto ret = worker.try_produce(
-			e.size(), [&](pmem::obj::slice<char *> range) {
-				std::copy_n(e.begin(), e.size(), range.begin());
-			});
+		auto ret = worker.try_produce(e);
 		UT_ASSERT(ret);
 	}
 	/* Consume all of it */

tests/mpsc_queue/mt.cpp

@@ -73,14 +73,14 @@ mt_test(pmem::obj::pool<root> pop, size_t concurrency)
 				bool insert_succeed = false;
 				while (!insert_succeed) {
 					insert_succeed = worker.try_produce(
-						e.size(),
-						[&](pmem::obj::slice<char *>
-							    range) {
+						e,
+						[&](pmem::obj::string_view
+							    target) {
+							UT_ASSERT(
+								pmem::obj::string_view(
+									e) ==
+								target);
 							x++;
-							std::copy_n(
-								e.begin(),
-								e.size(),
-								range.begin());
 						});
 				};
 			}

tests/mpsc_queue/pmreorder/recovery.cpp

@@ -31,7 +31,7 @@ struct root {
 static constexpr size_t QUEUE_SIZE = 10000;
 static const auto produce_size = 128ULL;
 static const size_t concurrency = 4;
-static const auto fill_pattern = char(1);
+static const auto fill_pattern = std::string(produce_size, 'z');
 
 /* Break application during produce. */
 static void
@@ -54,8 +54,8 @@ run_consistent(pmem::obj::pool<root> pop, bool break_produce, bool synchronized)
 			if (id == 0 && break_produce)
 				VALGRIND_PMC_EMIT_LOG("PMREORDER_MARKER.BEGIN");
 
-			worker.try_produce(produce_size,
-					   [&](pmem::obj::slice<char *> range) {
+			worker.try_produce(fill_pattern,
+					   [&](pmem::obj::string_view v) {
 						   if (synchronized) {
 							   /* Make sure that all
 							    * other threads
@@ -64,10 +64,6 @@ run_consistent(pmem::obj::pool<root> pop, bool break_produce, bool synchronized)
 							    */
 							   syncthreads();
 						   }
-
-						   std::fill_n(range.begin(),
-							       produce_size,
-							       fill_pattern);
 					   });
 
 			if (id == 0 && break_produce)
@@ -102,7 +98,7 @@ check_consistency(pmem::obj::pool<root> pop, bool already_consumed)
 		UT_ASSERT(values_on_pmem.size() >= expected);
 
 	for (auto &str : values_on_pmem) {
-		UT_ASSERT(str == std::string(produce_size, fill_pattern));
+		UT_ASSERT(str == fill_pattern);
 	}
 }
 
@@ -131,7 +127,7 @@ run_break_recovery(pmem::obj::pool<root> pop)
 	UT_ASSERT(values_on_pmem.size() >= expected);
 
 	for (auto &str : values_on_pmem) {
-		UT_ASSERT(str == std::string(produce_size, fill_pattern));
+		UT_ASSERT(str == fill_pattern);
 	}
 }
 

tests/mpsc_queue/pmreorder/recovery_after_consume.cpp

@@ -31,15 +31,11 @@ struct root {
 static constexpr size_t QUEUE_SIZE = 3000;
 static const auto produce_size = 128ULL;
 static const size_t concurrency = 4;
-static const auto fill_pattern = char(1);
+const auto fill_pattern = std::string(produce_size, 'x');
 
 void
 run_consistent(pmem::obj::pool<root> pop)
 {
-	const auto produce_size = 128ULL;
-	const size_t concurrency = 4;
-	const auto fill_pattern = char(1);
-
 	auto proot = pop.root();
 	auto queue = queue_type(*proot->log, concurrency);
 
@@ -56,12 +52,7 @@ run_consistent(pmem::obj::pool<root> pop)
 	proot->capacity = capacity;
 	pop.persist(proot->capacity);
 
-	make_queue_with_first_half_empty(queue, capacity, produce_size,
-					 [&](pmem::obj::slice<char *> range) {
-						 std::fill_n(range.begin(),
-							     produce_size, 1);
-					 });
-
+	make_queue_with_first_half_empty(queue, capacity, produce_size);
 	/* Run this under pmreorder. After crash state of the queue should be
 	 * something like this: | produced | crashed | produced | empty |
 	 * produced |
@@ -73,12 +64,7 @@ run_consistent(pmem::obj::pool<root> pop)
 			if (id == 0)
 				VALGRIND_PMC_EMIT_LOG("PMREORDER_MARKER.BEGIN");
 
-			auto ret = worker.try_produce(
-				produce_size,
-				[&](pmem::obj::slice<char *> range) {
-					std::fill_n(range.begin(), produce_size,
-						    fill_pattern);
-				});
+			auto ret = worker.try_produce(fill_pattern);
 
 			if (id == 0)
 				VALGRIND_PMC_EMIT_LOG("PMREORDER_MARKER.END");
@@ -111,20 +97,16 @@ check_consistency(pmem::obj::pool<root> pop)
 	UT_ASSERT(values_on_pmem.size() >= expected);
 
 	for (auto &str : values_on_pmem) {
-		UT_ASSERT(str == std::string(produce_size, fill_pattern));
+		UT_ASSERT(str == fill_pattern);
 	}
 
 	auto worker = queue.register_worker();
 
-	static const auto overwrite_pattern = char(2);
 	static const auto overwrite_size = 64;
+	static const auto overwrite_pattern = std::string(overwrite_size, 'y');
 
 	while (true) {
-		auto ret = worker.try_produce(
-			overwrite_size, [&](pmem::obj::slice<char *> range) {
-				std::fill_n(range.begin(), overwrite_size,
-					    overwrite_pattern);
-			});
+		auto ret = worker.try_produce(overwrite_pattern);
 
 		if (!ret)
 			break;
@@ -138,8 +120,7 @@ check_consistency(pmem::obj::pool<root> pop)
 	UT_ASSERT(ret);
 
 	for (auto &str : values_on_pmem) {
-		UT_ASSERT(str ==
-			  std::string(overwrite_size, overwrite_pattern));
+		UT_ASSERT(str == overwrite_pattern);
 	}
 }
 

tests/mpsc_queue/queue.hpp

@@ -7,18 +7,17 @@
 
 using queue_type = pmem::obj::experimental::mpsc_queue;
 
+/* Returns capacity in bytes */
 size_t
-get_queue_capacity(queue_type &q, size_t element_size)
+get_queue_capacity(queue_type &q, size_t element_size = 1)
 {
 	auto worker = q.register_worker();
-
+	auto element = std::string(element_size, 'b');
 	size_t capacity = 0;
-
 	/* Check how many elements fit in the log. */
-	while (worker.try_produce(
-		element_size,
-		[&](pmem::obj::slice<char *> range) { capacity++; }))
-		;
+	while (worker.try_produce(element)) {
+		capacity++;
+	}
 
 	/* Clear the queue */
 	auto ret = q.try_consume_batch([](queue_type::batch_type acc) {
@@ -30,26 +29,20 @@ get_queue_capacity(queue_type &q, size_t element_size)
 	return capacity;
 }
 
-template <typename F>
 void
 make_queue_with_first_half_empty(queue_type &q, size_t capacity,
-				 size_t element_size, F &&f)
+				 size_t element_size = 1)
 {
 	auto worker = q.register_worker();
-
+	auto element = std::string(element_size, 'x');
 	size_t produced = 0;
-
 	while (produced < capacity) {
 		/* Produce half of the elements, call consume and
 		 * produce the rest. This should result in log being
 		 * consumed at the
 		 * beginning and unconsumed at the end. */
-		auto ret = worker.try_produce(
-			element_size, [&](pmem::obj::slice<char *> range) {
-				f(range);
-				produced++;
-			});
-
+		auto ret = worker.try_produce(element);
+		produced++;
 		UT_ASSERT(ret);
 
 		if (produced == capacity / 2) {