sync.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 <algorithm>
#include <array>
#include <concepts>
#include <cstddef>
#include <cstdint>
#include <type_traits>
#include <utility>
 
#include "tsm/runtime/coroutine.h"
#include "tsm/runtime/policy.h"
 
namespace tsm {
 
namespace runtime::detail {
 
template<std::size_t MaxWaiters>
class waiter_list
{
  public:
    [[nodiscard]] bool add(task_context& context) noexcept
    {
        std::replace_if(
          waiters_.begin(),
          waiters_.end(),
          [](auto const& waiter) {
              return waiter.context != nullptr &&
                     waiter.context->generation() != waiter.generation;
          },
          waiter_record{});
        // Re-registering the same waiter is harmless. This matters when a task
        // observes a primitive more than once before the scheduler resumes it.
        if (std::any_of(
              waiters_.begin(), waiters_.end(), [&context](auto const& waiter) {
                  return waiter.context == &context &&
                         waiter.generation == context.generation();
              })) {
            return true;
        }
        auto slot = std::find_if(
          waiters_.begin(), waiters_.end(), [](auto const& waiter) {
              return waiter.context == nullptr;
          });
        if (slot != waiters_.end()) {
            *slot = waiter_record{ &context, context.generation() };
            return true;
        }
        // Waiter overflow is a configuration error for static systems. Marking
        // the task failed makes the failure visible without throwing.
        context.mark_failed(task_failure_reason::wait_queue_full);
        return false;
    }
 
    void remove(task_context& context) noexcept
    {
        std::replace_if(
          waiters_.begin(),
          waiters_.end(),
          [&context](auto const& waiter) { return waiter.context == &context; },
          waiter_record{});
    }
 
    void wake_one() noexcept
    {
        // Wake order is stable by waiter-slot order. It is not a priority
        // mechanism; scheduling fairness is handled by the task ready queue.
        for (auto& waiter : waiters_) {
            if (waiter.context == nullptr) {
                continue;
            }
            if (waiter.context->generation() == waiter.generation) {
                waiter.context->mark_ready();
            }
            waiter = {};
            return;
        }
    }
 
    void wake_all() noexcept
    {
        for (auto& waiter : waiters_) {
            if (waiter.context != nullptr) {
                if (waiter.context->generation() == waiter.generation) {
                    waiter.context->mark_ready();
                }
                waiter = {};
            }
        }
    }
 
  private:
    struct waiter_record
    {
        task_context* context{};
        std::uint32_t generation{};
    };
 
    std::array<waiter_record, MaxWaiters> waiters_{};
};
 
} // namespace runtime::detail
 
template<std::size_t MaxWaiters = 4U>
class event_flag
{
  public:
    void set() noexcept
    {
        set_ = true;
        waiters_.wake_all();
    }
 
    void set_from_isr() noexcept
    {
        set();
    }
 
    void reset() noexcept
    {
        set_ = false;
    }
    [[nodiscard]] bool is_set() const noexcept
    {
        return set_;
    }
 
    class wait_awaitable
    {
      public:
        explicit wait_awaitable(event_flag& flag) noexcept
          : flag_(&flag)
        {
        }
 
        [[nodiscard]] bool await_ready() const noexcept
        {
            return flag_->is_set();
        }
 
        void await_suspend(task::handle_type handle) noexcept
        {
            (void)flag_->waiters_.add(handle.promise().context());
        }
 
        void await_cancel(task_context& context) noexcept
        {
            flag_->waiters_.remove(context);
        }
 
        void await_resume() const noexcept {}
 
      private:
        event_flag* flag_{};
    };
 
    [[nodiscard]] wait_awaitable wait() noexcept
    {
        return wait_awaitable{ *this };
    }
 
  private:
    bool set_{};
    runtime::detail::waiter_list<MaxWaiters> waiters_{};
};
 
template<std::size_t MaxWaiters = 4U>
class irq_event
{
  public:
    void notify() noexcept
    {
        flag_.set();
    }
    void notify_from_isr() noexcept
    {
        flag_.set();
    }
    void reset() noexcept
    {
        flag_.reset();
    }
    [[nodiscard]] bool is_set() const noexcept
    {
        return flag_.is_set();
    }
 
    [[nodiscard]] auto wait() noexcept
    {
        return flag_.wait();
    }
 
  private:
    event_flag<MaxWaiters> flag_{};
};
 
template<std::size_t MaxWaiters = 4U>
using os_event = irq_event<MaxWaiters>;
 
template<std::size_t MaxWaiters>
class cancellation_source;
 
template<std::size_t MaxWaiters = 4U>
class cancellation_token
{
  public:
    cancellation_token() = default;
 
    [[nodiscard]] bool stop_requested() const noexcept
    {
        return source_ != nullptr && source_->stop_requested();
    }
 
    class wait_awaitable
    {
      public:
        explicit wait_awaitable(
          cancellation_source<MaxWaiters>& source) noexcept
          : source_(&source)
        {
        }
 
        [[nodiscard]] bool await_ready() const noexcept
        {
            return source_->stop_requested();
        }
 
        void await_suspend(task::handle_type handle) noexcept
        {
            (void)source_->waiters_.add(handle.promise().context());
        }
 
        void await_cancel(task_context& context) noexcept
        {
            source_->waiters_.remove(context);
        }
 
        void await_resume() const noexcept {}
 
      private:
        cancellation_source<MaxWaiters>* source_{};
    };
 
    [[nodiscard]] wait_awaitable wait() const noexcept
    {
        return wait_awaitable{ *source_ };
    }
 
  private:
    friend class cancellation_source<MaxWaiters>;
 
    explicit cancellation_token(
      cancellation_source<MaxWaiters>& source) noexcept
      : source_(&source)
    {
    }
 
    cancellation_source<MaxWaiters>* source_{};
};
 
template<std::size_t MaxWaiters = 4U>
class cancellation_source
{
  public:
    [[nodiscard]] cancellation_token<MaxWaiters> token() noexcept
    {
        return cancellation_token<MaxWaiters>{ *this };
    }
 
    void request_cancel() noexcept
    {
        requested_ = true;
        waiters_.wake_all();
    }
 
    void reset() noexcept
    {
        requested_ = false;
    }
 
    [[nodiscard]] bool stop_requested() const noexcept
    {
        return requested_;
    }
 
  private:
    friend class cancellation_token<MaxWaiters>;
 
    bool requested_{};
    runtime::detail::waiter_list<MaxWaiters> waiters_{};
};
 
template<typename T, std::size_t MaxWaiters = 4U>
class signal
{
  public:
    void publish(T const& value)
    {
        value_ = value;
        ready_ = true;
        waiters_.wake_all();
    }
 
    void publish(T&& value)
    {
        value_ = std::move(value);
        ready_ = true;
        waiters_.wake_all();
    }
 
    void publish_from_isr(T const& value)
    {
        publish(value);
    }
 
    void publish_from_isr(T&& value)
    {
        publish(std::move(value));
    }
 
    [[nodiscard]] bool ready() const noexcept
    {
        return ready_;
    }
    [[nodiscard]] T const& value() const noexcept
    {
        return value_;
    }
 
    class wait_awaitable
    {
      public:
        explicit wait_awaitable(signal& source) noexcept
          : source_(&source)
        {
        }
 
        [[nodiscard]] bool await_ready() const noexcept
        {
            return source_->ready();
        }
 
        void await_suspend(task::handle_type handle) noexcept
        {
            (void)source_->waiters_.add(handle.promise().context());
        }
 
        void await_cancel(task_context& context) noexcept
        {
            source_->waiters_.remove(context);
        }
 
        [[nodiscard]] T await_resume() const
        {
            return source_->value_;
        }
 
      private:
        signal* source_{};
    };
 
    [[nodiscard]] wait_awaitable wait() noexcept
    {
        return wait_awaitable{ *this };
    }
 
  private:
    T value_{};
    bool ready_{};
    runtime::detail::waiter_list<MaxWaiters> waiters_{};
};
 
namespace runtime {
 
template<typename T>
class sender
{
  public:
    sender() = default;
 
    template<typename Channel>
    explicit sender(Channel& channel) noexcept
      : object_(&channel)
      , send_const_([](void* object, T const& value) {
          return static_cast<Channel*>(object)->try_send(value);
      })
      , send_move_([](void* object, T&& value) {
          return static_cast<Channel*>(object)->try_send(std::move(value));
      })
    {
    }
 
    [[nodiscard]] bool try_send(T const& value) const
    {
        return send_const_ != nullptr && send_const_(object_, value);
    }
 
    [[nodiscard]] bool try_send(T&& value) const
    {
        return send_move_ != nullptr && send_move_(object_, std::move(value));
    }
 
  private:
    using send_const_fn = bool (*)(void*, T const&);
    using send_move_fn = bool (*)(void*, T&&);
 
    void* object_{};
    send_const_fn send_const_{};
    send_move_fn send_move_{};
};
 
template<typename T>
class receiver
{
  public:
    receiver() = default;
 
    template<typename Channel>
    explicit receiver(Channel& channel) noexcept
      : object_(&channel)
      , receive_([](void* object, T& value) {
          return static_cast<Channel*>(object)->try_receive(value);
      })
    {
    }
 
    [[nodiscard]] bool try_receive(T& value) const
    {
        return receive_ != nullptr && receive_(object_, value);
    }
 
  private:
    using receive_fn = bool (*)(void*, T&);
 
    void* object_{};
    receive_fn receive_{};
};
 
template<typename T>
class latest_reader
{
  public:
    latest_reader() = default;
 
    template<typename Channel>
    explicit latest_reader(Channel const& channel) noexcept
      : object_(&channel)
      , latest_([](void const* object, T& value) {
          return static_cast<Channel const*>(object)->latest(value);
      })
    {
    }
 
    [[nodiscard]] bool latest(T& value) const
    {
        return latest_ != nullptr && latest_(object_, value);
    }
 
  private:
    using latest_fn = bool (*)(void const*, T&);
 
    void const* object_{};
    latest_fn latest_{};
};
 
template<typename T,
         std::size_t Capacity,
         typename OverflowPolicy = overflow::reject_newest,
         std::size_t MaxWaiters = 4U>
class channel
{
    static_assert(Capacity > 0U, "tsm: channel requires non-zero capacity");
    static_assert(overflow_policy<OverflowPolicy>,
                  "tsm: channel requires a known overflow policy");
 
  public:
    using value_type = T;
    using overflow_policy = OverflowPolicy;
    static constexpr std::size_t static_capacity = Capacity;
 
    [[nodiscard]] bool try_send(T const& value)
    {
        const bool accepted = push(value);
        if (accepted) {
            latest_ = data_[latest_index()];
            has_latest_ = true;
            receivers_.wake_one();
        }
        return accepted;
    }
 
    [[nodiscard]] bool try_send(T&& value)
    {
        const bool accepted = push(std::move(value));
        if (accepted) {
            latest_ = data_[latest_index()];
            has_latest_ = true;
            receivers_.wake_one();
        }
        return accepted;
    }
 
    [[nodiscard]] bool try_send_from_isr(T const& value)
    {
        return try_send(value);
    }
 
    [[nodiscard]] bool try_send_from_isr(T&& value)
    {
        return try_send(std::move(value));
    }
 
    [[nodiscard]] bool try_receive(T& value)
    {
        if (empty()) {
            return false;
        }
        // Receive transfers from the fixed ring slot into caller-owned storage.
        // The slot is reused by later sends.
        value = std::move(data_[pop_index_]);
        pop_index_ = next(pop_index_);
        --size_;
        return true;
    }
 
    [[nodiscard]] bool empty() const noexcept
    {
        return size_ == 0U;
    }
    [[nodiscard]] bool full() const noexcept
    {
        return size_ == Capacity;
    }
    [[nodiscard]] std::size_t size() const noexcept
    {
        return size_;
    }
    [[nodiscard]] static constexpr std::size_t capacity() noexcept
    {
        return Capacity;
    }
 
    [[nodiscard]] bool latest(T& value) const
    {
        if (!has_latest_) {
            return false;
        }
        value = latest_;
        return true;
    }
 
    [[nodiscard]] runtime::sender<T> sender() noexcept
    {
        return runtime::sender<T>{ *this };
    }
 
    [[nodiscard]] runtime::receiver<T> receiver() noexcept
    {
        return runtime::receiver<T>{ *this };
    }
 
    [[nodiscard]] runtime::latest_reader<T> latest_reader() const noexcept
    {
        return runtime::latest_reader<T>{ *this };
    }
 
    template<typename Value>
    class send_awaitable
    {
      public:
        send_awaitable(channel& target, Value&& value)
          : target_(&target)
          , value_(std::forward<Value>(value))
        {
        }
 
        [[nodiscard]] bool await_ready()
        {
            // Send is a non-blocking awaitable: it completes immediately with
            // an accepted/rejected result so backpressure is explicit.
            accepted_ = target_->try_send(std::move(value_));
            return true;
        }
 
        void await_suspend(task::handle_type) noexcept {}
        [[nodiscard]] bool await_resume() const noexcept
        {
            return accepted_;
        }
 
      private:
        channel* target_{};
        T value_;
        bool accepted_{};
    };
 
    template<typename Value>
    [[nodiscard]] auto send(Value&& value)
    {
        return send_awaitable<Value>{ *this, std::forward<Value>(value) };
    }
 
    class receive_awaitable
    {
      public:
        explicit receive_awaitable(channel& source) noexcept
          : source_(&source)
        {
        }
 
        [[nodiscard]] bool await_ready()
        {
            has_value_ = source_->try_receive(value_);
            return has_value_;
        }
 
        void await_suspend(task::handle_type handle) noexcept
        {
            // Empty receives suspend the task. A later successful send wakes
            // one receiver through the scheduler ready queue.
            (void)source_->receivers_.add(handle.promise().context());
        }
 
        void await_cancel(task_context& context) noexcept
        {
            source_->receivers_.remove(context);
        }
 
        [[nodiscard]] T await_resume()
        {
            if (!has_value_) {
                // The value may have been provided by the send that woke this
                // task; take it now that the coroutine is running again.
                has_value_ = source_->try_receive(value_);
            }
            return std::move(value_);
        }
 
      private:
        channel* source_{};
        T value_{};
        bool has_value_{};
    };
 
    [[nodiscard]] receive_awaitable receive() noexcept
    {
        return receive_awaitable{ *this };
    }
 
  private:
    static constexpr std::size_t next(std::size_t index) noexcept
    {
        return (index + 1U) % Capacity;
    }
 
    static constexpr std::size_t previous(std::size_t index) noexcept
    {
        return index == 0U ? Capacity - 1U : index - 1U;
    }
 
    [[nodiscard]] std::size_t latest_index() const noexcept
    {
        return previous(push_index_);
    }
 
    template<typename Value>
    [[nodiscard]] bool push(Value&& value)
    {
        if (full()) {
            if constexpr (std::same_as<OverflowPolicy,
                                       overflow::reject_newest>) {
                return false;
            } else if constexpr (std::same_as<OverflowPolicy,
                                              overflow::drop_oldest>) {
                pop_index_ = next(pop_index_);
                --size_;
            } else {
                data_[previous(push_index_)] = std::forward<Value>(value);
                return true;
            }
        }
        // The channel owns a copy of each payload. Transport-specific zero-copy
        // storage belongs in runtime adapters, not in this cooperative
        // primitive.
        data_[push_index_] = std::forward<Value>(value);
        push_index_ = next(push_index_);
        ++size_;
        return true;
    }
 
    std::array<T, Capacity> data_{};
    T latest_{};
    std::size_t push_index_{};
    std::size_t pop_index_{};
    std::size_t size_{};
    bool has_latest_{};
    runtime::detail::waiter_list<MaxWaiters> receivers_{};
};
 
} // namespace runtime
 
template<typename T,
         std::size_t Capacity,
         typename Overflow = overflow::reject_newest,
         std::size_t MaxWaiters = 4U>
using channel = runtime::channel<T, Capacity, Overflow, MaxWaiters>;
 
template<typename T, std::size_t MaxWaiters = 4U>
class mutex
{
  public:
    template<typename... Args>
    explicit mutex(Args&&... args)
      : value_(std::forward<Args>(args)...)
    {
    }
 
    mutex() = default;
 
    class lock_awaitable
    {
      public:
        explicit lock_awaitable(mutex& target) noexcept
          : target_(&target)
        {
        }
 
        [[nodiscard]] bool await_ready() noexcept
        {
            if (!target_->locked_) {
                target_->locked_ = true;
                owns_ = true;
                return true;
            }
            return false;
        }
 
        void await_suspend(task::handle_type handle) noexcept
        {
            (void)target_->waiters_.add(handle.promise().context());
        }
 
        void await_cancel(task_context& context) noexcept
        {
            target_->waiters_.remove(context);
        }
 
        [[nodiscard]] T& await_resume() noexcept
        {
            if (!owns_) {
                // A woken waiter becomes owner when it resumes. The primitive
                // remains cooperative; no OS mutex or preemptive wait is used.
                target_->locked_ = true;
            }
            return target_->value_;
        }
 
      private:
        mutex* target_{};
        bool owns_{};
    };
 
    [[nodiscard]] lock_awaitable lock() noexcept
    {
        return lock_awaitable{ *this };
    }
 
    void unlock() noexcept
    {
        if (!locked_) {
            return;
        }
        locked_ = false;
        waiters_.wake_one();
    }
 
    template<typename Fn>
    decltype(auto) with_lock(Fn&& fn)
    {
        return std::forward<Fn>(fn)(value_);
    }
 
    [[nodiscard]] bool locked() const noexcept
    {
        return locked_;
    }
    [[nodiscard]] T& unsafe_value() noexcept
    {
        return value_;
    }
    [[nodiscard]] T const& unsafe_value() const noexcept
    {
        return value_;
    }
 
  private:
    T value_{};
    bool locked_{};
    runtime::detail::waiter_list<MaxWaiters> waiters_{};
};
 
} // namespace tsm