AsyncEvent.hpp

Go to the documentation of this file.

//
//  CeresEngine - A game development framework
//
//  Created by Rogiel Sulzbach.
//  Copyright (c) 2018-2022 Rogiel Sulzbach. All rights reserved.
//
 
#pragma once
 
#include "CeresEngine/DataTypes.hpp"
#include "CeresEngine/Macros.hpp"
 
#include "CeresEngine/Foundation/Coroutine.hpp"
#include "CeresEngine/Foundation/Container/Atomic.hpp"
 
namespace CeresEngine {
 
    class AsyncManualResetEvent {
    public:
        class Operation;
 
    public:
        AsyncManualResetEvent(bool initiallySet = false) noexcept;
        ~AsyncManualResetEvent();
 
        Operation operator co_await() const noexcept;
 
        bool isSet() const noexcept;
 
        void set() noexcept;
 
        void reset() noexcept;
 
    public:
        class Operation {
        public:
            explicit Operation(const AsyncManualResetEvent& event) noexcept;
 
            bool await_ready() const noexcept;
            bool await_suspend(CoroutineHandle<> awaiter) noexcept;
            void await_resume() const noexcept {}
 
        private:
            friend class AsyncManualResetEvent;
 
            const AsyncManualResetEvent& mEvent;
            Operation* mNext;
            CoroutineHandle<> mAwaiter;
        };
 
    private:
        friend class Operation;
 
        // This variable has 3 states:
        // - this    - The state is 'set'.
        // - nullptr - The state is 'not set' with no waiters.
        // - other   - The state is 'not set'.
        //             Points to an 'Operation' that is
        //             the head of a linked-list of waiters.
        mutable Atomic<void*> mState;
    };
 
    class AsyncAutoResetEvent {
    public:
        class Operation;
 
    public:
        AsyncAutoResetEvent(bool initiallySet = false) noexcept;
        ~AsyncAutoResetEvent();
 
        [[nodiscard]] Operation operator co_await() const noexcept;
 
        void set() noexcept;
 
        void reset() noexcept;
 
    public:
        class Operation {
        public:
            Operation() noexcept;
 
            explicit Operation(const AsyncAutoResetEvent& event) noexcept;
            Operation(const Operation& other) noexcept;
 
            [[nodiscard]] bool await_ready() const noexcept { return mEvent == nullptr; }
            [[nodiscard]] bool await_suspend(CoroutineHandle<> awaiter) noexcept;
            void await_resume() const noexcept {}
 
        private:
            friend class AsyncAutoResetEvent;
 
            const AsyncAutoResetEvent* mEvent;
            Operation* mNext;
            CoroutineHandle<> mAwaiter;
            Atomic<UInt32> mRefCount;
        };
 
    private:
        friend class Operation;
 
        void resume_waiters(UInt64 initialState) const noexcept;
 
        // Bits 0-31  - Set count
        // Bits 32-63 - Waiter count
        mutable Atomic<UInt64> mState;
        mutable Atomic<Operation*> mNewWaiters;
        mutable Operation* mWaiters;
    };
 
    class SingleConsumerAsyncAutoResetEvent {
    public:
        SingleConsumerAsyncAutoResetEvent(const bool initiallySet = false) noexcept : mState(initiallySet ? this : nullptr) {}
 
        void set() noexcept {
            void* oldValue = mState.exchange(this, std::memory_order_release);
            if(oldValue != nullptr && oldValue != this) {
                // There was a waiting coroutine that we now need to resume.
                const auto handle = *static_cast<CoroutineHandle<>*>(oldValue);
 
                // We also need to transition the state back to 'not set' before
                // resuming the coroutine. This operation needs to be 'acquire'
                // so that it synchronises with other calls to .set() that execute
                // concurrently with this call and execute the above mState.exchange(this)
                // operation with 'release' semantics.
                // This needs to be an exchange() instead of a store() so that it can have
                // 'acquire' semantics.
                (void)mState.exchange(nullptr, std::memory_order_acquire);
 
                // Finally, resume the waiting coroutine.
                handle.resume();
            }
        }
 
        [[nodiscard]] auto operator co_await() const noexcept {
            class Awaiter {
            public:
                explicit Awaiter(const SingleConsumerAsyncAutoResetEvent& event) noexcept : mEvent(event) {}
 
                [[nodiscard]] bool await_ready() const noexcept { return false; }
                [[nodiscard]] bool await_suspend(const CoroutineHandle<> awaitingCoroutine) noexcept {
                    mAwaitingCoroutine = awaitingCoroutine;
 
                    void* oldValue = nullptr;
                    if(!mEvent.mState.compare_exchange_strong(oldValue, &mAwaitingCoroutine, std::memory_order_release, std::memory_order_relaxed)) {
                        // This will only fail if the event was already 'set'
                        // In which case we can just reset back to 'not set'
                        // Need to use exchange() rather than store() here so we can make this
                        // operation an 'acquire' operation so that we get visibility of all
                        // writes prior to all preceding calls to .set().
                        CE_ASSERT(oldValue == &mEvent);
                        (void)mEvent.mState.exchange(nullptr, std::memory_order_acquire);
                        return false;
                    }
 
                    return true;
                }
 
                void await_resume() noexcept {}
 
            private:
                const SingleConsumerAsyncAutoResetEvent& mEvent;
                CoroutineHandle<> mAwaitingCoroutine;
            };
 
            return Awaiter{*this};
        }
 
    private:
        // nullptr - not set, no waiter
        // this    - set
        // other   - not set, pointer is address of a coroutine_handle<> to resume.
        mutable Atomic<void*> mState;
    };
 
    class SingleConsumerEvent {
    public:
        SingleConsumerEvent(const bool initiallySet = false) noexcept : mState(initiallySet ? state::set : state::not_set) {}
 
        [[nodiscard]] bool is_set() const noexcept { return mState.load(std::memory_order_acquire) == state::set; }
 
        void set() {
            const state oldState = mState.exchange(state::set, std::memory_order_acq_rel);
            if(oldState == state::not_set_consumer_waiting) {
                mAwaiter.resume();
            }
        }
 
        void reset() noexcept {
            state oldState = state::set;
            mState.compare_exchange_strong(oldState, state::not_set, std::memory_order_relaxed);
        }
 
        [[nodiscard]] auto operator co_await() noexcept {
            class Awaiter {
            public:
                explicit Awaiter(SingleConsumerEvent& event) : mEvent(event) {}
 
                [[nodiscard]] bool await_ready() const noexcept { return mEvent.is_set(); }
                [[nodiscard]] bool await_suspend(const CoroutineHandle<> awaiter) {
                    mEvent.mAwaiter = awaiter;
 
                    state oldState = state::not_set;
                    return mEvent.mState.compare_exchange_strong(oldState, state::not_set_consumer_waiting, std::memory_order_release, std::memory_order_acquire);
                }
 
                void await_resume() noexcept {}
 
            private:
                SingleConsumerEvent& mEvent;
            };
 
            return Awaiter{*this};
        }
 
    private:
        enum class state {
            not_set,
            not_set_consumer_waiting,
            set
        };
 
        // TODO: Merge these two fields into a single Atomic<std::uintptr_t>
        // by encoding 'not_set' as 0 (nullptr), 'set' as 1 and
        // 'not_set_consumer_waiting' as a coroutine handle pointer.
        Atomic<state> mState;
        CoroutineHandle<> mAwaiter;
    };
 
} // namespace CeresEngine