Channel.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/Threading/Threading.hpp"
 
#include <continuable/continuable.hpp>
 
#include <cassert>
#include <mutex>
#include <optional>
 
namespace CeresEngine {
 
    template<typename T, typename Lockable = NullLockable> class Channel final {
        // Channels can only work with value types.
        static_assert(std::is_reference_v<T> == false, "Using reference for channel is forbidden.");
 
        using CoroutineHandle = typename std::coroutine_handle<void>;
 
    public:
        using Pointer = T*;
 
        using Reference = T&;
 
        using Mutex = Lockable;
 
        class Reader;
 
        class Writer;
 
    private:
        Mutex mutex;
 
        template<typename NodeType> class LinkedList;
 
        LinkedList<Reader> readers;
 
        LinkedList<Writer> writers;
 
    public:
        Channel() = default;
 
        Channel(const Channel&) = delete;
 
        Channel(Channel&&) = delete;
 
        Channel& operator=(const Channel&) = delete;
 
        Channel& operator=(Channel&&) = delete;
 
        ~Channel() noexcept;
 
    public:
        [[nodiscard]] Writer write(T value) noexcept { return Writer{*this, std::forward<T>(value)}; }
 
        [[nodiscard]] Reader read() noexcept { return Reader{*this}; }
 
    private:
        [[nodiscard]] static constexpr void* poison() noexcept { return nullptr; }
 
        template<typename NodeType> class LinkedList final {
            NodeType* head = nullptr;
 
            NodeType* tail = nullptr;
 
        public:
            LinkedList() noexcept = default;
 
        public:
            [[nodiscard]] bool empty() const noexcept { return head == nullptr; }
 
            void push(NodeType* node) noexcept {
                if(tail) {
                    tail->next = node;
                    tail = node;
                } else {
                    head = tail = node;
                }
            }
 
            [[nodiscard]] NodeType* pop() noexcept {
                NodeType* node = head;
                if(head == tail) { // empty or 1
                    head = tail = nullptr;
                } else { // 2 or more
                    head = head->next;
                }
 
                return node; // this can be nullptr
            }
        };
 
    public:
        class Reader final {
            friend LinkedList<Reader>;
            friend Channel;
 
        private:
            mutable Optional<T> value = std::nullopt;
 
            mutable Optional<CoroutineHandle> handle = nullptr;
 
            union {
                Reader* next = nullptr;
 
                Channel* channel;
            };
 
        private:
            explicit Reader(Channel& ch) noexcept : channel(std::addressof(ch)) {}
 
        public:
            Reader(const Reader&) = delete;
 
            Reader& operator=(const Reader&) = delete;
 
            Reader(Reader&& rhs) noexcept {
                std::swap(value, rhs.value);
                std::swap(handle, rhs.handle);
                std::swap(channel, rhs.channel);
            }
 
            Reader& operator=(Reader&& rhs) noexcept {
                std::swap(value, rhs.value);
                std::swap(handle, rhs.handle);
                std::swap(channel, rhs.channel);
                return *this;
            }
 
            ~Reader() noexcept = default;
 
        public:
            bool await_ready() const&;
            void await_suspend(CoroutineHandle coro) &;
            Optional<T> await_resume() &;
        };
 
        class Writer final {
            friend LinkedList<Writer>;
            friend Channel;
 
        private:
            mutable Optional<T> value = std::nullopt;
 
            mutable Optional<CoroutineHandle> handle = nullptr;
 
            union {
                Writer* next = nullptr;
 
                Channel* channel;
            };
 
        private:
            explicit Writer(Channel& channel, T&& value) noexcept : value(std::forward<T>(value)), channel(std::addressof(channel)) {}
 
        public:
            Writer(const Writer&) = delete;
 
            Writer& operator=(const Writer&) = delete;
 
            Writer(Writer&& rhs) noexcept {
                std::swap(value, rhs.value);
                std::swap(handle, rhs.handle);
                std::swap(channel, rhs.channel);
            }
 
            Writer& operator=(Writer&& rhs) noexcept {
                std::swap(value, rhs.value);
                std::swap(handle, rhs.handle);
                std::swap(channel, rhs.channel);
                return *this;
            }
 
            ~Writer() noexcept = default;
 
        public:
            bool await_ready() const&;
            void await_suspend(CoroutineHandle coro) &;
            bool await_resume() &;
        };
    };
 
    template<typename T, typename Lockable> Channel<T, Lockable>::~Channel() noexcept {
        // Because of thread scheduling,
        // Some coroutines can be enqueued into list just after
        // this destructor unlocks.
        //
        // But this can't be detected at once since
        // we have 2 list in the channel...
        //
        // Current implementation triggers scheduling repeatedly
        // to reduce the possibility. As repeat count becomes greater,
        // the possibility drops to zero. But notice that it is NOT zero.
        //
        size_t repeat = 1; // recommend 5'000+ repeat for hazard usage
        do {
            Lock<Mutex> lock(mutex);
 
            while(!writers.empty()) {
                Writer* w = writers.pop();
                auto rh = w->handle;
                w->handle = std::nullopt;
 
                rh->resume();
            }
            while(!readers.empty()) {
                Reader* r = readers.pop();
                auto rh = r->handle;
                r->handle = std::nullopt;
 
                rh->resume();
            }
        } while(repeat--);
    }
 
    template<typename T, typename Lockable> bool Channel<T, Lockable>::Reader::await_ready() const& {
        Lock<Mutex> lock(channel->mutex);
        if(channel->writers.empty())
            CE_UNLIKELY { return false; }
 
        Writer* w = channel->writers.pop();
        //        assert(w != nullptr);
        //        assert(w->ptr != nullptr);
        //        assert(w->handle != nullptr);
 
        // exchange address & resumeable_handle
        std::swap(this->value, w->value);
        std::swap(this->handle, w->handle);
 
        return true;
    }
 
    template<typename T, typename Lockable> void Channel<T, Lockable>::Reader::await_suspend(CoroutineHandle coro) & {
        // notice that next & chan are sharing memory
        Channel& ch = *(this->channel);
 
        this->handle = coro; // remember handle before push/unlock
        this->next = nullptr; // clear to prevent confusing
 
        ch.readers.push(this); // push to channel
        // ch.mutex.unlock();
    }
 
    template<typename T, typename Lockable> Optional<T> Channel<T, Lockable>::Reader::await_resume() & {
        // frame holds poision if the channel is going to destroy
        if(!handle)
            CE_UNLIKELY { return std::nullopt; }
 
        // Store first. we have to do this
        // because the resume operation can destroy the writer coroutine
        T value = std::move(this->value.get());
        if(const auto coroutine = *handle) {
            //            assert(this->handle != nullptr);
            //            assert(*reinterpret_cast<UInt64*>(handle) != 0);
            coroutine.resume();
        }
 
        return std::move(value);
    }
 
    template<typename T, typename Lockable> bool Channel<T, Lockable>::Writer::await_ready() const& {
        Lock<Mutex> lock(channel->mutex);
        if(channel->readers.empty())
            CE_UNLIKELY { return false; }
 
        Reader* r = channel->readers.pop();
        // exchange address & resumeable_handle
        std::swap(this->value, r->value);
        std::swap(this->handle, r->handle);
 
        return true;
    }
 
    template<typename T, typename Lockable> void Channel<T, Lockable>::Writer::await_suspend(CoroutineHandle coro) & {
        // notice that next & chan are sharing memory
        Channel& ch = *(this->channel);
 
        this->handle = coro; // remember handle before push/unlock
        this->next = nullptr; // clear to prevent confusing
 
        ch.writers.push(this); // push to channel
        // ch.mutex.unlock();
    }
 
    template<typename T, typename Lockable> bool Channel<T, Lockable>::Writer::await_resume() & {
        // frame holds poision if the channel is going to destroy
        if(!handle)
            CE_UNLIKELY { return false; }
 
        if(const auto coroutine = *handle) {
            //            assert(this->handle != nullptr);
            //            assert(*reinterpret_cast<UInt64*>(handle) != 0);
            coroutine.resume();
        }
        return true;
    }
 
} // namespace CeresEngine