scheduler_policy.hpp

Go to the documentation of this file.

#pragma once

 
#include "expected.hpp"
#include <algorithm>
#include <cstdint>
#include <fstream>
#include <optional>
#include <sstream>
#include <string>
#include <system_error>
#include <vector>
 
#ifdef _WIN32
#include <windows.h>
#else
#include <pthread.h>
#include <sched.h>
#include <sys/resource.h>
#include <sys/types.h>
#endif
 
namespace threadschedule
{
// expected/result are provided by expected.hpp
 
#ifdef _WIN32
using Tid = unsigned long; // DWORD thread id
#else
using Tid = pid_t; // Linux TID via gettid()
#endif

enum class SchedulingPolicy : std::uint_fast8_t
{
#ifdef _WIN32
    // Windows doesn't have the same scheduling policies as Linux
    // We'll use generic values
    OTHER = 0, 
    FIFO = 1,  
    RR = 2,    
    BATCH = 3, 
    IDLE = 4   
#else
    OTHER = SCHED_OTHER, 
    FIFO = SCHED_FIFO,   
    RR = SCHED_RR,       
    BATCH = SCHED_BATCH, 
    IDLE = SCHED_IDLE,   
#ifdef SCHED_DEADLINE
    DEADLINE = SCHED_DEADLINE 
#endif
#endif
};

class ThreadPriority
{
  public:
    constexpr explicit ThreadPriority(int priority = 0) : priority_(std::clamp(priority, min_priority, max_priority))
    {
    }
 
    [[nodiscard]] constexpr auto value() const noexcept -> int
    {
        return priority_;
    }
    [[nodiscard]] constexpr auto is_valid() const noexcept -> bool
    {
        return priority_ >= min_priority && priority_ <= max_priority;
    }
 
    [[nodiscard]] static constexpr auto lowest() noexcept -> ThreadPriority
    {
        return ThreadPriority(min_priority);
    }
    [[nodiscard]] static constexpr auto normal() noexcept -> ThreadPriority
    {
        return ThreadPriority(0);
    }
    [[nodiscard]] static constexpr auto highest() noexcept -> ThreadPriority
    {
        return ThreadPriority(max_priority);
    }
 
    [[nodiscard]] constexpr auto operator==(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ == other.priority_;
    }
    [[nodiscard]] constexpr auto operator!=(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ != other.priority_;
    }
    [[nodiscard]] constexpr auto operator<(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ < other.priority_;
    }
    [[nodiscard]] constexpr auto operator<=(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ <= other.priority_;
    }
    [[nodiscard]] constexpr auto operator>(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ > other.priority_;
    }
    [[nodiscard]] constexpr auto operator>=(ThreadPriority const& other) const noexcept -> bool
    {
        return priority_ >= other.priority_;
    }
 
    [[nodiscard]] auto to_string() const -> std::string
    {
        std::ostringstream oss;
        oss << "ThreadPriority(" << priority_ << ")";
        return oss.str();
    }
 
  private:
    static constexpr int min_priority = -20;
    static constexpr int max_priority = 19;
    int priority_;
};

class ThreadAffinity
{
  public:
    ThreadAffinity()
    {
#ifdef _WIN32
        group_ = 0;
        mask_ = 0;
#else
        CPU_ZERO(&cpuset_);
#endif
    }
 
    explicit ThreadAffinity(std::vector<int> const& cpus) : ThreadAffinity()
    {
        for (int cpu : cpus)
        {
            add_cpu(cpu);
        }
    }
 
    // Adds a CPU index. On Windows, indices >= 64 select group = cpu/64 automatically.
    void add_cpu(int cpu)
    {
#ifdef _WIN32
        if (cpu < 0)
            return;
        WORD g = static_cast<WORD>(cpu / 64);
        int bit = cpu % 64;
        if (!has_any())
        {
            group_ = g;
        }
        if (g != group_)
        {
            // Single-group affinity object: ignore CPUs from other groups
            return;
        }
        mask_ |= (static_cast<unsigned long long>(1) << bit);
#else
        if (cpu >= 0 && cpu < CPU_SETSIZE)
        {
            CPU_SET(cpu, &cpuset_);
        }
#endif
    }
 
    void remove_cpu(int cpu)
    {
#ifdef _WIN32
        if (cpu < 0)
            return;
        WORD g = static_cast<WORD>(cpu / 64);
        int bit = cpu % 64;
        if (g == group_)
        {
            mask_ &= ~(static_cast<unsigned long long>(1) << bit);
        }
#else
        if (cpu >= 0 && cpu < CPU_SETSIZE)
        {
            CPU_CLR(cpu, &cpuset_);
        }
#endif
    }
 
    [[nodiscard]] auto is_set(int cpu) const -> bool
    {
#ifdef _WIN32
        if (cpu < 0)
            return false;
        WORD g = static_cast<WORD>(cpu / 64);
        int bit = cpu % 64;
        return g == group_ && (mask_ & (static_cast<unsigned long long>(1) << bit)) != 0;
#else
        return cpu >= 0 && cpu < CPU_SETSIZE && CPU_ISSET(cpu, &cpuset_);
#endif
    }
 
    [[nodiscard]] auto has_cpu(int cpu) const -> bool
    {
        return is_set(cpu);
    }
 
    void clear()
    {
#ifdef _WIN32
        mask_ = 0;
#else
        CPU_ZERO(&cpuset_);
#endif
    }
 
    [[nodiscard]] auto get_cpus() const -> std::vector<int>
    {
        std::vector<int> cpus;
#ifdef _WIN32
        for (int i = 0; i < 64; ++i)
        {
            if (mask_ & (static_cast<unsigned long long>(1) << i))
            {
                cpus.push_back(static_cast<int>(group_) * 64 + i);
            }
        }
#else
        for (int i = 0; i < CPU_SETSIZE; ++i)
        {
            if (CPU_ISSET(i, &cpuset_))
            {
                cpus.push_back(i);
            }
        }
#endif
        return cpus;
    }
 
#ifdef _WIN32
    [[nodiscard]] unsigned long long get_mask() const
    {
        return mask_;
    }
    [[nodiscard]] WORD get_group() const
    {
        return group_;
    }
    [[nodiscard]] bool has_any() const
    {
        return mask_ != 0;
    }
#else
    [[nodiscard]] auto native_handle() const -> cpu_set_t const&
    {
        return cpuset_;
    }
#endif
 
    [[nodiscard]] auto to_string() const -> std::string
    {
        auto cpus = get_cpus();
        std::ostringstream oss;
        oss << "ThreadAffinity({";
        for (size_t i = 0; i < cpus.size(); ++i)
        {
            if (i > 0)
                oss << ", ";
            oss << cpus[i];
        }
        oss << "})";
        return oss.str();
    }
 
  private:
#ifdef _WIN32
    WORD group_;
    unsigned long long mask_;
#else
    cpu_set_t cpuset_;
#endif
};

class SchedulerParams
{
  public:
#ifdef _WIN32
    // Windows doesn't use sched_param, but we'll define a compatible type
    struct sched_param_win
    {
        int sched_priority;
    };
 
    static expected<sched_param_win, std::error_code> create_for_policy(SchedulingPolicy policy,
                                                                        ThreadPriority priority)
    {
        sched_param_win param{};
        // On Windows, priority is directly used
        param.sched_priority = priority.value();
        return param;
    }
 
    static expected<int, std::error_code> get_priority_range(SchedulingPolicy policy)
    {
        // Windows thread priorities range from -15 to +15
        return 30;
    }
#else
    static auto create_for_policy(SchedulingPolicy policy, ThreadPriority priority)
        -> expected<sched_param, std::error_code>
    {
        sched_param param{};
 
        int const policy_int = static_cast<int>(policy);
        int const min_prio = sched_get_priority_min(policy_int);
        int const max_prio = sched_get_priority_max(policy_int);
 
        if (min_prio == -1 || max_prio == -1)
        {
            return unexpected(std::make_error_code(std::errc::invalid_argument));
        }
 
        param.sched_priority = std::clamp(priority.value(), min_prio, max_prio);
        return param;
    }
 
    static auto get_priority_range(SchedulingPolicy policy) -> expected<int, std::error_code>
    {
        int const policy_int = static_cast<int>(policy);
        int const min_prio = sched_get_priority_min(policy_int);
        int const max_prio = sched_get_priority_max(policy_int);
 
        if (min_prio == -1 || max_prio == -1)
        {
            return unexpected(std::make_error_code(std::errc::invalid_argument));
        }
 
        return max_prio - min_prio;
    }
#endif
};

inline auto to_string(SchedulingPolicy policy) -> std::string
{
    switch (policy)
    {
    case SchedulingPolicy::OTHER:
        return "OTHER";
    case SchedulingPolicy::FIFO:
        return "FIFO";
    case SchedulingPolicy::RR:
        return "RR";
    case SchedulingPolicy::BATCH:
        return "BATCH";
    case SchedulingPolicy::IDLE:
        return "IDLE";
#if defined(SCHED_DEADLINE) && !defined(_WIN32)
    case SchedulingPolicy::DEADLINE:
        return "DEADLINE";
#endif
    default:
        return "UNKNOWN";
    }
}
 
// ---------------------------------------------------------------------------
// detail:: free functions for thread configuration (priority, policy, affinity)
//
// Overloaded by handle type so that every wrapper class can delegate with a
// single call: detail::apply_priority(handle, priority).
// ---------------------------------------------------------------------------
 
namespace detail
{
 
#ifdef _WIN32
 
inline auto map_priority_to_win32(int prio_val) -> int
{
    if (prio_val <= -10)
        return THREAD_PRIORITY_IDLE;
    if (prio_val <= -5)
        return THREAD_PRIORITY_LOWEST;
    if (prio_val < 0)
        return THREAD_PRIORITY_BELOW_NORMAL;
    if (prio_val == 0)
        return THREAD_PRIORITY_NORMAL;
    if (prio_val <= 5)
        return THREAD_PRIORITY_ABOVE_NORMAL;
    if (prio_val <= 10)
        return THREAD_PRIORITY_HIGHEST;
    return THREAD_PRIORITY_TIME_CRITICAL;
}
 
inline auto apply_priority(HANDLE handle, ThreadPriority priority) -> expected<void, std::error_code>
{
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
    if (SetThreadPriority(handle, map_priority_to_win32(priority.value())) != 0)
        return {};
    return unexpected(std::make_error_code(std::errc::operation_not_permitted));
}
 
inline auto apply_scheduling_policy(HANDLE handle, SchedulingPolicy /*policy*/, ThreadPriority priority)
    -> expected<void, std::error_code>
{
    return apply_priority(handle, priority);
}
 
inline auto apply_affinity(HANDLE handle, ThreadAffinity const& affinity) -> expected<void, std::error_code>
{
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
    using SetThreadGroupAffinityFn = BOOL(WINAPI*)(HANDLE, const GROUP_AFFINITY*, PGROUP_AFFINITY);
    HMODULE hMod = GetModuleHandleW(L"kernel32.dll");
    if (hMod)
    {
        auto set_group_affinity = reinterpret_cast<SetThreadGroupAffinityFn>(
            reinterpret_cast<void*>(GetProcAddress(hMod, "SetThreadGroupAffinity")));
        if (set_group_affinity && affinity.has_any())
        {
            GROUP_AFFINITY ga{};
            ga.Mask = static_cast<KAFFINITY>(affinity.get_mask());
            ga.Group = affinity.get_group();
            if (set_group_affinity(handle, &ga, nullptr) != 0)
                return {};
            return unexpected(std::make_error_code(std::errc::operation_not_permitted));
        }
    }
    DWORD_PTR mask = static_cast<DWORD_PTR>(affinity.get_mask());
    if (SetThreadAffinityMask(handle, mask) != 0)
        return {};
    return unexpected(std::make_error_code(std::errc::operation_not_permitted));
}
 
inline auto apply_name(HANDLE handle, std::string const& name) -> expected<void, std::error_code>
{
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
    using SetThreadDescriptionFn = HRESULT(WINAPI*)(HANDLE, PCWSTR);
    HMODULE hMod = GetModuleHandleW(L"kernel32.dll");
    if (!hMod)
        return unexpected(std::make_error_code(std::errc::function_not_supported));
    auto set_desc = reinterpret_cast<SetThreadDescriptionFn>(
        reinterpret_cast<void*>(GetProcAddress(hMod, "SetThreadDescription")));
    if (!set_desc)
        return unexpected(std::make_error_code(std::errc::function_not_supported));
    std::wstring wide(name.begin(), name.end());
    if (SUCCEEDED(set_desc(handle, wide.c_str())))
        return {};
    return unexpected(std::make_error_code(std::errc::operation_not_permitted));
}
 
inline auto read_name(HANDLE handle) -> std::optional<std::string>
{
    if (!handle)
        return std::nullopt;
    using GetThreadDescriptionFn = HRESULT(WINAPI*)(HANDLE, PWSTR*);
    HMODULE hMod = GetModuleHandleW(L"kernel32.dll");
    if (!hMod)
        return std::nullopt;
    auto get_desc = reinterpret_cast<GetThreadDescriptionFn>(
        reinterpret_cast<void*>(GetProcAddress(hMod, "GetThreadDescription")));
    if (!get_desc)
        return std::nullopt;
    PWSTR thread_name = nullptr;
    if (SUCCEEDED(get_desc(handle, &thread_name)) && thread_name)
    {
        int size = WideCharToMultiByte(CP_UTF8, 0, thread_name, -1, nullptr, 0, nullptr, nullptr);
        if (size > 0)
        {
            std::string result(size - 1, '\0');
            WideCharToMultiByte(CP_UTF8, 0, thread_name, -1, &result[0], size, nullptr, nullptr);
            LocalFree(thread_name);
            return result;
        }
        LocalFree(thread_name);
    }
    return std::nullopt;
}
 
inline auto read_affinity(HANDLE handle) -> std::optional<ThreadAffinity>
{
    if (!handle)
        return std::nullopt;
    using GetThreadGroupAffinityFn = BOOL(WINAPI*)(HANDLE, PGROUP_AFFINITY);
    HMODULE hMod = GetModuleHandleW(L"kernel32.dll");
    if (!hMod)
        return std::nullopt;
    auto get_group_affinity = reinterpret_cast<GetThreadGroupAffinityFn>(
        reinterpret_cast<void*>(GetProcAddress(hMod, "GetThreadGroupAffinity")));
    if (!get_group_affinity)
        return std::nullopt;
    GROUP_AFFINITY ga{};
    if (get_group_affinity(handle, &ga) != 0)
    {
        ThreadAffinity affinity;
        for (int i = 0; i < 64; ++i)
        {
            if ((ga.Mask & (static_cast<KAFFINITY>(1) << i)) != 0)
                affinity.add_cpu(static_cast<int>(ga.Group) * 64 + i);
        }
        if (affinity.has_any())
            return affinity;
    }
    return std::nullopt;
}
 
inline auto read_priority(HANDLE handle) -> std::optional<int>
{
    if (!handle)
        return std::nullopt;
    int const priority = GetThreadPriority(handle);
    if (priority == THREAD_PRIORITY_ERROR_RETURN)
        return std::nullopt;
    return priority;
}
 
inline auto read_scheduling_policy(HANDLE handle) -> std::optional<SchedulingPolicy>
{
    if (!handle)
        return std::nullopt;
    return SchedulingPolicy::OTHER;
}
 
inline auto apply_priority(Tid tid, ThreadPriority priority) -> expected<void, std::error_code>
{
    HANDLE handle = OpenThread(THREAD_SET_INFORMATION, FALSE, tid);
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
 
    auto result = apply_priority(handle, priority);
    CloseHandle(handle);
    return result;
}
 
inline auto apply_scheduling_policy(Tid tid, SchedulingPolicy policy, ThreadPriority priority)
    -> expected<void, std::error_code>
{
    HANDLE handle = OpenThread(THREAD_SET_INFORMATION, FALSE, tid);
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
 
    auto result = apply_scheduling_policy(handle, policy, priority);
    CloseHandle(handle);
    return result;
}
 
inline auto apply_affinity(Tid tid, ThreadAffinity const& affinity) -> expected<void, std::error_code>
{
    HANDLE handle = OpenThread(THREAD_SET_INFORMATION, FALSE, tid);
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
 
    auto result = apply_affinity(handle, affinity);
    CloseHandle(handle);
    return result;
}
 
inline auto apply_name(Tid tid, std::string const& name) -> expected<void, std::error_code>
{
    HANDLE handle = OpenThread(THREAD_SET_LIMITED_INFORMATION, FALSE, tid);
    if (!handle)
        return unexpected(std::make_error_code(std::errc::no_such_process));
 
    auto result = apply_name(handle, name);
    CloseHandle(handle);
    return result;
}
 
inline auto read_name(Tid tid) -> std::optional<std::string>
{
    HANDLE handle = OpenThread(THREAD_QUERY_LIMITED_INFORMATION, FALSE, tid);
    if (!handle)
        return std::nullopt;
 
    auto result = read_name(handle);
    CloseHandle(handle);
    return result;
}
 
inline auto read_affinity(Tid tid) -> std::optional<ThreadAffinity>
{
    HANDLE handle = OpenThread(THREAD_QUERY_INFORMATION, FALSE, tid);
    if (!handle)
        return std::nullopt;
 
    auto result = read_affinity(handle);
    CloseHandle(handle);
    return result;
}
 
inline auto read_priority(Tid tid) -> std::optional<int>
{
    HANDLE handle = OpenThread(THREAD_QUERY_INFORMATION, FALSE, tid);
    if (!handle)
        return std::nullopt;
 
    auto result = read_priority(handle);
    CloseHandle(handle);
    return result;
}
 
inline auto read_scheduling_policy(Tid tid) -> std::optional<SchedulingPolicy>
{
    HANDLE handle = OpenThread(THREAD_QUERY_INFORMATION, FALSE, tid);
    if (!handle)
        return std::nullopt;
 
    auto result = read_scheduling_policy(handle);
    CloseHandle(handle);
    return result;
}
 
#else // POSIX
 
// --- shared implementation for pthread_t and pid_t scheduling ---
 
template <typename SetSchedFn>
inline auto apply_sched_params(SchedulingPolicy policy, ThreadPriority priority, SetSchedFn&& set_sched)
    -> expected<void, std::error_code>
{
    int const policy_int = static_cast<int>(policy);
    auto params_result = SchedulerParams::create_for_policy(policy, priority);
    if (!params_result.has_value())
        return unexpected(params_result.error());
    if (set_sched(policy_int, &params_result.value()) == 0)
        return {};
    return unexpected(std::error_code(errno, std::generic_category()));
}
 
// --- pthread_t overloads (BaseThreadWrapper, ThreadControlBlock, PThreadWrapper) ---
 
inline auto apply_scheduling_policy(pthread_t handle, SchedulingPolicy policy, ThreadPriority priority)
    -> expected<void, std::error_code>
{
    return apply_sched_params(policy, priority,
                              [handle](int p, sched_param* sp) { return pthread_setschedparam(handle, p, sp); });
}
 
inline auto apply_priority(pthread_t handle, ThreadPriority priority) -> expected<void, std::error_code>
{
    return apply_scheduling_policy(handle, SchedulingPolicy::OTHER, priority);
}
 
inline auto apply_affinity(pthread_t handle, ThreadAffinity const& affinity) -> expected<void, std::error_code>
{
    if (pthread_setaffinity_np(handle, sizeof(cpu_set_t), &affinity.native_handle()) == 0)
        return {};
    return unexpected(std::error_code(errno, std::generic_category()));
}
 
inline auto apply_name(pthread_t handle, std::string const& name) -> expected<void, std::error_code>
{
    if (name.length() > 15)
        return unexpected(std::make_error_code(std::errc::invalid_argument));
    if (pthread_setname_np(handle, name.c_str()) == 0)
        return {};
    return unexpected(std::error_code(errno, std::generic_category()));
}
 
inline auto read_name(pthread_t handle) -> std::optional<std::string>
{
    char name[16];
    if (pthread_getname_np(handle, name, sizeof(name)) == 0)
        return std::string(name);
    return std::nullopt;
}
 
inline auto read_affinity(pthread_t handle) -> std::optional<ThreadAffinity>
{
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    if (pthread_getaffinity_np(handle, sizeof(cpu_set_t), &cpuset) == 0)
    {
        std::vector<int> cpus;
        for (int i = 0; i < CPU_SETSIZE; ++i)
        {
            if (CPU_ISSET(i, &cpuset))
                cpus.push_back(i);
        }
        return ThreadAffinity(cpus);
    }
    return std::nullopt;
}
 
inline auto read_priority(pthread_t handle) -> std::optional<int>
{
    int policy = 0;
    sched_param param{};
    if (pthread_getschedparam(handle, &policy, &param) == 0)
        return param.sched_priority;
    return std::nullopt;
}
 
inline auto read_scheduling_policy(pthread_t handle) -> std::optional<SchedulingPolicy>
{
    int policy = 0;
    sched_param param{};
    if (pthread_getschedparam(handle, &policy, &param) == 0)
        return static_cast<SchedulingPolicy>(policy);
    return std::nullopt;
}
 
// --- pid_t / TID overloads (ThreadByNameView) ---
 
inline auto apply_scheduling_policy(pid_t tid, SchedulingPolicy policy, ThreadPriority priority)
    -> expected<void, std::error_code>
{
    return apply_sched_params(policy, priority,
                              [tid](int p, sched_param* sp) { return sched_setscheduler(tid, p, sp); });
}
 
inline auto apply_priority(pid_t tid, ThreadPriority priority) -> expected<void, std::error_code>
{
    return apply_scheduling_policy(tid, SchedulingPolicy::OTHER, priority);
}
 
inline auto apply_affinity(pid_t tid, ThreadAffinity const& affinity) -> expected<void, std::error_code>
{
    if (sched_setaffinity(tid, sizeof(cpu_set_t), &affinity.native_handle()) == 0)
        return {};
    return unexpected(std::error_code(errno, std::generic_category()));
}
 
inline auto apply_name(pid_t tid, std::string const& name) -> expected<void, std::error_code>
{
    if (name.length() > 15)
        return unexpected(std::make_error_code(std::errc::invalid_argument));
 
    std::string const path = std::string("/proc/self/task/") + std::to_string(tid) + "/comm";
    std::ofstream out(path);
    if (!out)
        return unexpected(std::error_code(errno, std::generic_category()));
 
    out << name;
    out.flush();
    if (!out)
        return unexpected(std::error_code(errno, std::generic_category()));
    return {};
}
 
inline auto read_name(pid_t tid) -> std::optional<std::string>
{
    std::string const path = std::string("/proc/self/task/") + std::to_string(tid) + "/comm";
    std::ifstream in(path);
    if (!in)
        return std::nullopt;
 
    std::string current;
    std::getline(in, current);
    if (!current.empty() && current.back() == '\n')
        current.pop_back();
    return current;
}
 
inline auto read_affinity(pid_t tid) -> std::optional<ThreadAffinity>
{
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    if (sched_getaffinity(tid, sizeof(cpu_set_t), &cpuset) != 0)
        return std::nullopt;
 
    std::vector<int> cpus;
    for (int i = 0; i < CPU_SETSIZE; ++i)
    {
        if (CPU_ISSET(i, &cpuset))
            cpus.push_back(i);
    }
    return ThreadAffinity(cpus);
}
 
inline auto read_priority(pid_t tid) -> std::optional<int>
{
    sched_param param{};
    if (sched_getparam(tid, &param) == 0)
        return param.sched_priority;
    return std::nullopt;
}
 
inline auto read_scheduling_policy(pid_t tid) -> std::optional<SchedulingPolicy>
{
    int const policy = sched_getscheduler(tid);
    if (policy == -1)
        return std::nullopt;
    return static_cast<SchedulingPolicy>(policy);
}
 
#endif
 
} // namespace detail
 
} // namespace threadschedule