SignalPolicies.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <atomic>
#include <chrono>
#include <cstdint>
 
#if defined(_WIN32) || defined(_WIN64) || defined(__unix__) || defined(__APPLE__) || (defined(_GLIBCXX_HAS_GTHREADS) && _GLIBCXX_HAS_GTHREADS) || defined(_LIBCPP_HAS_THREAD_API_PTHREAD)
#include <condition_variable>
#include <mutex>
#include <thread>
#ifndef CORIUM_HAS_STD_MUTEX
#define CORIUM_HAS_STD_MUTEX 1
#endif
#else
#ifndef CORIUM_HAS_STD_MUTEX
#define CORIUM_HAS_STD_MUTEX 0
#endif
#endif
 
#ifdef __linux__
#include <poll.h>
#include <sys/eventfd.h>
#include <unistd.h>
#endif
 
namespace corium {
 
struct StaticCallback {
    using SimpleFn = void (*)();
    using ContextFn = void (*)(void* arg);
 
    ContextFn fn = nullptr;
    void* arg = nullptr;
 
    StaticCallback() = default;
 
    /* implicit */ StaticCallback(SimpleFn simpleFn)
        : fn(reinterpret_cast<ContextFn>(simpleFn))
    {
        if (simpleFn) {
            // Helper trampoline for parameterless function pointers
            fn = [](void* context) {
                reinterpret_cast<SimpleFn>(context)();
            };
            arg = reinterpret_cast<void*>(simpleFn);
        }
    }
 
    StaticCallback(ContextFn contextFn, void* contextArg)
        : fn(contextFn), arg(contextArg)
    {}
 
    void operator()() const {
        if (fn) {
            fn(arg);
        }
    }
 
    explicit operator bool() const noexcept {
        return fn != nullptr;
    }
};
 
class NoSignalPolicy {
public:
    void setOnQueueNonEmpty(StaticCallback cb) noexcept { (void)cb; }
    void signal() noexcept {}
 
    template <typename Rep, typename Period>
    void wait_for(const std::chrono::duration<Rep, Period>&) noexcept
    {
#if defined(__arm__) || defined(__aarch64__)
        asm volatile("yield");
#elif defined(__x86_64__)
        __builtin_ia32_pause();
#else
        // zero-op spin yield
#endif
    }
};
 
class CallbackSignalPolicy {
public:
    void setOnQueueNonEmpty(StaticCallback callback)
    {
#if CORIUM_HAS_STD_MUTEX
        std::lock_guard<std::mutex> lock(_mutex);
#endif
        _callback = callback;
    }
 
    void signal()
    {
#if CORIUM_HAS_STD_MUTEX
        std::lock_guard<std::mutex> lock(_mutex);
        _hasEvents = true;
        _cv.notify_one();
#else
        _hasEvents = true;
#endif
        if (_callback) {
            _callback();
        }
    }
 
    template <typename Rep, typename Period>
    void wait_for(const std::chrono::duration<Rep, Period>& timeout)
    {
#if CORIUM_HAS_STD_MUTEX
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait_for(lock, timeout, [this]() { return _hasEvents; });
        _hasEvents = false;
#else
        (void)timeout;
#if defined(__arm__) || defined(__aarch64__)
        asm volatile("yield");
#endif
        _hasEvents = false;
#endif
    }
 
private:
    StaticCallback _callback;
#if CORIUM_HAS_STD_MUTEX
    std::mutex _mutex;
    std::condition_variable _cv;
#endif
    bool _hasEvents = false;
};
 
class AtomicWaitSignalPolicy {
public:
    void setOnQueueNonEmpty(StaticCallback callback)
    {
        _userCallback = callback;
    }
 
    void signal()
    {
        _flag.store(true, std::memory_order_release);
#if defined(__cpp_lib_atomic_wait) && __cpp_lib_atomic_wait >= 201907L
        _flag.notify_one();
#endif
        if (_userCallback) {
            _userCallback();
        }
    }
 
    template <typename Rep, typename Period>
    void wait_for(const std::chrono::duration<Rep, Period>& timeout)
    {
        if (_flag.load(std::memory_order_acquire)) {
            _flag.store(false, std::memory_order_relaxed);
            return;
        }
        auto deadline = std::chrono::steady_clock::now() + timeout;
        while (!_flag.load(std::memory_order_acquire)) {
            if (std::chrono::steady_clock::now() >= deadline) {
                return;
            }
#if CORIUM_HAS_STD_MUTEX
            std::this_thread::yield();
#elif defined(__arm__) || defined(__aarch64__)
            asm volatile("yield");
#endif
        }
        _flag.store(false, std::memory_order_relaxed);
    }
 
private:
    std::atomic<bool> _flag{false};
    StaticCallback _userCallback;
};
 
#ifdef __linux__
class EventFdSignalPolicy {
public:
    EventFdSignalPolicy()
    {
        _fd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    }
 
    ~EventFdSignalPolicy()
    {
        if (_fd >= 0) {
            ::close(_fd);
        }
    }
 
    EventFdSignalPolicy(const EventFdSignalPolicy&) = delete;
    EventFdSignalPolicy& operator=(const EventFdSignalPolicy&) = delete;
 
    EventFdSignalPolicy(EventFdSignalPolicy&& rhs) noexcept : _fd(rhs._fd)
    {
        rhs._fd = -1;
    }
 
    EventFdSignalPolicy& operator=(EventFdSignalPolicy&& rhs) noexcept
    {
        if (this != &rhs) {
            if (_fd >= 0) ::close(_fd);
            _fd = rhs._fd;
            rhs._fd = -1;
        }
        return *this;
    }
 
    void setOnQueueNonEmpty(StaticCallback callback)
    {
        _userCallback = callback;
    }
 
    void signal()
    {
        uint64_t val = 1;
        [[maybe_unused]] auto res = ::write(_fd, &val, sizeof(val));
        if (_userCallback) {
            _userCallback();
        }
    }
 
    template <typename Rep, typename Period>
    void wait_for(const std::chrono::duration<Rep, Period>& timeout)
    {
        if (_fd < 0) return;
        struct pollfd pfd;
        pfd.fd = _fd;
        pfd.events = POLLIN;
        int timeoutMs = static_cast<int>(std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count());
        int res = ::poll(&pfd, 1, timeoutMs);
        if (res > 0 && (pfd.revents & POLLIN)) {
            uint64_t val = 0;
            [[maybe_unused]] auto bytes = ::read(_fd, &val, sizeof(val));
        }
    }
 
    [[nodiscard]] int nativeHandle() const noexcept
    {
        return _fd;
    }
 
private:
    int _fd = -1;
    StaticCallback _userCallback;
};
#endif // __linux__
 
} // namespace corium