✨ Add test concurrency policy

Problem:
- In order to find threading bugs more easily, TSan must have some
  non-determinism to work with.

Solution:
- Introduce the test concurrency policy, which adds random-duration sleeps on
  entering a critical section.

Author: elbeno

include/conc/test.hpp

@@ -0,0 +1,73 @@
+#pragma once
+
+#include <conc/concurrency.hpp>
+
+#if __STDC_HOSTED__ == 0
+#error conc::test_policy is designed for desktop testing and requires a hosted implementation
+#endif
+
+#include <algorithm>
+#include <array>
+#include <functional>
+#include <iterator>
+#include <mutex>
+#include <random>
+#include <thread>
+#include <utility>
+
+namespace conc {
+class test_policy {
+    template <typename> static inline std::mutex m{};
+
+    [[maybe_unused]] static auto get_rng() -> auto & {
+        thread_local auto rng = [] {
+            std::array<int, std::mt19937::state_size> seed_data;
+            std::random_device r;
+            std::generate_n(seed_data.data(), seed_data.size(), std::ref(r));
+            std::seed_seq seq(std::begin(seed_data), std::end(seed_data));
+            return std::mt19937{seq};
+        }();
+        return rng;
+    }
+
+    template <typename Uniq> struct [[nodiscard]] cs_raii_t {
+        cs_raii_t() {
+            m<Uniq>.lock();
+            ++lock_count;
+        }
+        ~cs_raii_t() {
+            ++unlock_count;
+            m<Uniq>.unlock();
+        }
+    };
+
+  public:
+    static inline std::atomic<std::uint32_t> lock_count{};
+    static inline std::atomic<std::uint32_t> unlock_count{};
+
+    static auto reset_counts() {
+        lock_count = 0;
+        unlock_count = 0;
+    }
+
+    template <typename Uniq = void, std::invocable F, std::predicate... Pred>
+        requires(sizeof...(Pred) < 2)
+    static inline auto call_in_critical_section(F &&f, Pred &&...pred)
+        -> decltype(std::forward<F>(f)()) {
+        while (true) {
+            std::uniform_int_distribution<> dis{5, 10};
+            auto const d1 = std::chrono::milliseconds{dis(get_rng())};
+            auto const d2 = std::chrono::milliseconds{dis(get_rng())};
+            std::this_thread::sleep_for(d1);
+
+            [[maybe_unused]] cs_raii_t<Uniq> lock{};
+            std::this_thread::sleep_for(d2);
+            if ((... and pred())) {
+                return std::forward<F>(f)();
+            }
+        }
+    }
+};
+
+template <> inline auto injected_policy<> = test_policy{};
+} // namespace conc

test/CMakeLists.txt

@@ -22,11 +22,13 @@ if(${CMAKE_CXX_STANDARD} GREATER_EQUAL 20)
     add_tests(
         FILES
         conc_standard_policy
+        conc_test_policy
         concepts
         freestanding_conc_injected_policy
         hosted_conc_injected_policy
         MULL_EXCLUSIONS
-        conc_standard_policy)
+        conc_standard_policy
+        conc_test_policy)
 
     add_compile_fail_test(fail_no_conc_policy.cpp LIBRARIES concurrency)
 endif()

test/conc_test_policy.cpp

@@ -0,0 +1,80 @@
+#include <conc/test.hpp>
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <functional>
+#include <iterator>
+#include <random>
+#include <thread>
+
+TEST_CASE("test policy allows 'recursive' critical_sections", "[test_policy]") {
+    auto const value = conc::call_in_critical_section(
+        [] { return conc::call_in_critical_section([] { return 1; }); });
+    CHECK(value == 1);
+}
+
+namespace {
+struct rng_CS;
+struct count_CS;
+
+auto get_rng() -> auto & {
+    std::array<int, std::mt19937::state_size> seed_data;
+    std::random_device r;
+    std::generate_n(seed_data.data(), seed_data.size(), std::ref(r));
+    std::seed_seq seq(std::begin(seed_data), std::end(seed_data));
+    static std::mt19937 rng(seq);
+    return rng;
+}
+} // namespace
+
+TEST_CASE("test policy works", "[test_policy]") {
+    conc::test_policy::reset_counts();
+    constexpr auto N = 10u;
+    auto &rng = get_rng();
+    auto count = 0;
+    auto dis = std::uniform_int_distribution{1, 10};
+
+    std::array<std::thread, N> threads{};
+    for (auto i = 0u; i < N; ++i) {
+        threads[i] = std::thread{[&] {
+            auto const d = conc::call_in_critical_section(
+                [&] { return std::chrono::milliseconds{dis(rng)}; });
+            std::this_thread::sleep_for(d);
+            conc::call_in_critical_section([&] { ++count; });
+        }};
+    }
+    for (auto i = 0u; i < N; ++i) {
+        threads[i].join();
+    }
+
+    CHECK(count == N);
+    CHECK(conc::test_policy::lock_count == N * 2);
+    CHECK(conc::test_policy::unlock_count == N * 2);
+}
+
+TEST_CASE("test policy allows different-ID critical_sections",
+          "[test_policy]") {
+    auto &rng = get_rng();
+    auto count = 0;
+    auto dis = std::uniform_int_distribution{1, 10};
+
+    auto t1 = std::thread{[&] {
+        auto const d = conc::call_in_critical_section<rng_CS>(
+            [&] { return std::chrono::milliseconds{dis(rng)}; });
+        std::this_thread::sleep_for(d);
+        conc::call_in_critical_section<count_CS>([&] { ++count; });
+    }};
+    auto t2 = std::thread{[&] {
+        auto const d = conc::call_in_critical_section<rng_CS>(
+            [&] { return std::chrono::milliseconds{dis(rng)}; });
+        std::this_thread::sleep_for(d);
+        conc::call_in_critical_section<count_CS>([&] { ++count; });
+    }};
+    t1.join();
+    t2.join();
+
+    CHECK(count == 2);
+}