concurrency_test_utils.H

Go to the documentation of this file.

/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
#ifndef ALEPH_TESTS_CONCURRENCY_TEST_UTILS_H
#define ALEPH_TESTS_CONCURRENCY_TEST_UTILS_H
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <initializer_list>
#include <mutex>
#include <random>
#include <system_error>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
 
#include <ah-errors.H>
 
namespace Aleph::Testing
{
  class Deterministic_Start_Gate
  {
    const size_t expected_ = 0;
    size_t arrived_ = 0;
    bool released_ = false;
    bool cancelled_ = false;
    mutable std::mutex mutex_;
    std::condition_variable cv_;
 
  public:
    explicit Deterministic_Start_Gate(const size_t expected)
      : expected_(expected)
    {
      ah_invalid_argument_if(expected == 0)
        << "Deterministic_Start_Gate requires at least one participant";
    }
 
    bool arrive_and_wait()
    {
      std::unique_lock lock(mutex_);
      ++arrived_;
      cv_.notify_all();
      cv_.wait(lock, [this] { return released_ or cancelled_; });
      return released_;
    }
 
    void wait_until_ready()
    {
      std::unique_lock lock(mutex_);
      cv_.wait(lock, [this] { return arrived_ == expected_ or cancelled_; });
    }
 
    void release()
    {
      std::lock_guard lock(mutex_);
      released_ = true;
      cv_.notify_all();
    }
 
    void cancel() noexcept
    {
      std::lock_guard lock(mutex_);
      cancelled_ = true;
      cv_.notify_all();
    }
 
    [[nodiscard]] size_t arrived() const
    {
      std::lock_guard lock(mutex_);
      return arrived_;
    }
  };
 
  namespace detail
  {
    class Joining_Thread_Guard
    {
      std::vector<std::thread> &threads_;
      Deterministic_Start_Gate &gate_;
      bool launch_completed_ = false;
 
    public:
      Joining_Thread_Guard(std::vector<std::thread> &threads,
                            Deterministic_Start_Gate &gate) noexcept
        : threads_(threads), gate_(gate)
      {}
 
      Joining_Thread_Guard(const Joining_Thread_Guard &) = delete;
      Joining_Thread_Guard &operator = (const Joining_Thread_Guard &) = delete;
 
      void dismiss() noexcept { launch_completed_ = true; }
 
      ~Joining_Thread_Guard()
      {
        if (not launch_completed_)
          gate_.cancel();
        for (auto &thread : threads_)
          if (thread.joinable())
            {
              try
                {
                  thread.join();
                }
              catch (const std::system_error &)
                {
                  try
                    {
                      thread.detach();
                    }
                  catch (const std::system_error &)
                    {
                      // Nothing actionable left from a destructor.
                    }
                }
            }
      }
    };
  } // namespace detail
 
  template <typename Worker>
  void run_workers(const size_t thread_count, Worker worker)
  {
    Deterministic_Start_Gate gate(thread_count);
    std::vector<std::thread> threads;
    threads.reserve(thread_count);
    std::mutex exception_mutex;
    std::exception_ptr first_exception;
 
    {
      detail::Joining_Thread_Guard guard(threads, gate);
 
      for (size_t i = 0; i < thread_count; ++i)
        threads.emplace_back([&, i, worker]() mutable
          {
            if (not gate.arrive_and_wait())
              return;
            try
              {
                worker(i);
              }
            catch (...)
              {
                std::lock_guard lock(exception_mutex);
                if (not first_exception)
                  first_exception = std::current_exception();
              }
          });
 
      guard.dismiss();  // every thread launched; let them reach the gate
                         // on their own instead of racing to cancel() it.
      gate.wait_until_ready();
      gate.release();
    } // guard destructor joins every thread (see Joining_Thread_Guard).
 
    if (first_exception)
      std::rethrow_exception(first_exception);
  }
 
  struct Producer_Consumer_Stress_Config
  {
    size_t producers = 1;
    size_t consumers = 1;
    size_t items_per_producer = 1024;
    std::chrono::milliseconds timeout = std::chrono::seconds(10);
  };
 
  struct Producer_Consumer_Stress_Result
  {
    std::vector<size_t> consumed;
    bool timed_out = false;
 
    [[nodiscard]] size_t size() const noexcept { return consumed.size(); }
  };
 
  namespace detail
  {
    template <typename Push>
    bool invoke_push(Push &push, const size_t value)
    {
      if constexpr (std::is_void_v<std::invoke_result_t<Push &, size_t>>)
        {
          push(value);
          return true;
        }
      else
        {
          return static_cast<bool>(push(value));
        }
    }
  }
 
  template <typename Push, typename TryPop>
  Producer_Consumer_Stress_Result run_producer_consumer_stress(
      const Producer_Consumer_Stress_Config &config,
      Push push,
      TryPop try_pop)
  {
    const size_t total = config.producers * config.items_per_producer;
    Producer_Consumer_Stress_Result result;
    result.consumed.resize(total);
 
    std::atomic<size_t> consumed_count{0};
    std::atomic<bool> timed_out{false};
    const auto deadline = std::chrono::steady_clock::now() + config.timeout;
 
    // Workers `[0, producers)` produce; `[producers, producers+consumers)`
    // consume. `push`/`try_pop` are captured by value here once, then
    // copied again per-thread by run_workers() itself (each thread's copy
    // of this worker closure carries its own copy of both callables).
    run_workers(config.producers + config.consumers,
      [&, push, try_pop](const size_t index) mutable
      {
        if (index < config.producers)
          {
            const size_t p = index;
            for (size_t i = 0; i < config.items_per_producer; ++i)
              {
                const size_t value = p * config.items_per_producer + i;
                while (not detail::invoke_push(push, value))
                  {
                    if (std::chrono::steady_clock::now() >= deadline)
                      {
                        timed_out.store(true, std::memory_order_relaxed);
                        return;
                      }
                    std::this_thread::yield();
                  }
              }
          }
        else
          {
            while (consumed_count.load(std::memory_order_acquire) < total)
              {
                size_t value = 0;
                if (try_pop(value))
                  {
                    const size_t slot =
                        consumed_count.fetch_add(1, std::memory_order_acq_rel);
                    if (slot < total)
                      result.consumed[slot] = value;
                  }
                else
                  {
                    if (std::chrono::steady_clock::now() >= deadline)
                      {
                        timed_out.store(true, std::memory_order_relaxed);
                        return;
                      }
                    std::this_thread::yield();
                  }
              }
          }
      });
 
    const size_t n = consumed_count.load(std::memory_order_acquire);
    if (n < result.consumed.size())
      result.consumed.resize(n);
    result.timed_out = timed_out.load(std::memory_order_relaxed);
    return result;
  }
 
  enum class Trace_Operation_Kind
  {
    push,
    pop,
    insert,
    erase,
    contains
  };
 
  struct Trace_Operation
  {
    Trace_Operation_Kind kind = Trace_Operation_Kind::contains;
    size_t key = 0;
    size_t value = 0;
 
    [[nodiscard]] bool operator == (const Trace_Operation &rhs) const noexcept
    {
      return kind == rhs.kind and key == rhs.key and value == rhs.value;
    }
  };
 
  inline std::vector<Trace_Operation> make_random_operation_trace(
      const size_t count,
      const uint32_t seed,
      const size_t key_range,
      const std::initializer_list<Trace_Operation_Kind> kinds =
        {
          Trace_Operation_Kind::insert,
          Trace_Operation_Kind::erase,
          Trace_Operation_Kind::contains
        })
  {
    std::vector<Trace_Operation_Kind> allowed(kinds.begin(), kinds.end());
    ah_invalid_argument_if(key_range == 0) << "key_range must be > 0";
    ah_invalid_argument_if(allowed.empty()) << "kinds must not be empty";
 
    std::mt19937 rng(seed);
    std::uniform_int_distribution<size_t> kind_dist(0, allowed.size() - 1);
    std::uniform_int_distribution<size_t> key_dist(0, key_range - 1);
    std::uniform_int_distribution<size_t> value_dist;
 
    std::vector<Trace_Operation> trace;
    trace.reserve(count);
    for (size_t i = 0; i < count; ++i)
      trace.push_back(
        {
          allowed[kind_dist(rng)],
          key_dist(rng),
          value_dist(rng)
        });
 
    return trace;
  }
 
  template <typename SubjectStep, typename ReferenceStep>
  std::vector<size_t> replay_trace_and_collect_mismatches(
      const std::vector<Trace_Operation> &trace,
      SubjectStep subject_step,
      ReferenceStep reference_step)
  {
    std::vector<size_t> mismatches;
    for (size_t i = 0; i < trace.size(); ++i)
      if (not (subject_step(trace[i]) == reference_step(trace[i])))
        mismatches.push_back(i);
    return mismatches;
  }
} // namespace Aleph::Testing
 
#endif // ALEPH_TESTS_CONCURRENCY_TEST_UTILS_H