Tests for the disruptor

.gitignore

@@ -1,2 +1,3 @@
 /cmake-build-*/
-/.idea/
+/.idea/
+/docs/

CMakeLists.txt

@@ -6,4 +6,4 @@ set(CMAKE_CXX_STANDARD_REQUIRED True)
 
 add_subdirectory(LibSnugLog)
 
-#add_subdirectory(Tests)
+add_subdirectory(Tests)

LibSnugLog/include/disruptor.h

@@ -5,6 +5,7 @@
 #include <new>
 #include <cassert>
 #include <optional>
+#include <iostream>
 
 #ifdef __cpp_lib_hardware_interference_size
 static constexpr size_t max_interference_size = std::max(std::hardware_constructive_interference_size, std::hardware_destructive_interference_size);
@@ -30,7 +31,7 @@ enum class overflow_response_t {
 #ifdef __cpp_concepts
 template<typename T>
 concept OverflowStrategyType = requires (T strategy) {
-	{T::on_overflow} -> std::same_as<overflow_response_t>;
+	{T::on_overflow} -> std::same_as<k">const overflow_response_t&>;
 	{strategy.wait()};
 };
 #endif
@@ -88,7 +89,7 @@ public:
 	template<bool must_be_contiguous = false>
 	std::optional<token_t> try_advance(
 		std::atomic<offset_t>& lead,
-		std::atomic<offset_t> fence,
+		std::atomic<offset_t>& fence,
 		offset_t amount
 	) {
 		offset_t  old_offset = lead.load(std::memory_order_relaxed);
@@ -149,6 +150,7 @@ public:
 			if(tok) break;
 			waiter.wait();
 		}
+		// std::cout << tok.value().start << " rw " << tok.value().end << std::endl;
 		return tok.value();
 	}
 	token_t reserve_write(size_t sz, force_contiguous_mode) {
@@ -159,9 +161,13 @@ public:
 			if(tok) break;
 			waiter.wait();
 		}
+		// std::cout << tok.value().start << " rw " << tok.value().end << std::endl;
 		return tok.value();
 	}
-	void conclude_write(token_t tok) noexcept(std::is_nothrow_invocable_v<typename OverflowStrategy::wait>) {
+	
+	static constexpr const auto& tmp_fn = &OverflowStrategy::wait;
+	
+	void conclude_write(token_t tok) noexcept(std::is_nothrow_invocable_v<decltype(tmp_fn), OverflowStrategy>) {
 		OverflowStrategy waiter;
 		while(!write_trailer.compare_exchange_weak(tok.start, tok.end, std::memory_order_release, std::memory_order_relaxed)) {
 			waiter.wait();
@@ -177,21 +183,30 @@ public:
 		while(!read_lead.compare_exchange_weak(old_offset, new_offset, std::memory_order_acquire, std::memory_order_relaxed)) {
 			waiter.wait();
 		}
-		
+		// std::cout << old_offset << " rr " << new_offset << std::endl;
 		return token_t{old_offset, new_offset};
 	}
-	void conclude_read(token_t tok) noexcept(std::is_nothrow_invocable_v<kr">typename OverflowStrategy::wait>) {
+	void conclude_read(token_t tok) noexcept(std::is_nothrow_invocable_v<n">decltype(tmp_fn), OverflowStrategy>) {
 		OverflowStrategy waiter;
 		while(!read_trailer.compare_exchange_weak(tok.start, tok.end, std::memory_order_release, std::memory_order_relaxed)) {
 			waiter.wait();
 		}
+		// std::cout << tok.start << " cr " << tok.end << std::endl;
+	}
+	
+	char& operator[](size_t offset) {
+		return data_buffer[offset];
+	}
+	
+	[[nodiscard]] offset_t size() const {
+		return max_offset;
 	}
 };
 
 struct OverflowWait {
 	static constexpr overflow_response_t on_overflow = overflow_response_t::must_wait;
 	void wait() {
-#ifdef __clang__ or __GNUC__
+#if defined(__clang__) || defined(__GNUC__)
 		__asm__("nop\n\t");
 #elif _MSC_VER
 		__nop;

Tests/CMakeLists.txt

@@ -0,0 +1,17 @@
+Include(FetchContent)
+
+FetchContent_Declare(
+        Catch2
+        GIT_REPOSITORY https://github.com/catchorg/Catch2.git
+        GIT_TAG        v3.1.0
+)
+
+FetchContent_MakeAvailable(Catch2)
+
+add_executable(tests disruptor.cpp)
+target_link_libraries(tests PRIVATE LibSnugLog)
+target_link_libraries(tests PRIVATE Catch2::Catch2WithMain)
+
+include(CTest)
+include(Catch)
+catch_discover_tests(tests)

Tests/disruptor.cpp

@@ -0,0 +1,61 @@
+#include "catch2/catch_all.hpp"
+#include "../LibSnugLog/include/disruptor.h"
+
+struct strategy {
+	static constexpr overflow_response_t on_overflow = overflow_response_t::must_wait;
+	void wait() {}
+};
+
+TEST_CASE("Disruptor works sequentially") {
+	std::array<char, 8192> buffer{};
+	disruptor<strategy> v{buffer.data(), buffer.size()};
+
+	SECTION("117") {
+		auto W = v.reserve_write(100);
+		v[W.start] = 117;
+		v.conclude_write(W);
+		auto R = v.reserve_read();
+		REQUIRE(v[R.start]== 117);
+		v.conclude_read(R);
+	}
+
+	SECTION("12") {
+		{
+			auto W = v.reserve_write(6);
+			v[W.start] = 12;
+			v.conclude_write(W);
+			auto R = v.reserve_read();
+			REQUIRE(v[R.start]== 12);
+			v.conclude_read(R);
+		}
+
+		{
+			auto W = v.reserve_write(6);
+			v[W.start] = 8;
+			v.conclude_write(W);
+			auto R = v.reserve_read();
+			REQUIRE(v[R.start]== 8);
+			v.conclude_read(R);
+		}
+	}
+
+	SECTION("Disruptor loop around") {
+		std::multiset<char> mset;
+		for(int i = 0; i != 255; i++) {
+			auto W = v.reserve_write(100);
+			v[W.start] = (char)i;
+			for(size_t idx = W.start; idx != W.end; idx = (idx+1)%v.size()) {
+				v[idx] = (char)i;
+			}
+			v.conclude_write(W);
+			auto R = v.reserve_read();
+			for(size_t idx = W.start; idx != W.end; idx = (idx+1)%v.size()) {
+				mset.insert(v[idx]);
+			}
+			v.conclude_read(R);
+		}
+		for(int i = 0; i != 255; i++) {
+			REQUIRE(mset.count((char)i) == 100);
+		}
+	}
+}