linux.h

Go to the documentation of this file.

// Copyright (c) 2026 Tinverse LLC. All rights reserved.
// SPDX-License-Identifier: LicenseRef-Tinverse-Commercial
 
 
#pragma once
 
#include <atomic>
#include <cerrno>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <pthread.h>
#include <sched.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/mman.h>
#include <sys/utsname.h>
#include <thread>
#include <time.h>
#include <type_traits>
#include <unistd.h>
 
#include "tsm/chrono_ticks.h"
#include "tsm/runtime/executor.h"
 
namespace tsm::detail {
 
struct linux_syscalls
{
    static int clock_gettime(clockid_t clock_id, struct timespec* ts) noexcept
    {
        return ::clock_gettime(clock_id, ts);
    }
 
    static int nanosleep(struct timespec const* request,
                         struct timespec* remaining) noexcept
    {
        return ::nanosleep(request, remaining);
    }
 
    static FILE* fopen(char const* path, char const* mode) noexcept
    {
        return std::fopen(path, mode);
    }
 
    static int fgetc(FILE* file) noexcept
    {
        return std::fgetc(file);
    }
 
    static int fclose(FILE* file) noexcept
    {
        return std::fclose(file);
    }
 
    static int uname(struct utsname* kernel) noexcept
    {
        return ::uname(kernel);
    }
 
    static pthread_t pthread_self() noexcept
    {
        return ::pthread_self();
    }
 
    static int pthread_setschedparam(pthread_t thread,
                                     int policy,
                                     struct sched_param const* param) noexcept
    {
        return ::pthread_setschedparam(thread, policy, param);
    }
 
    static int pthread_setaffinity_np(pthread_t thread,
                                      std::size_t cpusetsize,
                                      cpu_set_t const* cpuset) noexcept
    {
        return ::pthread_setaffinity_np(thread, cpusetsize, cpuset);
    }
 
    static int mlockall(int flags) noexcept
    {
        return ::mlockall(flags);
    }
 
    static void perror(char const* message) noexcept
    {
        std::perror(message);
    }
 
    static int eventfd(unsigned int initval, int flags) noexcept
    {
        return ::eventfd(initval, flags);
    }
 
    static int epoll_create1(int flags) noexcept
    {
        return ::epoll_create1(flags);
    }
 
    static int epoll_ctl(int epfd,
                         int op,
                         int fd,
                         struct epoll_event* event) noexcept
    {
        return ::epoll_ctl(epfd, op, fd, event);
    }
 
    static int epoll_wait(int epfd,
                          struct epoll_event* events,
                          int maxevents,
                          int timeout) noexcept
    {
        return ::epoll_wait(epfd, events, maxevents, timeout);
    }
 
    static ssize_t read(int fd, void* buffer, std::size_t count) noexcept
    {
        return ::read(fd, buffer, count);
    }
 
    static ssize_t write(int fd, void const* buffer, std::size_t count) noexcept
    {
        return ::write(fd, buffer, count);
    }
 
    static int close(int fd) noexcept
    {
        return ::close(fd);
    }
};
 
struct AccurateClock
{
    using duration = std::chrono::nanoseconds;
    using period = duration::period;
    using rep = duration::rep;
    using time_point = std::chrono::time_point<AccurateClock>;
 
    static constexpr bool is_steady = true;
 
    static time_point now() noexcept
    {
        struct timespec ts;
        linux_syscalls::clock_gettime(CLOCK_MONOTONIC, &ts);
        auto duration_since_epoch = std::chrono::seconds(ts.tv_sec) +
                                    std::chrono::nanoseconds(ts.tv_nsec);
        return time_point(duration(duration_since_epoch));
    }
};
 
template<typename Clock = std::chrono::steady_clock,
         typename Duration = typename Clock::duration>
struct Timer
{
    using ClockType = Clock;
    using DurationType = Duration;
    void start()
    {
        start_time_ = Clock::now();
        started_ = true;
    }
 
    Duration elapsed() const
    {
        if (!started_) {
            return Duration(0);
        }
        auto now = Clock::now();
        auto interval = now - start_time_;
        return std::chrono::duration_cast<Duration>(interval);
    }
 
    template<typename ToDuration = Duration>
    Duration elapsed(ToDuration since) const
    {
        auto now = Clock::now();
        if constexpr (std::is_same<ToDuration, Duration>::value) {
            return std::chrono::duration_cast<Duration>(now - since);
        } else {
            return std::chrono::duration_cast<ToDuration>(now - since);
        }
    }
 
    template<typename ToDuration>
    ToDuration elapsed() const
    {
        return std::chrono::duration_cast<ToDuration>(elapsed());
    }
 
    bool started() const
    {
        return started_;
    }
    void reset()
    {
        start_time_ = Clock::now();
    }
    void stop()
    {
        started_ = false;
    }
 
  protected:
    typename Clock::time_point start_time_;
    bool started_{ false };
};
 
template<typename Clock = std::chrono::steady_clock,
         typename Duration = typename Clock::duration>
struct IntervalTimer : public Timer<Clock, Duration>
{
    Duration interval()
    {
        if (!this->started()) {
            this->start();
            return Duration(0);
        }
        auto now = Clock::now();
        auto interval = now - this->start_time_;
        this->start_time_ = now;
        return std::chrono::duration_cast<Duration>(interval);
    }
};
 
template<typename Clock = AccurateClock,
         typename Duration = typename Clock::duration,
         typename Sys = linux_syscalls>
struct PeriodicSleepTimer : public Timer<Clock, Duration>
{
    PeriodicSleepTimer(Duration period = Duration(1))
      : period_(period)
    {
    }
    void start()
    {
        Timer<Clock, Duration>::start();
    }
 
    void wait()
    {
        auto remaining = period_ - Timer<Clock, Duration>::elapsed();
        struct timespec ts;
        ts.tv_sec =
          std::chrono::duration_cast<std::chrono::seconds>(remaining).count();
        ts.tv_nsec =
          std::chrono::duration_cast<std::chrono::nanoseconds>(remaining)
            .count();
 
        struct timespec remaining_ts;
        while (Sys::nanosleep(&ts, &remaining_ts) != 0 && errno == EINTR) {
            ts = remaining_ts;
        }
        Timer<Clock, Duration>::reset();
    }
 
    Duration get_period() const
    {
        return period_;
    }
 
  protected:
    Duration period_;
};
 
template<typename Sys>
struct basic_realtime_configurator
{
    basic_realtime_configurator() = default;
    basic_realtime_configurator(int priority, std::array<int, 4> affinity)
      : PROCESS_PRIORITY(priority)
      , CPU_AFFINITY(affinity)
    {
    }
 
    static bool running_realtime_kernel()
    {
        if (auto* realtime = Sys::fopen("/sys/kernel/realtime", "r")) {
            int value = Sys::fgetc(realtime);
            Sys::fclose(realtime);
            return value == '1';
        }
 
        struct utsname kernel
        {};
        if (Sys::uname(&kernel) != 0) {
            return false;
        }
        return std::strstr(kernel.release, "-rt") != nullptr ||
               std::strstr(kernel.version, "PREEMPT_RT") != nullptr;
    }
 
    void config_realtime_thread()
    {
        if (!running_realtime_kernel()) {
            return;
        }
 
        struct sched_param param;
        param.sched_priority = PROCESS_PRIORITY - 3;
        const auto sched_error =
          Sys::pthread_setschedparam(Sys::pthread_self(), SCHED_RR, &param);
        if (sched_error != 0 && sched_error != EPERM) {
            Sys::perror("pthread_setschedparam");
        }
 
        cpu_set_t cpuset;
        CPU_ZERO(&cpuset);
        for (auto cpu : CPU_AFFINITY) {
            CPU_SET(cpu, &cpuset);
        }
 
        if (Sys::pthread_setaffinity_np(
              Sys::pthread_self(), sizeof(cpu_set_t), &cpuset) != 0) {
            Sys::perror("sched_setaffinity");
        }
 
        if (Sys::mlockall(MCL_CURRENT | MCL_FUTURE) == -1) {
            Sys::perror("mlockall failed");
        }
    }
 
    template<typename Fn>
    std::thread real_time_thread(Fn fn)
    {
        return std::thread([this, fn] {
            this->config_realtime_thread();
            fn();
        });
    }
 
    std::thread make_real_time(std::thread&& t)
    {
        config_realtime_thread();
        return std::move(t);
    }
 
  protected:
    int PROCESS_PRIORITY{ 98 };
    std::array<int, 4> CPU_AFFINITY{ 0, 1, 2, 3 };
};
 
using RealtimeConfigurator = basic_realtime_configurator<linux_syscalls>;
 
struct host_thread_policy
{
    void configure_current_thread() {}
};
 
template<typename TickPeriod = std::chrono::milliseconds>
using host_tick_source =
  tsm::chrono_ticks::elapsed_tick_source<AccurateClock, TickPeriod>;
 
template<typename Sys = linux_syscalls>
struct basic_host_realtime_policy : basic_realtime_configurator<Sys>
{
    using basic_realtime_configurator<Sys>::basic_realtime_configurator;
 
    void configure_current_thread()
    {
        this->config_realtime_thread();
    }
};
 
using host_realtime_policy = basic_host_realtime_policy<linux_syscalls>;
 
template<typename ThreadPolicy, typename Sys, typename... Tasks>
class basic_thread_executor
{
  public:
    explicit basic_thread_executor(Tasks&... tasks)
      : ready_(tasks...)
    {
        event_fd_ = Sys::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        epoll_fd_ = Sys::epoll_create1(EPOLL_CLOEXEC);
        if (event_fd_ >= 0 && epoll_fd_ >= 0) {
            epoll_event event{};
            event.events = EPOLLIN;
            event.data.fd = event_fd_;
            static_cast<void>(
              Sys::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &event));
        }
    }
 
    basic_thread_executor(basic_thread_executor const&) = delete;
    basic_thread_executor& operator=(basic_thread_executor const&) = delete;
 
    ~basic_thread_executor()
    {
        stop();
        if (event_fd_ >= 0) {
            Sys::close(event_fd_);
        }
        if (epoll_fd_ >= 0) {
            Sys::close(epoll_fd_);
        }
    }
 
    void start()
    {
        if (worker_.joinable()) {
            return;
        }
        stop_requested_.store(false);
        worker_ = std::thread([this] {
            policy_.configure_current_thread();
            while (!stop_requested_.load()) {
                // The worker drains cooperative work to quiescence, then waits
                // for an OS wake event. HSM dispatch remains runtime-owned.
                if (ready_.run_ready() == 0U) {
                    wait_for_work();
                }
            }
        });
    }
 
    void stop()
    {
        stop_requested_.store(true);
        wake();
        if (worker_.joinable()) {
            worker_.join();
        }
    }
 
    void wake()
    {
        if (event_fd_ < 0) {
            return;
        }
        // eventfd collapses multiple wakes into a counter. That is sufficient
        // because ready work is drained until no runtime reports progress.
        std::uint64_t value = 1U;
        const auto written = Sys::write(event_fd_, &value, sizeof(value));
        static_cast<void>(written);
    }
 
    void wake_from_isr()
    {
        wake();
    }
 
    void wait_for_work()
    {
        if (epoll_fd_ < 0 || event_fd_ < 0) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return;
        }
        // The timeout keeps stop() bounded even if the wake file descriptor was
        // not created, while the normal path wakes immediately through eventfd.
        epoll_event event{};
        const int ready = Sys::epoll_wait(epoll_fd_, &event, 1, 1);
        if (ready > 0 && event.data.fd == event_fd_) {
            std::uint64_t value{};
            while (Sys::read(event_fd_, &value, sizeof(value)) > 0) {}
        }
    }
 
    [[nodiscard]] bool step()
    {
        return ready_.step();
    }
 
    std::size_t run_ready()
    {
        return ready_.run_ready();
    }
 
    std::size_t tick(tsm::tick_rep elapsed_ticks = 1U)
    {
        auto resumed = ready_.tick(elapsed_ticks);
        if (resumed != 0U) {
            wake();
        }
        return resumed;
    }
 
    std::size_t tick(tsm::tick_count elapsed_ticks)
    {
        return tick(elapsed_ticks.count());
    }
 
    void start_all()
    {
        ready_.start_all();
    }
 
    template<auto Entry, std::size_t Instance = 0U>
    [[nodiscard]] spawn_result start()
    {
        auto result = ready_.template start<Entry, Instance>();
        if (result == spawn_result::started) {
            wake();
        }
        return result;
    }
 
    template<auto Entry, typename... Args>
    [[nodiscard]] spawn_result spawn(Args&&... args)
    {
        auto result = ready_.template spawn<Entry>(std::forward<Args>(args)...);
        if (result == spawn_result::started) {
            wake();
        }
        return result;
    }
 
    [[nodiscard]] runtime::task_spawner<basic_thread_executor> spawner()
    {
        return runtime::task_spawner<basic_thread_executor>{ *this };
    }
 
    template<auto Entry, std::size_t Instance = 0U>
    [[nodiscard]] task_status task_status() const
    {
        return ready_.template task_status<Entry, Instance>();
    }
 
    template<auto Entry, std::size_t Instance = 0U>
    [[nodiscard]] task_failure_reason task_failure_reason() const
    {
        return ready_.template task_failure_reason<Entry, Instance>();
    }
 
    template<auto Entry, std::size_t Instance = 0U>
    [[nodiscard]] bool cancel() noexcept
    {
        const bool cancelled = ready_.template cancel<Entry, Instance>();
        if (cancelled) {
            wake();
        }
        return cancelled;
    }
 
    void cancel_all() noexcept
    {
        ready_.cancel_all();
        wake();
    }
 
  private:
    ThreadPolicy policy_{};
    tsm::runtime::cooperative_executor<Tasks...> ready_;
    std::thread worker_{};
    std::atomic_bool stop_requested_{ true };
    int event_fd_{ -1 };
    int epoll_fd_{ -1 };
};
 
template<typename... Tasks>
class thread_executor
  : public basic_thread_executor<host_thread_policy, linux_syscalls, Tasks...>
{
    using base =
      basic_thread_executor<host_thread_policy, linux_syscalls, Tasks...>;
 
  public:
    using base::base;
};
 
template<typename... Tasks>
thread_executor(Tasks&...) -> thread_executor<Tasks...>;
 
template<typename... Tasks>
class realtime_thread_executor
  : public basic_thread_executor<host_realtime_policy, linux_syscalls, Tasks...>
{
    using base =
      basic_thread_executor<host_realtime_policy, linux_syscalls, Tasks...>;
 
  public:
    using base::base;
};
 
template<typename... Tasks>
realtime_thread_executor(Tasks&...) -> realtime_thread_executor<Tasks...>;
 
template<typename... Tasks>
using task_executor = thread_executor<Tasks...>;
 
} // namespace tsm::detail