Benchmarks: sync v streaming

.clang-format

+BasedOnStyle: LLVM
+AlignAfterOpenBracket: true
+AlignConsecutiveAssignments: AcrossComments
+AlignConsecutiveDeclarations: AcrossComments
+AlignConsecutiveMacros: AcrossComments
+AlignOperands: AlignAfterOperator
+AllowShortBlocksOnASingleLine: Always
+AllowShortCaseLabelsOnASingleLine: true
+AlwaysBreakAfterReturnType: AllDefinitions
+BinPackParameters: false
+BreakBeforeBinaryOperators: All
+BreakBeforeTernaryOperators: false
+BreakBeforeBraces: WebKit
+ColumnLimit: 120
+ConstructorInitializerIndentWidth: 2
+ContinuationIndentWidth: 2
+Cpp11BracedListStyle: true
+EmptyLinesBeforeAccessModifier: Always
+IndentPPDirectives: AfterHash
+IndentWidth: 4
+PenaltyBreakAssignment: 4000
+PointerAlignment: Left
+SpaceAfterCStyleCase: true
+UseTab: Never

.gitignore

+build/

CMakeLists.txt

+cmake_minimum_required(VERSION 3.18)
+project(streaming_producer CXX)
+
+add_compile_options(-Wall -Wextra -Wpedantic)
+include_directories(include)
+
+add_library(producer_bench_deps INTERFACE)
+target_compile_features(producer_bench_deps INTERFACE cxx_std_20)
+
+
+find_package(fmt REQUIRED CONFIG)
+target_link_libraries(producer_bench_deps INTERFACE fmt::fmt)
+
+find_package(benchmark REQUIRED)
+target_link_libraries(producer_bench_deps INTERFACE benchmark::benchmark)
+
+find_package(Threads REQUIRED)
+target_link_libraries(producer_bench_deps INTERFACE Threads::Threads)
+
+
+add_subdirectory(bench)
+add_subdirectory(app)
+
+#message("Compile features")
+#foreach( F ${CMAKE_CXX_COMPILE_FEATURES})
+#	message("\t${F}")
+#endforeach()
+

app/CMakeLists.txt

+add_executable(sample main.cpp)
+
+target_link_libraries(sample LINK_PUBLIC producer_bench_deps)

app/main.cpp

+#include <thread>
+#include "streaming_producer.hpp"
+
+static constexpr size_t n_items = 10;
+constexpr size_t produce_duration = 20;
+constexpr size_t consume_duration = 20;
+
+using ProdStream = Producer_Streaming<produce_duration>;
+int main()
+{
+	ProdStream prod_str{};
+	for (size_t i = 0; i < 100; i++){
+		printf("\nRUN #%d\n",i);
+		prod_str.reset();
+		std::thread t(&ProdStream::produce, &prod_str,n_items);
+		Item item;
+		do {
+		    item = prod_str.consume();
+		    printf("Item{%d}\n",item.id);
+		} while (item.id != -1);
+		t.join();
+	}
+
+	return 0;
+}

bench/CMakeLists.txt

+add_executable(bench main.cpp)
+target_link_libraries(bench LINK_PUBLIC producer_bench_deps)

bench/main.cpp

+#include <stdio.h>
+#include <iostream>
+
+#include <benchmark/benchmark.h>
+#include <fmt/core.h>
+
+#include "synchronous_producer.hpp"
+#include "streaming_producer.hpp"
+#include <chrono>
+#include <thread>
+
+
+static constexpr size_t n_items = 10;
+constexpr size_t produce_duration = 20;
+constexpr size_t consume_duration = 20;
+
+using ProdSync = Producer_Sync<produce_duration>;
+using ProdStream = Producer_Streaming<produce_duration>;
+
+static void sync_producer(benchmark::State& state) {
+    ProdSync prod_sync{};
+    for (auto _ : state) {
+	std::thread t(&ProdSync::produce, &prod_sync, n_items);
+	t.join();
+	Item item;
+	do {
+	    item = prod_sync.consume();
+	    std::this_thread::sleep_for(std::chrono::milliseconds{consume_duration});
+	} while (item.id != -1);
+    }
+}
+BENCHMARK(sync_producer)->Unit(benchmark::kMillisecond);
+
+static void stream_producer(benchmark::State& state) {
+    ProdStream prod_str{};
+    for (auto _ : state) {
+	prod_str.reset();
+	std::thread t(&ProdStream::produce, &prod_str,n_items);
+	Item item;
+	do {
+	    item = prod_str.consume();
+	    std::this_thread::sleep_for(std::chrono::milliseconds{consume_duration});
+	} while (item.id != -1);
+	t.join();
+    }
+}
+BENCHMARK(stream_producer)->Unit(benchmark::kMillisecond);
+
+BENCHMARK_MAIN();

include/items.hpp

+#ifndef ITEM_HPP
+#define ITEM_HPP
+
+struct Item {
+	int id;
+};
+#endif

include/main.cpp

+#include <stdio.h>
+#include <iostream>
+
+#include <benchmark/benchmark.h>
+#include <fmt/core.h>
+
+#include "synchronous_producer.hpp"
+#include "streaming_producer.hpp"
+#include <chrono>
+#include <thread>
+
+
+static constexpr size_t n_items = 10;
+constexpr size_t produce_duration = 20;
+constexpr size_t consume_duration = 20;
+
+using ProdSync = Producer_Sync<produce_duration>;
+using ProdStream = Producer_Streaming<produce_duration>;
+
+static void sync_producer(benchmark::State& state) {
+    ProdSync prod_sync{};
+    for (auto _ : state) {
+	std::thread t(&ProdSync::produce, &prod_sync, n_items);
+	t.join();
+	Item item;
+	do {
+	    item = prod_sync.consume();
+	} while (item.id != -1);
+
+    }
+	
+}
+BENCHMARK(sync_producer);
+
+static void stream_producer(benchmark::State& state) {
+    ProdStream prod_str{};
+    for (auto _ : state) {
+	std::thread t(&ProdStream::produce, &prod_str,n_items);
+	Item item;
+	do {
+		item = prod_str.consume();
+	} while (item.id != -1);
+	t.join();
+    }
+        
+
+}
+BENCHMARK(stream_producer);
+
+BENCHMARK_MAIN();

include/streaming_producer.hpp

+#include <mutex>
+#include <thread>
+#include <chrono>
+#include <deque>
+using namespace std::literals::chrono_literals;
+
+#include "items.hpp"
+
+template<size_t sleep_duration_ms>
+struct Producer_Streaming {
+	Item consume(){
+		Item ret;
+		
+		while(!done && channel.empty()){
+			std::this_thread::sleep_for(10ms);
+		}
+		if (done && channel.empty()){
+			return Item{-1};
+		}
+		m.lock();
+		ret = channel.front();
+		channel.pop_front();
+		m.unlock();
+		return ret;
+	};
+
+	void produce(size_t n_times)
+	{
+		done = false;
+		for (size_t i = 0; i < n_times; i++){
+			std::this_thread::sleep_for(std::chrono::milliseconds{sleep_duration_ms});
+			m.lock();
+			channel.push_back(Item{(int)i});
+			m.unlock();
+		}
+		done = true;
+	}
+	void reset(){
+		done = false;
+	}
+private:
+	bool done = false;
+	std::mutex m;
+	std::deque<Item> channel;
+};

include/synchronous_producer.hpp

+#include <mutex>
+#include <thread>
+#include <chrono>
+#include <deque>
+using namespace std::literals::chrono_literals;
+
+#include "items.hpp"
+
+template <size_t sleep_duration_ms>
+struct Producer_Sync {
+	Item consume()
+	{
+		while (!done){
+			std::this_thread::sleep_for(10ms);
+		}
+		if (channel.empty()){
+			return Item{-1};
+		}
+		auto ret = channel.front();
+		channel.pop_front();
+		return ret;
+
+	};
+
+	void produce(size_t n_times)
+	{
+		done = false;
+		for (size_t i = 0; i < n_times; i++){
+			std::this_thread::sleep_for(std::chrono::milliseconds{sleep_duration_ms});
+			channel.push_back(Item{(int)i});
+		}
+		done = true;
+
+	}
+private:
+	bool done = false;
+	std::mutex m{};
+	std::deque<Item> channel;
+};