futex.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "condy/detail/execution_backend.hpp"
#include "condy/detail/intrusive.hpp"
#include "condy/detail/invoker.hpp"
#include "condy/detail/type_traits.hpp"
#include "condy/detail/utils.hpp"
#include "condy/runtime.hpp"
#include <atomic>
#include <cerrno>
#include <optional>
#ifdef CONDY_DETAIL_HAS_EXECUTION
#include "condy/detail/execution.hpp"
#endif
 
namespace condy {

template <typename T> class Futex {
public:
    Futex(std::atomic<T> &futex) : futex_(futex) {}
 
    ~Futex() { notify_all_(-EIDRM); }
 
    CONDY_DELETE_COPY_MOVE(Futex);
 
private:
    struct [[nodiscard]] WaitSenderImpl;
 
public:
#ifdef CONDY_DETAIL_HAS_EXECUTION
    using WaitSender = detail::StandardSender<WaitSenderImpl>;
#else
    using WaitSender = WaitSenderImpl;
#endif
    WaitSender wait(T old) noexcept { return {*this, old}; }

    void notify_one() noexcept {
        WaitFinishHandleBase *handle = nullptr;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            handle = wait_awaiters_.pop_front();
        }
        if (handle) {
            handle->set_result(0);
            handle->schedule();
        }
    }

    void notify_all() noexcept { notify_all_(0); }
 
private:
    class WaitFinishHandleBase;
    template <typename Receiver> class WaitFinishHandle;
 
    bool cancel_wait_(WaitFinishHandleBase *handle) noexcept {
        std::lock_guard<std::mutex> lock(mutex_);
        return wait_awaiters_.remove(handle);
    }
 
    int32_t request_wait_(WaitFinishHandleBase *handle, T old) noexcept {
        std::lock_guard<std::mutex> lock(mutex_);
        auto val = futex_.load(std::memory_order_relaxed);
        if (val != old) {
            return 0; // No need to wait
        }
        wait_awaiters_.push_back(handle);
        return -EAGAIN; // Need to wait
    }
 
private:
    void notify_all_(int32_t result) noexcept {
        HandleList handles;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            handles = std::move(wait_awaiters_);
        }
        while (auto *handle = handles.pop_front()) {
            handle->set_result(result);
            handle->schedule();
        }
    }
 
private:
    using HandleList =
        detail::IntrusiveDoubleList<WaitFinishHandleBase,
                                    &WaitFinishHandleBase::link_entry_>;
 
    mutable std::mutex mutex_;
    HandleList wait_awaiters_;
    std::atomic<T> &futex_;
};
 
template <typename T>
class Futex<T>::WaitFinishHandleBase : public detail::WorkInvoker {
public:
    void schedule() noexcept {
        assert(runtime_ != nullptr);
        runtime_->schedule_internal(this);
    }
 
    void set_result(int32_t result) noexcept { result_ = result; }
 
public:
    detail::DoubleLinkEntry link_entry_;
 
protected:
    Runtime *runtime_ = nullptr;
    int32_t result_ = -ENOTRECOVERABLE; // Internal error if not set
};
 
template <typename T>
template <typename Receiver>
class Futex<T>::WaitFinishHandle
    : public detail::InvokerAdapter<WaitFinishHandle<Receiver>,
                                    WaitFinishHandleBase> {
public:
    using Base = detail::InvokerAdapter<WaitFinishHandle, WaitFinishHandleBase>;
 
    WaitFinishHandle(Futex &futex, Receiver receiver)
        : futex_(futex), receiver_(std::move(receiver)) {}
 
    void start(Runtime *runtime, T old) noexcept {
        this->runtime_ = runtime;
        int32_t r = futex_.request_wait_(this, old);
        if (r != -EAGAIN) {
            std::move(receiver_)(r);
            return;
        }
        runtime->pend_work_internal();
 
        auto stop_token = receiver_.get_stop_token();
        if (stop_token.stop_possible()) {
            stop_callback_.emplace(std::move(stop_token), Cancellation{this});
        }
    }
 
    void invoke() noexcept {
        stop_callback_.reset();
        assert(this->runtime_ != nullptr);
        this->runtime_->resume_work_internal();
        std::move(receiver_)(this->result_);
    }
 
private:
    void cancel_() noexcept {
        if (futex_.cancel_wait_(this)) {
            // Successfully canceled
            this->result_ = -ECANCELED;
            assert(this->runtime_ != nullptr);
            this->runtime_->schedule_internal(this);
        }
    }
 
    struct Cancellation {
        WaitFinishHandle *self;
        void operator()() noexcept { self->cancel_(); }
    };
 
    using StopCallbackType =
        detail::stop_callback_t<detail::stop_token_t<Receiver>, Cancellation>;
 
private:
    Futex &futex_;
    Receiver receiver_;
    std::optional<StopCallbackType> stop_callback_;
};
 
template <typename T> struct Futex<T>::WaitSenderImpl {
public:
    using CondySender = void;
    using ReturnType = int32_t;
 
    WaitSenderImpl(Futex &futex, T old) : futex_(futex), old_(old) {}
 
    template <typename Receiver> auto connect_impl(Receiver receiver) noexcept {
        return OperationState<Receiver>(futex_, old_, std::move(receiver));
    }
 
private:
    template <typename Receiver>
    class OperationState : public WaitFinishHandle<Receiver> {
    public:
        using Base = WaitFinishHandle<Receiver>;
        OperationState(Futex &futex, T old, Receiver receiver)
            : Base(futex, std::move(receiver)), old_(old) {}
 
        void start(unsigned int /*flags*/) noexcept {
            auto *runtime = detail::Context::current().runtime();
            Base::start(runtime, old_);
        }
 
    private:
        T old_;
    };
 
private:
    Futex &futex_;
    T old_;
};
 
} // namespace condy