chaos.hpp

Go to the documentation of this file.

#pragma once

 
#include "scheduler_policy.hpp"
#include "thread_registry.hpp"
#include "thread_wrapper.hpp"
#include "topology.hpp"
#include <atomic>
#include <chrono>
#include <future>
#include <random>
#include <thread>
 
namespace threadschedule
{

struct ChaosConfig
{
    std::chrono::milliseconds interval{250};

    int priority_jitter{0};

    bool shuffle_affinity{true};
};

class ChaosController
{
  public:
    template <typename Predicate>
    ChaosController(ChaosConfig cfg, Predicate pred) : config_(cfg), stop_(false)
    {
        std::promise<Tid> worker_started;
        auto worker_ready = worker_started.get_future();
 
        worker_ = ThreadWrapper([this, pred, started = std::move(worker_started)]() mutable {
            started.set_value(ThreadInfo::get_thread_id());
            run_loop(pred);
        });
 
        worker_tid_ = worker_ready.get();
        (void)ThreadInfo(worker_tid_).set_name("ts_chaos_ctl");
    }
 
    ~ChaosController()
    {
        stop_ = true;
        if (worker_.joinable())
            worker_.join();
    }
 
    ChaosController(ChaosController const&) = delete;
    auto operator=(ChaosController const&) -> ChaosController& = delete;
 
    [[nodiscard]] auto thread_info() const -> std::optional<ThreadInfo>
    {
        if (!worker_.joinable() || worker_tid_ == Tid{})
            return std::nullopt;
        return ThreadInfo(worker_tid_);
    }
 
    auto configure_thread(std::string const& name, SchedulingPolicy policy = SchedulingPolicy::OTHER,
                          ThreadPriority priority = ThreadPriority::normal()) -> expected<void, std::error_code>
    {
        auto info = thread_info();
        if (!info.has_value())
            return unexpected(std::make_error_code(std::errc::no_such_process));
        return detail::configure_thread(info.value(), name, policy, priority);
    }
 
  private:
    template <typename Predicate>
    void run_loop(Predicate pred)
    {
        std::mt19937 rng(std::random_device{}());
        while (!stop_)
        {
            registry().apply(pred, [&](RegisteredThreadInfo const& info) {
                auto blk = registry().get(info.tid);
                (void)blk;
            });
 
            // Affinity shuffle using topology
            if (config_.shuffle_affinity)
            {
                auto topo = read_topology();
                size_t idx = 0;
                registry().apply(pred, [&](RegisteredThreadInfo const& info) {
                    ThreadAffinity aff = affinity_for_node(
                        static_cast<int>(idx % (topo.numa_nodes > 0 ? topo.numa_nodes : 1)), static_cast<int>(idx));
                    (void)registry().set_affinity(info.tid, aff);
                    ++idx;
                });
            }
 
            // Priority jitter around the thread's actual priority
            if (config_.priority_jitter != 0)
            {
                std::uniform_int_distribution<int> dist(-config_.priority_jitter, config_.priority_jitter);
                registry().apply(pred, [&](RegisteredThreadInfo const& info) {
                    int delta = dist(rng);
                    int baseline = ThreadPriority::normal().value();
#ifndef _WIN32
                    sched_param sp{};
                    if (sched_getparam(info.tid, &sp) == 0)
                        baseline = sp.sched_priority;
#endif
                    (void)registry().set_priority(info.tid, ThreadPriority{baseline + delta});
                });
            }
 
            std::this_thread::sleep_for(config_.interval);
        }
    }
 
    ChaosConfig config_;
    std::atomic<bool> stop_;
    ThreadWrapper worker_;
    Tid worker_tid_{};
};
 
} // namespace threadschedule