diff options
author | Claudius "keldu" Holeksa <mail@keldu.de> | 2023-12-19 14:04:04 +0100 |
---|---|---|
committer | Claudius "keldu" Holeksa <mail@keldu.de> | 2023-12-19 14:04:04 +0100 |
commit | 55dd66d8dd773ed98580d1fac0e15da6244d35b8 (patch) | |
tree | 0040ab8535855f3d25412089876021192df9d242 /modules/async/async.h | |
parent | 9de31a58c0ba5d7045ee5a9e1436cb01906fc6de (diff) |
moving async to c++ folder
Diffstat (limited to 'modules/async/async.h')
-rw-r--r-- | modules/async/async.h | 1023 |
1 files changed, 0 insertions, 1023 deletions
diff --git a/modules/async/async.h b/modules/async/async.h deleted file mode 100644 index 66afa38..0000000 --- a/modules/async/async.h +++ /dev/null @@ -1,1023 +0,0 @@ -#pragma once - -#include <forstio/common.h> -#include <forstio/error.h> - -#include <chrono> -#include <functional> -#include <limits> -#include <list> -#include <queue> -#include <type_traits> - -namespace saw { -class conveyor_storage; -class conveyor_node { -public: - conveyor_node(); - virtual ~conveyor_node() = default; - - /** - * Internal method to retrieve results from children - */ - virtual void get_result(error_or_value &err_or_val) = 0; - - /** - * Swap out child with another one - */ - virtual error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee_) = 0; - - /** - * Retrieve the next storage node - */ - virtual conveyor_storage *next_storage() = 0; - - /** - * Notify that a new parent was attached - * Only relevant for the feeding nodes - */ - virtual void notify_parent_attached(conveyor_node &){}; -}; - -class conveyor_node_with_child_mixin final { -public: - own<conveyor_node> child = nullptr; - - conveyor_node_with_child_mixin(own<conveyor_node> &&child_, - conveyor_node &owner_); - ~conveyor_node_with_child_mixin() = default; - - /** - * Swap out children and return the child ptr, since the caller is the child - * itself. Stack needs to be cleared before the child is destroyed, so the - * swapped out node is returned as well. - */ - error_or<own<conveyor_node>> swap_child(own<conveyor_node> &&swapee); -}; - -class conveyor_node_with_parent_mixin final { -public: - conveyor_node *parent = nullptr; - - error_or<own<conveyor_node>> - swap_child_of_parent(own<conveyor_node> &&swapee) { - SAW_ASSERT(parent) { - return make_error<err::invalid_state>( - "Can't swap child, because parent doesn't exist"); - } - - return parent->swap_child(std::move(swapee)); - } - void change_parent(conveyor_node *p) { parent = p; } -}; - -class event_loop; -class wait_scope; -/* - * Event class similar to capn'proto. - * https://github.com/capnproto/capnproto - */ -class event { -private: - event_loop &loop_; - event **prev_ = nullptr; - event *next_ = nullptr; - - friend class event_loop; - -public: - event(); - event(event_loop &loop); - virtual ~event(); - - virtual void fire() = 0; - - void arm_next(); - void arm_later(); - void arm_last(); - void disarm(); - - bool is_armed() const; -}; - -class conveyor_storage { -protected: - conveyor_storage *parent_ = nullptr; - -public: - conveyor_storage(); - virtual ~conveyor_storage(); - - virtual size_t space() const = 0; - virtual size_t queued() const = 0; - virtual void child_has_fired() = 0; - virtual void parent_has_fired() = 0; - - virtual void set_parent(conveyor_storage *parent) = 0; - conveyor_storage *get_parent() const; -}; - -class conveyor_event_storage : public conveyor_storage, public event { -public: - conveyor_event_storage(); - virtual ~conveyor_event_storage() = default; - - void set_parent(conveyor_storage *parent) override; -}; - -class conveyor_base { -protected: - own<conveyor_node> node_; - -public: - conveyor_base(own<conveyor_node> &&node_p); - virtual ~conveyor_base() = default; - - conveyor_base(conveyor_base &&) = default; - conveyor_base &operator=(conveyor_base &&) = default; - - void get(error_or_value &err_or_val); -}; - -template <typename T> class conveyor; - -template <typename T> conveyor<T> chained_conveyor_type(T *); - -// template <typename T> Conveyor<T> chainedConveyorType(Conveyor<T> *); - -template <typename T> T remove_error_or_type(T *); - -template <typename T> T remove_error_or_type(error_or<T> *); - -template <typename T> -using remove_error_or = decltype(remove_error_or_type((T *)nullptr)); - -template <typename T> -using chained_conveyors = decltype(chained_conveyor_type((T *)nullptr)); - -template <typename Func, typename T> -using conveyor_result = - chained_conveyors<remove_error_or<return_type<Func, T>>>; - -struct propagate_error { -public: - error operator()(const error &err) const; - error operator()(error &&err); -}; - -class conveyor_sink { -private: - own<conveyor_node> node_; - -public: - conveyor_sink(); - conveyor_sink(own<conveyor_node> &&node); - - conveyor_sink(conveyor_sink &&) = default; - conveyor_sink &operator=(conveyor_sink &&) = default; -}; - -template <typename T> class merge_conveyor_node_data; - -template <typename T> class merge_conveyor { -private: - lent<merge_conveyor_node_data<T>> data_; - -public: - merge_conveyor() = default; - merge_conveyor(lent<merge_conveyor_node_data<T>> d); - ~merge_conveyor(); - - void attach(conveyor<T> conv); -}; - -/** - * Main interface for async operations. - */ -template <typename T> class conveyor final : public conveyor_base { -public: - /** - * Construct an immediately fulfilled node - */ - conveyor(fix_void<T> value); - - /** - * Construct an immediately failed node - */ - conveyor(error &&err); - - /** - * Construct a conveyor with a child node - */ - conveyor(own<conveyor_node> node_p); - - conveyor(conveyor<T> &&) = default; - conveyor<T> &operator=(conveyor<T> &&) = default; - - /** - * This method converts values or errors from children - */ - template <typename Func, typename ErrorFunc = propagate_error> - [[nodiscard]] conveyor_result<Func, T> - then(Func &&func, ErrorFunc &&error_func = propagate_error()); - - /** - * This method adds a buffer node in the conveyor chains which acts as a - * scheduler interrupt point and collects elements up to the supplied limit. - */ - [[nodiscard]] conveyor<T> - buffer(size_t limit = std::numeric_limits<size_t>::max()); - - /** - * This method just takes ownership of any supplied types, - * which are destroyed when the chain gets destroyed. - * Useful for resource lifetime control. - */ - template <typename... Args> - [[nodiscard]] conveyor<T> attach(Args &&...args); - - /** @todo implement - * This method limits the total amount of passed elements - * Be careful where you place this node into the chain. - * If you meant to fork it and destroy paths you shouldn't place - * an interrupt point between the fork and this limiter - */ - [[nodiscard]] conveyor<T> limit(size_t val = 1); - - /** - * - */ - [[nodiscard]] std::pair<conveyor<T>, merge_conveyor<T>> merge(); - - /** - * Moves the conveyor chain into a thread local storage point which drops - * every element. Use sink() if you want to control the lifetime of a - * conveyor chain - */ - template <typename ErrorFunc = propagate_error> - void detach(ErrorFunc &&err_func = propagate_error()); - /** - * Creates a local sink which drops elements, but lifetime control remains - * in your hand. - */ - template <typename ErrorFunc = propagate_error> - [[nodiscard]] conveyor_sink - sink(ErrorFunc &&error_func = propagate_error()); - - /** - * If no sink() or detach() is used you have to take elements out of the - * chain yourself. - */ - error_or<fix_void<T>> take(); - - /** @todo implement - * Specifically pump elements through this chain with the provided - * wait_scope - */ - void poll(wait_scope &wait_scope); - - // helper - static conveyor<T> to_conveyor(own<conveyor_node> node); - - // helper - static own<conveyor_node> from_conveyor(conveyor<T> conveyor); -}; - -template <typename Func> conveyor_result<Func, void> exec_later(Func &&func); - -/* - * Join Conveyors into a single one - */ -template <typename... Args> -conveyor<std::tuple<Args...>> -join_conveyors(std::tuple<conveyor<Args>...> &conveyors); - -template <typename T> class conveyor_feeder { -public: - virtual ~conveyor_feeder() = default; - - virtual void feed(T &&data) = 0; - virtual void fail(error &&error) = 0; - - virtual size_t space() const = 0; - virtual size_t queued() const = 0; - - virtual error swap(conveyor<T> &&conveyor) noexcept = 0; -}; - -template <> class conveyor_feeder<void> { -public: - virtual ~conveyor_feeder() = default; - - virtual void feed(void_t &&value = void_t{}) = 0; - virtual void fail(error &&error) = 0; - - virtual size_t space() const = 0; - virtual size_t queued() const = 0; - - virtual error swap(conveyor<void_t> &&conveyor) noexcept = 0; -}; - -template <typename T> struct conveyor_and_feeder { - own<conveyor_feeder<T>> feeder; - class conveyor<T> conveyor; -}; - -template <typename T> conveyor_and_feeder<T> new_conveyor_and_feeder(); - -template <typename T> conveyor_and_feeder<T> one_time_conveyor_and_feeder(); - -enum class Signal : uint8_t { Terminate, User1 }; - -/** - * Class which acts as a correspondent between the running framework and outside - * events which may be signals from the operating system or just other threads. - * Default EventPorts are supplied by setupAsyncIo() in io.h - */ -class event_port { -public: - virtual ~event_port() = default; - - virtual conveyor<void> on_signal(Signal signal) = 0; - - virtual void poll() = 0; - virtual void wait() = 0; - virtual void wait(const std::chrono::steady_clock::duration &) = 0; - virtual void wait(const std::chrono::steady_clock::time_point &) = 0; - - virtual void wake() = 0; -}; - -class sink_conveyor_node; - -class conveyor_sink_set final : public event { -private: - /* - class Helper final : public Event { - private: - void destroySinkConveyorNode(ConveyorNode& sink); - void fail(Error&& error); - - std::vector<Own<ConveyorNode>> sink_nodes; - std::queue<ConveyorNode*> delete_nodes; - std::function<void(Error&& error)> error_handler; - - public: - ConveyorSinks() = default; - ConveyorSinks(EventLoop& event_loop); - - void add(Conveyor<void> node); - - void fire() override {} - }; - - gin::Own<Helper> helper; - */ - friend class sink_conveyor_node; - - void destroy_sink_conveyor_node(conveyor_node &sink_node); - void fail(error &&err); - - std::list<own<conveyor_node>> sink_nodes_; - - std::queue<conveyor_node *> delete_nodes_; - - std::function<void(error &&)> error_handler_; - -public: - // ConveyorSinks(); - // ConveyorSinks(EventLoop& event_loop); - conveyor_sink_set() = default; - conveyor_sink_set(event_loop &event_loop); - - void add(conveyor<void> &&node); - - void fire() override; -}; - -/* - * EventLoop class similar to capn'proto. - * https://github.com/capnproto/capnproto - */ -class event_loop { -private: - friend class event; - event *head_ = nullptr; - event **tail_ = &head_; - event **next_insert_point_ = &head_; - event **later_insert_point_ = &head_; - - bool is_runnable_ = false; - - own<event_port> event_port_ = nullptr; - - own<conveyor_sink_set> daemon_sink_ = nullptr; - - // functions - void set_runnable(bool runnable); - - friend class wait_scope; - void enter_scope(); - void leave_scope(); - - bool turn_loop(); - bool turn(); - -public: - event_loop(); - event_loop(own<event_port> &&port); - ~event_loop(); - - event_loop(event_loop &&) = default; - event_loop &operator=(event_loop &&) = default; - - bool wait(); - bool wait(const std::chrono::steady_clock::duration &); - bool wait(const std::chrono::steady_clock::time_point &); - bool poll(); - - event_port *get_event_port(); - - conveyor_sink_set &daemon(); -}; - -/* - * WaitScope class similar to capn'proto. - * https://github.com/capnproto/capnproto - */ -class wait_scope { -private: - event_loop &loop_; - -public: - wait_scope(event_loop &loop); - ~wait_scope(); - - void wait(); - void wait(const std::chrono::steady_clock::duration &); - void wait(const std::chrono::steady_clock::time_point &); - void poll(); -}; - -template <typename Func> conveyor_result<Func, void> yield_next(Func &&func); - -template <typename Func> conveyor_result<Func, void> yield_later(Func &&func); - -template <typename Func> conveyor_result<Func, void> yield_last(Func &&func); -} // namespace saw - -// Secret stuff -// Aka private semi hidden classes -namespace saw { - -template <typename Out, typename In> struct fix_void_caller { - template <typename Func> static Out apply(Func &func, In &&in) { - return func(std::move(in)); - } -}; - -template <typename Out> struct fix_void_caller<Out, void_t> { - template <typename Func> static Out apply(Func &func, void_t &&in) { - (void)in; - return func(); - } -}; - -template <typename In> struct fix_void_caller<void_t, In> { - template <typename Func> static void_t apply(Func &func, In &&in) { - func(std::move(in)); - return void_t{}; - } -}; - -template <> struct fix_void_caller<void_t, void_t> { - template <typename Func> static void_t apply(Func &func, void_t &&in) { - (void)in; - func(); - return void_t{}; - } -}; - -template <typename T> class adapt_conveyor_node; - -template <typename T> -class adapt_conveyor_feeder final : public conveyor_feeder<unfix_void<T>> { -private: - adapt_conveyor_node<T> *feedee_ = nullptr; - -public: - ~adapt_conveyor_feeder(); - - void set_feedee(adapt_conveyor_node<T> *feedee); - - void feed(T &&value) override; - void fail(error &&error) override; - - size_t space() const override; - size_t queued() const override; - - error swap(conveyor<T> &&conv) noexcept override; -}; - -template <typename T> -class adapt_conveyor_node final : public conveyor_node, - public conveyor_event_storage { -private: - adapt_conveyor_feeder<T> *feeder_ = nullptr; - - std::queue<error_or<unfix_void<T>>> storage_; - - conveyor_node_with_parent_mixin parent_node_; - -public: - adapt_conveyor_node(); - ~adapt_conveyor_node(); - - void set_feeder(adapt_conveyor_feeder<T> *feeder); - - void feed(T &&value); - void fail(error &&error); - - // ConveyorNode - void get_result(error_or_value &err_or_val) override; - - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee) noexcept override; - - conveyor_storage *next_storage() noexcept override; - void notify_parent_attached(conveyor_node &) noexcept override; - - // ConveyorStorage - size_t space() const override; - size_t queued() const override; - - void child_has_fired() override; - void parent_has_fired() override; - - // Event - void fire() override; -}; - -template <typename T> class one_time_conveyor_node; - -template <typename T> -class one_time_conveyor_feeder final : public conveyor_feeder<unfix_void<T>> { -private: - one_time_conveyor_node<T> *feedee_ = nullptr; - -public: - ~one_time_conveyor_feeder(); - - void set_feedee(one_time_conveyor_node<T> *feedee); - - void feed(T &&value) override; - void fail(error &&error) override; - - size_t space() const override; - size_t queued() const override; -}; - -template <typename T> -class one_time_conveyor_node final : public conveyor_node, - public conveyor_storage, - public event { -private: - one_time_conveyor_feeder<T> *feeder_ = nullptr; - - bool passed_ = false; - maybe<error_or<T>> storage_ = std::nullopt; - -public: - ~one_time_conveyor_node(); - - void set_feeder(one_time_conveyor_feeder<T> *feeder); - - void feed(T &&value); - void fail(error &&error); - - // ConveyorNode - void get_result(error_or_value &err_or_val) override; - - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee) override; - - // ConveyorStorage - size_t space() const override; - size_t queued() const override; - - void child_has_fired() override {} - void parent_has_fired() override; - - // Event - void fire() override; -}; - -/** - * This class buffers and saves incoming elements and acts as an interrupt node - * for processing calls - */ -class queue_buffer_conveyor_node_base : public conveyor_node, - public conveyor_event_storage { -protected: - conveyor_node_with_child_mixin child_mixin_; - -public: - queue_buffer_conveyor_node_base(own<conveyor_node> child_) - : conveyor_event_storage{}, child_mixin_{std::move(child_), *this} {} - virtual ~queue_buffer_conveyor_node_base() = default; - - /** - * Use mixin - */ - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee_) noexcept override; - - conveyor_storage *next_storage() noexcept override { - return static_cast<conveyor_storage *>(this); - } -}; - -template <typename T> -class queue_buffer_conveyor_node final - : public queue_buffer_conveyor_node_base { -private: - std::queue<error_or<T>> storage_; - size_t max_store_; - -public: - queue_buffer_conveyor_node(own<conveyor_node> dep, size_t max_size) - : queue_buffer_conveyor_node_base{std::move(dep)}, max_store_{ - max_size} {} - // Event - void fire() override; - // ConveyorNode - void get_result(error_or_value &eov) noexcept override; - - // ConveyorStorage - size_t space() const override; - size_t queued() const override; - - void child_has_fired() override; - void parent_has_fired() override; -}; - -class attach_conveyor_node_base : public conveyor_node { -protected: - conveyor_node_with_child_mixin child_mixin_; - -public: - attach_conveyor_node_base(own<conveyor_node> &&child_) - : child_mixin_{std::move(child_), *this} {} - - virtual ~attach_conveyor_node_base() = default; - - void get_result(error_or_value &err_or_val) noexcept override; - - /** - * Use mixin - */ - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee_) noexcept override; - - conveyor_storage *next_storage() noexcept override; -}; - -template <typename... Args> -class attach_conveyor_node final : public attach_conveyor_node_base { -public: - attach_conveyor_node(own<conveyor_node> &&dep, Args &&...args) - : attach_conveyor_node_base(std::move(dep)), attached_data_{ - std::move(args...)} {} - -private: - std::tuple<Args...> attached_data_; -}; - -class convert_conveyor_node_base : public conveyor_node { -public: - convert_conveyor_node_base(own<conveyor_node> &&dep); - virtual ~convert_conveyor_node_base() = default; - - void get_result(error_or_value &err_or_val) override; - - virtual void get_impl(error_or_value &err_or_val) = 0; - - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee) noexcept override; - - conveyor_storage *next_storage() noexcept override; - -protected: - conveyor_node_with_child_mixin child_mixin_; -}; - -template <typename T, typename DepT, typename Func, typename ErrorFunc> -class convert_conveyor_node final : public convert_conveyor_node_base { -private: - Func func_; - ErrorFunc error_func_; - - static_assert(std::is_same<DepT, remove_error_or<DepT>>::value, - "Should never be of type ErrorOr"); - -public: - convert_conveyor_node(own<conveyor_node> &&dep, Func &&func, - ErrorFunc &&error_func) - : convert_conveyor_node_base(std::move(dep)), func_{std::move(func)}, - error_func_{std::move(error_func)} {} - - void get_impl(error_or_value &err_or_val) noexcept override { - error_or<unfix_void<DepT>> dep_eov; - error_or<unfix_void<remove_error_or<T>>> &eov = - err_or_val.as<unfix_void<remove_error_or<T>>>(); - if (child_mixin_.child) { - child_mixin_.child->get_result(dep_eov); - if (dep_eov.is_value()) { - try { - - eov = fix_void_caller<T, DepT>::apply( - func_, std::move(dep_eov.get_value())); - } catch (const std::bad_alloc &) { - eov = make_error<err::out_of_memory>("Out of memory"); - } catch (const std::exception &) { - eov = make_error<err::invalid_state>( - "Exception in chain occured. Return ErrorOr<T> if you " - "want to handle errors which are recoverable"); - } - } else if (dep_eov.is_error()) { - eov = error_func_(std::move(dep_eov.get_error())); - } else { - eov = make_error<err::invalid_state>("No value set in dependency"); - } - } else { - eov = make_error<err::invalid_state>("Conveyor doesn't have child"); - } - } -}; - -class sink_conveyor_node final : public conveyor_node, - public conveyor_event_storage { -private: - conveyor_node_with_child_mixin child_mixin_; - conveyor_sink_set *conveyor_sink_; - -public: - sink_conveyor_node(own<conveyor_node> node, conveyor_sink_set &conv_sink) - : conveyor_event_storage{}, child_mixin_{std::move(node), *this}, - conveyor_sink_{&conv_sink} {} - - sink_conveyor_node(own<conveyor_node> node) - : conveyor_event_storage{}, child_mixin_{std::move(node), *this}, - conveyor_sink_{nullptr} {} - - // Event only queued if a critical error occured - void fire() override { - // Queued for destruction of children, because this acts as a sink and - // no other event should be here - child_mixin_.child = nullptr; - - if (conveyor_sink_) { - conveyor_sink_->destroy_sink_conveyor_node(*this); - conveyor_sink_ = nullptr; - } - } - - // ConveyorStorage - size_t space() const override { return 1; } - size_t queued() const override { return 0; } - - // ConveyorNode - void get_result(error_or_value &err_or_val) noexcept override { - err_or_val.as<void_t>() = - make_error<err::invalid_state>("In a sink node no result can be returned"); - } - - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee) noexcept override { - return child_mixin_.swap_child(std::move(swapee)); - } - - // ConveyorStorage - void child_has_fired() override { - if (child_mixin_.child) { - error_or<void> dep_eov; - child_mixin_.child->get_result(dep_eov); - if (dep_eov.is_error()) { - if (dep_eov.get_error().is_critical()) { - if (!is_armed()) { - arm_last(); - } - } - if (conveyor_sink_) { - conveyor_sink_->fail(std::move(dep_eov.get_error())); - } - } - } - } - - /* - * No parent needs to be fired since we always have space - */ - void parent_has_fired() override {} - - conveyor_storage *next_storage() override { - // Should never happen though - assert(false); - return nullptr; - // return static_cast<ConveyorStorage*>(this); - } -}; - -class immediate_conveyor_node_base : public conveyor_node, - public conveyor_event_storage { -private: -public: - immediate_conveyor_node_base(); - - virtual ~immediate_conveyor_node_base() = default; - - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee) noexcept override { - (void)swapee; - return make_error<err::not_supported>("Node doesn't support swapping"); - } - - conveyor_storage *next_storage() noexcept override { - return static_cast<conveyor_storage *>(this); - } -}; - -template <typename T> -class immediate_conveyor_node final : public immediate_conveyor_node_base { -private: - error_or<fix_void<T>> value_; - uint8_t retrieved_; - -public: - immediate_conveyor_node(fix_void<T> &&val); - immediate_conveyor_node(error &&error); - - // ConveyorStorage - size_t space() const override; - size_t queued() const override; - - void child_has_fired() override; - void parent_has_fired() override; - - // ConveyorNode - void get_result(error_or_value &err_or_val) noexcept override { - if (retrieved_ > 0) { - err_or_val.as<fix_void<T>>() = - make_error<err::buffer_exhausted>("Already taken value"); - } else { - err_or_val.as<fix_void<T>>() = std::move(value_); - } - if (queued() > 0) { - ++retrieved_; - } - } - - // Event - void fire() override; -}; - -/* - * Collects every incoming value and throws it in one lane - */ -class merge_conveyor_node_base : public conveyor_node, - public conveyor_event_storage { -public: - merge_conveyor_node_base(); - - virtual ~merge_conveyor_node_base() = default; - - conveyor_storage *next_storage() noexcept override { - return static_cast<conveyor_storage *>(this); - } -}; - -template <typename T> -class merge_conveyor_node : public merge_conveyor_node_base { -private: - class appendage final : public conveyor_node, public conveyor_storage { - public: - own<conveyor_node> child; - merge_conveyor_node *merger; - - maybe<error_or<fix_void<T>>> error_or_value_; - - public: - appendage(own<conveyor_node> n, merge_conveyor_node &m) - : conveyor_storage{}, child{std::move(n)}, merger{&m}, - error_or_value_{std::nullopt} {} - - bool child_storage_has_element_queued() const { - if (!child) { - return false; - } - conveyor_storage *storage = child->next_storage(); - if (storage) { - return storage->queued() > 0; - } - return false; - } - - void get_appendage_result(error_or_value &eov); - - /** - * ConveyorNode - */ - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&swapee_) override; - - conveyor_storage *next_storage() noexcept override { - return static_cast<conveyor_storage *>(this); - } - - void get_result(error_or_value &err_or_val) override; - - /** - * ConveyorStorage - */ - size_t space() const override; - - size_t queued() const override; - - void child_has_fired() override; - - void parent_has_fired() override; - - void set_parent(conveyor_storage *par) override; - }; - - friend class merge_conveyor_node_data<T>; - friend class appendage; - - our<merge_conveyor_node_data<T>> data_; - size_t next_appendage_ = 0; - -public: - merge_conveyor_node(our<merge_conveyor_node_data<T>> data); - ~merge_conveyor_node(); - // ConveyorNode - error_or<own<conveyor_node>> - swap_child(own<conveyor_node> &&c) noexcept override; - - // Event - void get_result(error_or_value &err_or_val) noexcept override; - - void fire() override; - - // ConveyorStorage - size_t space() const override; - size_t queued() const override; - void child_has_fired() override; - void parent_has_fired() override; -}; - -template <typename T> class merge_conveyor_node_data { -public: - std::vector<own<typename merge_conveyor_node<T>::appendage>> appendages; - - merge_conveyor_node<T> *merger = nullptr; - -public: - void attach(conveyor<T> conv); - - void governing_node_destroyed(); -}; - -/* -class JoinConveyorNodeBase : public ConveyorNode, public ConveyorEventStorage { -private: - -public: -}; - -template <typename... Args> -class JoinConveyorNode final : public JoinConveyorNodeBase { -private: - template<typename T> - class Appendage : public ConveyorEventStorage { - private: - Maybe<T> data = std::nullopt; - - public: - size_t space() const override; - size_t queued() const override; - - void fire() override; - void get_result(ErrorOrValue& eov) override; - }; - - std::tuple<Appendage<Args>...> appendages; - -public: -}; - -*/ - -} // namespace saw - -#include "async.tmpl.h" |