wake_signal.hpp

Go to the documentation of this file.

#pragma once

 
#include <atomic>
#include <chrono>
#include <cstdint>
 
#if defined(__APPLE__)
#include <dispatch/dispatch.h>
#else
#include <semaphore>
#endif
 
namespace pulp::runtime {

class WakeSignal {
  public:
    using Clock = std::chrono::steady_clock;
 
    WakeSignal() noexcept
#if defined(__APPLE__)
        : semaphore_(dispatch_semaphore_create(0))
#endif
    {
    }
 
    ~WakeSignal() {
#if defined(__APPLE__)
        // libdispatch traps when a semaphore is released below its initial
        // value, i.e. with a thread still waiting. Surplus posts leave the
        // value above zero, which is allowed.
        if (semaphore_ != nullptr)
            dispatch_release(semaphore_);
#endif
    }
 
    WakeSignal(const WakeSignal&) = delete;
    WakeSignal& operator=(const WakeSignal&) = delete;

    void notify() noexcept {
        std::atomic_thread_fence(std::memory_order_seq_cst);
        if (parked_.load(std::memory_order_relaxed)) {
            posts_.fetch_add(1, std::memory_order_relaxed);
            post();
        }
    }

    template <typename Ready> bool wait_until(Ready&& ready, Clock::time_point deadline) {
        for (;;) {
            if (ready())
                return true;
            parked_.store(true, std::memory_order_relaxed);
            std::atomic_thread_fence(std::memory_order_seq_cst);
            if (ready()) {
                parked_.store(false, std::memory_order_relaxed);
                return true;
            }
            const auto now = Clock::now();
            if (now >= deadline) {
                parked_.store(false, std::memory_order_relaxed);
                return false;
            }
            park(deadline - now);
            parked_.store(false, std::memory_order_relaxed);
            wakes_.fetch_add(1, std::memory_order_relaxed);
        }
    }

    template <typename Ready> bool wait_for(Ready&& ready, Clock::duration timeout) {
        return wait_until(static_cast<Ready&&>(ready), Clock::now() + timeout);
    }

    std::uint64_t posts() const noexcept {
        return posts_.load(std::memory_order_relaxed);
    }
    std::uint64_t wakes() const noexcept {
        return wakes_.load(std::memory_order_relaxed);
    }

    bool consumer_parked() const noexcept {
        return parked_.load(std::memory_order_relaxed);
    }
 
  private:
    void post() noexcept {
#if defined(__APPLE__)
        (void)dispatch_semaphore_signal(semaphore_);
#else
        semaphore_.release();
#endif
    }
 
    void park(Clock::duration timeout) {
#if defined(__APPLE__)
        const auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(timeout).count();
        (void)dispatch_semaphore_wait(semaphore_, dispatch_time(DISPATCH_TIME_NOW, ns));
#else
        (void)semaphore_.try_acquire_for(timeout);
#endif
    }
 
    std::atomic<bool> parked_{false};
    std::atomic<std::uint64_t> posts_{0};
    std::atomic<std::uint64_t> wakes_{0};
#if defined(__APPLE__)
    dispatch_semaphore_t semaphore_ = nullptr;
#else
    std::counting_semaphore<> semaphore_{0};
#endif
};
 
} // namespace pulp::runtime