kevent.h

Go to the documentation of this file.

// Copyright (c) 2026 Tinverse LLC. All rights reserved.
// SPDX-License-Identifier: LicenseRef-Tinverse-Commercial
 
 
#pragma once
 
#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) ||      \
  defined(__NetBSD__)
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <sys/event.h>
#include <sys/time.h>
#include <thread>
#include <unistd.h>
#include <utility>
 
#include "tsm/runtime/executor.h"
 
namespace tsm {
 
template<typename... Tasks>
class kevent_task_executor
{
  public:
    explicit kevent_task_executor(Tasks&... tasks)
      : ready_(tasks...)
    {
        queue_fd_ = kqueue();
        if (queue_fd_ >= 0) {
            struct kevent event
            {};
            EV_SET(&event, 1, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, nullptr);
            static_cast<void>(
              kevent(queue_fd_, &event, 1, nullptr, 0, nullptr));
        }
    }
 
    kevent_task_executor(kevent_task_executor const&) = delete;
    kevent_task_executor& operator=(kevent_task_executor const&) = delete;
 
    ~kevent_task_executor()
    {
        stop();
        if (queue_fd_ >= 0) {
            close(queue_fd_);
        }
    }
 
    void start()
    {
        if (worker_.joinable()) {
            return;
        }
        stop_requested_.store(false);
        worker_ = std::thread([this] {
            while (!stop_requested_.load()) {
                if (ready_.run_ready() == 0U) {
                    wait_for_work();
                }
            }
        });
    }
 
    void stop()
    {
        stop_requested_.store(true);
        wake();
        if (worker_.joinable()) {
            worker_.join();
        }
    }
 
    void wake()
    {
        if (queue_fd_ < 0) {
            return;
        }
        // EVFILT_USER gives the executor a portable BSD/macOS wake signal
        // without coupling runtime queues to kqueue.
        struct kevent event
        {};
        EV_SET(&event, 1, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);
        static_cast<void>(kevent(queue_fd_, &event, 1, nullptr, 0, nullptr));
    }
 
    void wake_from_isr()
    {
        wake();
    }
 
    void wait_for_work()
    {
        if (queue_fd_ < 0) {
            return;
        }
        // The timeout keeps shutdown responsive; normal work arrives through
        // wake(), which triggers the user event immediately.
        struct kevent event
        {};
        struct timespec timeout
        {
            0, 1000000
        };
        static_cast<void>(kevent(queue_fd_, nullptr, 0, &event, 1, &timeout));
    }
 
    [[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)
    {
        const 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;
    }
 
    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:
    runtime::cooperative_executor<Tasks...> ready_;
    std::thread worker_{};
    std::atomic_bool stop_requested_{ true };
    int queue_fd_{ -1 };
};
 
template<typename... Tasks>
kevent_task_executor(Tasks&...) -> kevent_task_executor<Tasks...>;
 
template<typename... Tasks>
using task_executor = kevent_task_executor<Tasks...>;
 
} // namespace tsm
 
#endif