Add mpsc_queue basic example

Author: Pawel Karczewski

examples/CMakeLists.txt

@@ -173,3 +173,5 @@ add_example(make_persistent make_persistent/make_persistent.cpp)
 add_example(persistent persistent/persistent.cpp)
 
 add_example(transaction transaction/transaction.cpp)
+
+add_example(mpsc_queue mpsc_queue/mpsc_queue.cpp)

examples/mpsc_queue/CMakeLists.txt

@@ -0,0 +1,21 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright 2021, Intel Corporation
+
+cmake_minimum_required(VERSION 3.3)
+project(mpsc_queue CXX)
+
+set(CXX_STANDARD_REQUIRED ON)
+set(CMAKE_CXX_STANDARD 11)
+
+find_package(PkgConfig QUIET)
+if(PKG_CONFIG_FOUND)
+	pkg_check_modules(LIBPMEMOBJ++ REQUIRED libpmemobj++)
+else()
+	find_package(LIBPMEMOBJ++ REQUIRED)
+endif()
+
+link_directories(${LIBPMEMOBJ++_LIBRARY_DIRS})
+
+add_executable(mpsc_queue.cpp)
+target_include_directories(mpsc_queue PUBLIC ${LIBPMEMOBJ++_INCLUDE_DIRS} . ..)
+target_link_libraries(mpsc_queue ${LIBPMEMOBJ++_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})

examples/mpsc_queue/mpsc_queue.cpp

@@ -0,0 +1,122 @@
+// SPDX-License-Identifier: BSD-3-Clause
+/* Copyright 2021, Intel Corporation */
+
+/**
+ * mpsc_queue.cpp -- example which shows how to use
+ * pmem::obj::experimental::mpsc_queue
+ */
+
+#include <libpmemobj++/experimental/mpsc_queue.hpp>
+#include <libpmemobj++/make_persistent.hpp>
+#include <libpmemobj++/persistent_ptr.hpp>
+#include <libpmemobj++/transaction.hpp>
+
+#include <iostream>
+#include <string>
+
+void
+show_usage(char *argv[])
+{
+	std::cerr << "usage: " << argv[0] << " file-name" << std::endl;
+}
+
+//! [mpsc_queue_single_threaded_example]
+
+struct root {
+	pmem::obj::persistent_ptr<
+		pmem::obj::experimental::mpsc_queue::pmem_log_type>
+		log;
+};
+
+void
+single_threaded(pmem::obj::pool<root> pop)
+{
+
+	std::vector<std::string> values_to_produce = {"xxx", "aaaaaaa", "bbbbb",
+						      "cccc", "ddddddddddd"};
+	pmem::obj::persistent_ptr<root> proot = pop.root();
+
+	/* Create mpsc_queue, which uses pmem_log_type object to store
+	 * data. */
+	auto queue = pmem::obj::experimental::mpsc_queue(*proot->log, 1);
+
+	/* Consume data, which was stored in the queue in the previous run of
+	 * the application. */
+	//! [try_consume_batch]
+	queue.try_consume_batch(
+		[&](pmem::obj::experimental::mpsc_queue::batch_type rd_acc) {
+			for (pmem::obj::string_view str : rd_acc) {
+				std::cout << std::string(str.data(), str.size())
+					  << std::endl;
+			}
+		});
+	//! [try_consume_batch]
+	/* Produce and consume data. */
+	//! [register_worker]
+	pmem::obj::experimental::mpsc_queue::worker worker =
+		queue.register_worker();
+	//! [register_worker]
+	for (std::string &value : values_to_produce) {
+		//! [try_produce]
+		/* Produce data. */
+		worker.try_produce(value);
+		//! [try_produce]
+		/* Consume produced data. */
+		queue.try_consume_batch(
+			[&](pmem::obj::experimental::mpsc_queue::batch_type
+				    rd_acc) {
+				for (pmem::obj::string_view str : rd_acc) {
+					std::cout << std::string(str.data(),
+								 str.size())
+						  << std::endl;
+				}
+			});
+	}
+	//! [try_produce_string_view]
+	/* Produce data to be consumed in next run of the application. */
+	worker.try_produce("Left for next run");
+	//! [try_produce_string_view]
+}
+
+//! [mpsc_queue_single_threaded_example]
+
+int
+main(int argc, char *argv[])
+{
+	if (argc < 2) {
+		show_usage(argv);
+		return 1;
+	}
+
+	const char *path = argv[1];
+
+	static constexpr size_t QUEUE_SIZE = 1000;
+
+	pmem::obj::pool<root> pop;
+	try {
+		pop = pmem::obj::pool<root>::open(path, "mpsc_queue");
+		if (pop.root()->log == nullptr) {
+			pmem::obj::transaction::run(pop, [&] {
+				pop.root()->log = pmem::obj::make_persistent<
+					pmem::obj::experimental::mpsc_queue::
+						pmem_log_type>(QUEUE_SIZE);
+			});
+		}
+		single_threaded(pop);
+
+	} catch (pmem::pool_error &e) {
+		std::cerr << e.what() << std::endl;
+		std::cerr
+			<< "To create pool run: pmempool create obj --layout=mpsc_queue -s 100M path_to_pool"
+			<< std::endl;
+	} catch (std::exception &e) {
+		std::cerr << e.what() << std::endl;
+	}
+
+	try {
+		pop.close();
+	} catch (const std::logic_error &e) {
+		std::cerr << e.what() << std::endl;
+	}
+	return 0;
+}