ecor.hpp

 
 
#pragma once
 
#include <atomic>
#include <bit>
#include <concepts>
#include <coroutine>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <iterator>
#include <limits>
#include <memory>
#include <new>
#include <optional>
#include <span>
#include <type_traits>
#include <utility>
#include <variant>
#include <zll.hpp>
 
#ifdef ECOR_DEFAULT_ASSERT
 
#include <cassert>
#define ECOR_ASSERT( expr ) assert( expr )
 
#else
 
#ifndef ECOR_ASSERT
#define ECOR_ASSERT( expr ) ( (void) ( ( expr ) ) )
#endif
 
#endif
 
namespace ecor
{
 
template < typename T >
struct _tag
{
};
 
template < typename T, typename... Ts >
struct _contains_type;
 
template < typename T, typename U, typename... Ts >
        requires( std::same_as< T, U > )
struct _contains_type< T, U, Ts... >
{
        static constexpr bool value = true;
};
 
template < typename T, typename U, typename... Ts >
        requires( !std::same_as< T, U > )
struct _contains_type< T, U, Ts... > : _contains_type< T, Ts... >
{
};
 
template < typename T >
struct _contains_type< T >
{
        static constexpr bool value = false;
};
 
struct sender_t
{
};
 
struct receiver_t
{
};
 
struct operation_state_t
{
};
 
struct set_value_t
{
};
 
struct set_error_t
{
};
 
struct set_stopped_t
{
};
 
struct _get_stopped_t
{
};
 
template < typename... S >
struct completion_signatures
{
};
 
template < template < typename... > class C, typename T >
struct _apply_to_sigs;
 
template < template < typename... > class C, typename... Ts >
struct _apply_to_sigs< C, completion_signatures< Ts... > >
{
        using type = C< Ts... >;
};
 
template < template < typename... > class C, typename T >
using _apply_to_sigs_t = typename _apply_to_sigs< C, T >::type;
 
struct empty_env
{
};
 
struct noop_base
{
};
 
template < typename S, typename R >
using connect_type = decltype( std::move( std::declval< S >() ).connect( std::declval< R >() ) );
 
 
template < typename E >
struct with_error
{
        using type = std::remove_cvref_t< E >;
        E error;
 
        explicit with_error( E e )
          : error( std::move( e ) )
        {
        }
};
 
template < typename T >
struct _is_signature : std::false_type
{
};
 
template < typename... Args >
struct _is_signature< set_value_t( Args... ) > : std::true_type
{
};
template < typename Err >
struct _is_signature< set_error_t( Err ) > : std::true_type
{
};
template <>
struct _is_signature< set_stopped_t() > : std::true_type
{
};
template <>
struct _is_signature< _get_stopped_t() > : std::true_type
{
};
 
template < typename T >
concept signature = _is_signature< T >::value;
 
template < typename T >
struct _value_setter
{
        using type = set_value_t( T );
};
 
template <>
struct _value_setter< void >
{
        using type = set_value_t();
};
 
template < typename T >
using _value_setter_t = typename _value_setter< T >::type;
 
template < typename... S >
struct _multivalue_value_signature : std::false_type
{
};
 
template < typename T, typename U, typename... Ts >
struct _multivalue_value_signature< set_value_t( T, U, Ts... ) > : std::true_type
{
};
 
template < typename... S >
concept all_value_signatures_singular = ( ... && !_multivalue_value_signature< S >::value );
 
template < typename T >
struct _is_all_singular;
 
template < typename... S >
struct _is_all_singular< completion_signatures< S... > >
{
        static constexpr bool value = all_value_signatures_singular< S... >;
};
 
template < typename Sigs >
static constexpr bool is_all_singular_v = _is_all_singular< Sigs >::value;
 
 
// ------------------------------------------------------------------------------
 
template < typename T >
constexpr T _align_idx( uint8_t const* buff, T idx, std::size_t align ) noexcept
{
        auto* p = buff + idx;
        auto  a = ( (uintptr_t) p ) % align;
        if ( a )
                p += align - a;
        return (T) ( p - buff );
}
 
template < std::size_t N >
constexpr auto _pick_index_type() noexcept
{
        if constexpr ( N <= 0xff )
                return uint8_t{};
        else if constexpr ( N <= 0xffff )
                return uint16_t{};
        else if constexpr ( N <= 0xffffffff )
                return uint32_t{};
        else
                return;
}
 
template < std::size_t N >
using smallest_index_type = decltype( _pick_index_type< N >() );
 
 
struct _allocate_t
{
        template < typename T >
                requires( requires( T x ) { x.allocate( std::size_t{}, std::size_t{} ); } )
        void* operator()( T& x, std::size_t bytes, std::size_t align ) const
            noexcept( noexcept( x.allocate( bytes, align ) ) )
        {
                return x.allocate( bytes, align );
        }
};
static constexpr _allocate_t allocate{};
 
 
struct _deallocate_t
{
        template < typename T >
                requires( requires( T x, void* p, std::size_t b, std::size_t a ) {
                        x.deallocate( p, b, a );
                } )
        void operator()( T& x, void* p, std::size_t bytes, std::size_t align ) const
            noexcept( noexcept( x.deallocate( p, bytes, align ) ) )
        {
                x.deallocate( p, bytes, align );
        }
};
static constexpr _deallocate_t deallocate{};
 
template < typename T >
concept memory_resource = requires( T a, std::size_t bytes, std::size_t align, void* p ) {
        { allocate( a, bytes, align ) } -> std::same_as< void* >;
        { deallocate( a, p, bytes, align ) } -> std::same_as< void >;
};
 
 
template <
    typename T,
    std::size_t Extent = std::dynamic_extent,
    typename Deleter   = std::default_delete< T[] > >
struct unique_span : std::span< T, Extent >
{
        using span_type = std::span< T, Extent >;
 
        unique_span() noexcept = default;
 
        unique_span( T* data, std::size_t size, Deleter deleter = Deleter{} ) noexcept(
            std::is_nothrow_move_constructible_v< Deleter > )
          : span_type( data, size )
          , _deleter( std::move( deleter ) )
        {
        }
 
        unique_span( unique_span const& )            = delete;
        unique_span& operator=( unique_span const& ) = delete;
 
        unique_span( unique_span&& other ) noexcept
          : span_type( other.data(), other.size() )
          , _deleter( std::move( other._deleter ) )
        {
                *(span_type*) &other = span_type{};
        }
 
        unique_span& operator=( unique_span&& other ) noexcept
        {
                if ( this != &other ) {
                        if ( this->data() )
                                _deleter( this->data() );
                        *(span_type*) this   = span_type( other.data(), other.size() );
                        _deleter             = std::move( other._deleter );
                        *(span_type*) &other = span_type{};
                }
                return *this;
        }
 
        std::span< T, Extent > release() noexcept
        {
                T*          data   = this->data();
                std::size_t size   = this->size();
                *(span_type*) this = span_type{};
                return span_type( data, size );
        }
 
        ~unique_span()
        {
                if ( this->data() )
                        _deleter( this->data() );
        }
 
private:
        [[no_unique_address]] Deleter _deleter;
};
 
template < typename IndexType, typename Base = noop_base >
struct circular_buffer_memory : Base
{
        static_assert(
            std::is_unsigned< IndexType >::value,
            "IndexType must be an unsigned integer type" );
 
        using index_type = IndexType;
 
        template < std::size_t N >
                requires( std::numeric_limits< IndexType >::max() >= N )
        constexpr circular_buffer_memory( std::span< uint8_t, N > b ) noexcept
          : _buff( std::move( b ) )
        {
        }
        template < std::size_t N >
                requires( std::numeric_limits< IndexType >::max() >= N )
        constexpr circular_buffer_memory( uint8_t ( &b )[N] ) noexcept
          : _buff( b )
        {
        }
 
        struct _deleter
        {
                circular_buffer_memory< IndexType, Base >* mem;
 
                template < typename T >
                void operator()( T* p ) noexcept
                {
                        if ( p ) {
                                p->~T();
                                mem->deallocate( p );
                        }
                }
        };
 
        template < typename T >
        using uptr = std::unique_ptr< T, _deleter >;
 
        template < typename T, typename... Args >
        uptr< T > make( Args&&... args )
        {
                void* p = allocate( sizeof( T ), alignof( T ) );
                if ( !p )
                        return { nullptr, _deleter{ this } };
                T* t = new ( p ) T( (Args&&) args... );
                return { t, _deleter{ this } };
        }
 
        template < typename T, std::size_t Extent = std::dynamic_extent >
        using uspan = unique_span< T, Extent, _deleter >;
 
        template < typename T >
        uspan< T > make_span( std::size_t n )
        {
                void* p = allocate( sizeof( T ) * n, alignof( T ) );
                if ( !p )
                        return uspan< T >{ nullptr, 0, _deleter{ this } };
                auto* pp = new ( p ) T[n]();
                return uspan< T >{ pp, n, _deleter{ this } };
        }
 
        template < typename T, std::size_t Extent >
        uspan< T, Extent > make_span()
        {
                constexpr std::size_t n  = Extent;
                auto                  sp = make_span< T >( n );
                return sp;
        }
 
        [[nodiscard]] void* allocate( std::size_t bytes, std::size_t align ) noexcept
        {
                auto* p = _allocate( _buff, bytes, align );
                return p;
        }
 
        void deallocate( void* p, std::size_t bytes, std::size_t align ) noexcept
        {
                std::ignore = bytes;
                std::ignore = align;
                _deallocate( _buff.data(), (uint8_t*) p );
        }
 
        void deallocate( void* p ) noexcept
        {
                _deallocate( _buff.data(), (uint8_t*) p );
        }
 
        [[nodiscard]] std::size_t capacity() const noexcept
        {
                return _buff.size();
        }
 
        [[nodiscard]] std::size_t used_bytes() const noexcept
        {
                if ( _first == npos )
                        return 0;
                if ( _first <= _last )
                        return ( _next - _first );
                return ( _buff.size() - _first ) + _next;
        }
 
        static constexpr index_type npos = std::numeric_limits< index_type >::max();
 
        struct _node
        {
                index_type next_idx = npos;
                index_type prev_idx = npos;
        };
 
private:
 
        // Pick index where to allocate `bytes` with `align`, returns npos if no space
        // is available. The returned index points to the node header, the actual data
        // starts after sizeof(node).
        [[nodiscard]] index_type _pick_index(
            std::span< uint8_t > buff,
            std::size_t          bytes,
            std::size_t          align ) const noexcept
        {
                index_type idx      = npos;
                int        capacity = 0;
                if ( _last == npos ) {
                        // Empty buffer
                        ECOR_ASSERT( _first == npos && _last == npos );
                        idx      = _align_idx( buff.data(), sizeof( _node ), align );
                        capacity = buff.size() - idx;
                } else if ( _first <= _next ) {
                        // Non-overflow state: [ ][x][x][x][ ][ ]
                        {
                                idx = _align_idx( buff.data(), _next + sizeof( _node ), align );
                                capacity = buff.size() - idx;
                                if ( idx < buff.size() && capacity >= (int) bytes )
                                        return idx - sizeof( _node );
                        }
                        // Non-overflow state, but overflow triggered: [ ][ ][ ][x][x][x]
                        {
                                idx      = _align_idx( buff.data(), sizeof( _node ), align );
                                capacity = idx - _first;
                        }
 
                } else {  // _first > _next
                          // Overflow state: [x][x][ ][ ][x][x]
                        idx      = _align_idx( buff.data(), _next + sizeof( _node ), align );
                        capacity = _first - idx;
                }
                if ( idx < buff.size() && capacity >= (int) bytes )
                        return idx - sizeof( _node );
                return npos;
        }
 
        void _set_node( uint8_t* buff, index_type idx, _node const& n ) noexcept
        {
                std::memcpy( buff + idx, &n, sizeof( _node ) );
        }
 
        void* _allocate( std::span< uint8_t > buff, std::size_t bytes, std::size_t align ) noexcept
        {
                auto idx = _pick_index( buff, bytes, align );
                if ( idx == npos )
                        return nullptr;
                auto* p = buff.data() + idx;
                if ( _first == npos ) {
                        _first = idx;
                } else {
                        auto nn     = _get_node( buff.data(), _last );
                        nn.next_idx = idx;
                        _set_node( buff.data(), _last, nn );
                }
                _node const n{
                    .next_idx = npos,
                    .prev_idx = _last,
                };
                _last = idx;
                _next = idx + sizeof( _node ) + bytes;
                _set_node( buff.data(), idx, n );
                return p + sizeof( _node );
        }
 
        void _deallocate( uint8_t* buff, void* p ) noexcept
        {
                auto*            pp  = (uint8_t*) p - sizeof( _node );
                index_type const idx = pp - buff;
                _node const      n   = _get_node( buff, idx );
 
                if ( _first == idx ) {
                        _first = n.next_idx;
                } else {
                        auto pn     = _get_node( buff, n.prev_idx );
                        pn.next_idx = n.next_idx;
                        _set_node( buff, n.prev_idx, pn );
                }
 
                if ( _last == idx ) {
                        _last = n.prev_idx;
                        _next = idx;
                } else {
                        auto nn     = _get_node( buff, n.next_idx );
                        nn.prev_idx = n.prev_idx;
                        _set_node( buff, n.next_idx, nn );
                }
 
                // If buffer is now empty, reset _next to npos as well
                if ( _first == npos && _last == npos )
                        _next = npos;
 
                _set_node( buff, idx, _node{ .next_idx = 0, .prev_idx = 0 } );
        }
 
public:
        _node _get_node( uint8_t* buff, index_type idx ) const noexcept
        {
                _node n;
                std::memcpy( &n, buff + idx, sizeof( _node ) );
                return n;
        }
 
        index_type _first = npos;
        index_type _last = npos;
        index_type _next = npos;
 
private:
        std::span< uint8_t > _buff;
};
 
template < typename T, typename IndexType, typename Base >
struct circular_buffer_allocator
{
        using value_type      = T;
        using size_type       = std::size_t;
        using difference_type = std::ptrdiff_t;
 
        // Required for C++17 allocator requirements
        template < typename U >
        struct rebind
        {
                using other = circular_buffer_allocator< U, IndexType, Base >;
        };
 
        explicit circular_buffer_allocator(
            circular_buffer_memory< IndexType, Base >& buffer ) noexcept
          : _buffer( &buffer )
        {
        }
 
        circular_buffer_allocator( circular_buffer_allocator const& ) noexcept = default;
 
        template < typename U >
        circular_buffer_allocator(
            circular_buffer_allocator< U, IndexType, Base > const& other ) noexcept
          : _buffer( other._buffer )
        {
        }
 
        T* allocate( std::size_t n )
        {
                if ( n == 0 )
                        return nullptr;
 
                std::size_t bytes = n * sizeof( T );
                void*       ptr   = _buffer->allocate( bytes, alignof( T ) );
 
                if ( !ptr )
                        throw std::bad_alloc();
 
                return static_cast< T* >( ptr );
        }
 
        void deallocate( T* ptr, std::size_t n )
        {
                if ( ptr )
                        _buffer->deallocate( ptr, n * sizeof( T ), alignof( T ) );
        }
 
        bool operator==( circular_buffer_allocator const& other ) const noexcept
        {
                return _buffer == other._buffer;
        }
 
        bool operator!=( circular_buffer_allocator const& other ) const noexcept
        {
                return !( *this == other );
        }
 
        template < typename U, typename IdxType, typename Base2 >
        friend struct circular_buffer_allocator;
 
private:
        circular_buffer_memory< IndexType, Base >* _buffer;
};
 
// ------------------------------------------------------------------------------
 
struct _get_env_t
{
        template < typename T >
        decltype( auto ) operator()( T const& o ) const noexcept
        {
                static constexpr bool v = requires( T const& o ) {
                        { o.get_env() };
                };
                if constexpr ( v )
                        return o.get_env();
                else
                        return empty_env{};
        }
};
inline constexpr _get_env_t get_env{};
 
template < typename T >
using env_type = decltype( get_env( std::declval< T const& >() ) );
 
struct _get_completion_signatures_t
{
        template < typename S, typename E >
        decltype( auto ) operator()( S& s, E& e ) const
        {
                static constexpr bool has_member = requires( S& s, E& e ) {
                        { s.get_completion_signatures( e ) };
                };
                if constexpr ( has_member )
                        return s.get_completion_signatures( e );
                else {
                        using sigs = typename S::completion_signatures;
                        return sigs{};
                }
        }
};
inline constexpr _get_completion_signatures_t get_completion_signatures{};
 
template < typename S, typename Env >
using _sender_completions_t =
    decltype( get_completion_signatures( std::declval< S& >(), std::declval< Env& >() ) );
 
template < typename T >
concept queryable = std::is_object_v< T >;
 
template < typename T >
concept sender = std::derived_from< typename std::remove_cvref_t< T >::sender_concept, sender_t > &&
                 requires( T const& s ) {
                         { get_env( s ) } -> queryable;
                 } && std::move_constructible< std::remove_cvref_t< T > > &&
                 std::constructible_from< std::remove_cvref_t< T >, T >;
 
template < typename T >
concept receiver =
    std::derived_from< typename std::remove_cvref_t< T >::receiver_concept, receiver_t > &&
    requires( T const& r ) {
            { get_env( r ) } -> queryable;
    } && std::move_constructible< std::remove_cvref_t< T > > &&
    std::constructible_from< std::remove_cvref_t< T >, T >;
 
template < typename S, typename R >
concept _connectable = sender< S > && receiver< R > && requires( S&& s, R&& r ) {
        { std::move( s ).connect( std::move( r ) ) };
};
 
struct _connect_t
{
        template < typename S, typename R >
                requires _connectable< S, R >
        auto operator()( S&& s, R&& r ) const
            noexcept( noexcept( std::move( s ).connect( std::move( r ) ) ) )
        {
                return std::move( s ).connect( std::move( r ) );
        }
};
inline constexpr _connect_t connect{};
 
template < typename R, typename T >
struct _receiver_sig_callable;
 
template < typename R, typename... Args >
struct _receiver_sig_callable< R, set_value_t( Args... ) >
{
        static constexpr bool value =
            requires( R& r, Args&&... args ) { r.set_value( (Args&&) args... ); };
};
 
template < typename R, typename Err >
struct _receiver_sig_callable< R, set_error_t( Err ) >
{
        static constexpr bool value = requires( R& r, Err&& err ) { r.set_error( (Err&&) err ); };
};
 
template < typename R >
struct _receiver_sig_callable< R, set_stopped_t() >
{
        static constexpr bool value = requires( R& r ) { r.set_stopped(); };
};
 
template < typename R, typename Sigs >
struct _receiver_for_impl;
 
template < typename R, typename... S >
struct _receiver_for_impl< R, completion_signatures< S... > >
{
        static constexpr bool value = ( ... && _receiver_sig_callable< R, S >::value );
};
 
template < typename R, typename S >
concept receiver_for =
    receiver< R > && _receiver_for_impl< R, _sender_completions_t< S, env_type< R > > >::value;
 
template < typename R, typename... Sig >
concept receiver_for_sigs = receiver< R > && ( ... && _receiver_sig_callable< R, Sig >::value );
 
template < signature S >
struct _sig_vtable_row;
 
template < typename... Args >
struct _sig_vtable_row< set_value_t( Args... ) >
{
 
        using f_t = void ( * )( void*, Args... );
        f_t _set_value;
 
        template < typename D, typename B >
        constexpr _sig_vtable_row( _tag< D >, _tag< B > ) noexcept
          : _set_value{ +[]( void* self, Args... args ) {
                  static_cast< D* >( static_cast< B* >( self ) )->set_value( (Args&&) args... );
          } }
        {
        }
 
        void invoke( set_value_t, void* self, Args... args ) const
        {
                _set_value( self, (Args&&) args... );
        }
};
 
template < typename... Args >
struct _sig_vtable_row< set_error_t( Args... ) >
{
        using f_t = void ( * )( void*, Args... );
        f_t _set_error;
 
        template < typename D, typename B >
        constexpr _sig_vtable_row( _tag< D >, _tag< B > ) noexcept
          : _set_error{ +[]( void* self, Args... args ) {
                  static_cast< D* >( static_cast< B* >( self ) )->set_error( (Args&&) args... );
          } }
        {
        }
 
        void invoke( set_error_t, void* self, Args... args ) const
        {
                _set_error( self, (Args&&) args... );
        }
};
 
template <>
struct _sig_vtable_row< set_stopped_t() >
{
        using f_t = void ( * )( void* );
        f_t _set_stopped;
 
        template < typename D, typename B >
        constexpr _sig_vtable_row( _tag< D >, _tag< B > ) noexcept
          : _set_stopped{ +[]( void* self ) {
                  static_cast< D* >( static_cast< B* >( self ) )->set_stopped();
          } }
        {
        }
 
        void invoke( set_stopped_t, void* self ) const
        {
                _set_stopped( self );
        }
};
 
template <>
struct _sig_vtable_row< _get_stopped_t() >
{
        using f_t = bool ( * )( void* );
        f_t _get_stopped;
 
        template < typename D, typename B >
        constexpr _sig_vtable_row( _tag< D >, _tag< B > ) noexcept
          : _get_stopped{ +[]( void* self ) {
                  return static_cast< D* >( static_cast< B* >( self ) )->get_stopped();
          } }
        {
        }
 
        bool invoke( _get_stopped_t, void* self ) const
        {
                return _get_stopped( self );
        }
};
 
template < signature... S >
struct _sig_vtable : _sig_vtable_row< S >...
{
        template < typename D, typename B >
        constexpr _sig_vtable( _tag< D > dt, _tag< B > bt ) noexcept
          : _sig_vtable_row< S >( dt, bt )...
        {
        }
 
        using _sig_vtable_row< S >::invoke...;
};
 
template < typename D, typename B, typename VTable >
static constexpr VTable _vtable_of = { _tag< D >{}, _tag< B >{} };
 
template < typename VTable, typename... Args >
concept _vtable_can_call_value = requires( VTable v, void* self, Args&&... args ) {
        ( v.invoke( set_value_t{}, self, (Args&&) args... ) );
};
 
template < typename VTable, typename Arg >
concept _vtable_can_call_error = requires( VTable v, void* self, Arg&& arg ) {
        ( v.invoke( set_error_t{}, self, (Arg&&) arg ) );
};
 
template < typename VTable >
concept _vtable_can_call_stopped =
    requires( VTable v, void* self ) { ( v.invoke( set_stopped_t{}, self ) ); };
 
template < signature... S >
struct _vtable_mixin
{
        using _vtable = _sig_vtable< S... >;
 
        template < typename D >
        _vtable_mixin( _tag< D > ) noexcept
          : vtable( _vtable_of< D, _vtable_mixin< S... >, _vtable > )
        {
                static_assert(
                    std::derived_from< D, _vtable_mixin< S... > >,
                    "Derived type must derive from _vtable_mixin with the same signature list" );
        }
 
        template < typename... Args >
        void _set_value( Args... args )
        {
                vtable.invoke( set_value_t{}, this, (Args&&) args... );
        }
 
        template < typename... Args >
        void _set_error( Args... args )
        {
                vtable.invoke( set_error_t{}, this, (Args&&) args... );
        }
 
        void _set_stopped()
        {
                vtable.invoke( set_stopped_t{}, this );
        }
 
        bool _get_stopped() noexcept
        {
                return vtable.invoke( _get_stopped_t{}, this );
        }
 
private:
        _vtable const& vtable;
};
 
 
 
template < typename TL, typename Tag, typename Sigs, template < class... > class Tuple >
struct _filter_map_tag;
 
template <
    template < class... >
    class Variant,
    typename... S,
    typename Tag,
    template < class... >
    class Tuple >
struct _filter_map_tag< Variant< S... >, Tag, completion_signatures<>, Tuple >
{
        using type = Variant< S... >;
};
 
template <
    template < class... >
    class Variant,
    typename... S,
    typename Tag,
    typename... Us,
    typename... Ts,
    template < class... >
    class Tuple >
struct _filter_map_tag< Variant< S... >, Tag, completion_signatures< Tag( Us... ), Ts... >, Tuple >
  : _filter_map_tag< Variant< S..., Tuple< Us... > >, Tag, completion_signatures< Ts... >, Tuple >
{
};
 
template <
    template < class... >
    class Variant,
    typename... S,
    typename Tag,
    typename T,
    typename... Ts,
    template < class... >
    class Tuple >
struct _filter_map_tag< Variant< S... >, Tag, completion_signatures< T, Ts... >, Tuple >
  : _filter_map_tag< Variant< S... >, Tag, completion_signatures< Ts... >, Tuple >
{
};
 
template < typename T >
using _type_identity_t = T;
 
template < typename Tag >
struct _sig_generator_t
{
        template < typename... Args >
        using type = Tag( Args... );
};
 
template < typename Sigs >
using _filter_values_as_variant_tuple_t =
    typename _filter_map_tag< std::variant<>, set_value_t, Sigs, std::tuple >::type;
 
template < typename Sigs >
using _filter_erros_as_variant_t =
    typename _filter_map_tag< std::variant<>, set_error_t, Sigs, _type_identity_t >::type;
 
template < typename Sigs >
using _filter_values_of_t = typename _filter_map_tag<
    completion_signatures<>,
    set_value_t,
    Sigs,
    _sig_generator_t< set_value_t >::type >::type;
 
template < typename Sigs >
using _filter_errors_of_t = typename _filter_map_tag<
    completion_signatures<>,
    set_error_t,
    Sigs,
    _sig_generator_t< set_error_t >::type >::type;
 
template < typename Sigs >
using _filter_stopped_of_t = typename _filter_map_tag<
    completion_signatures<>,
    set_stopped_t,
    Sigs,
    _sig_generator_t< set_stopped_t >::type >::type;
 
template < typename T, typename U >
struct _sigs_contains_type;
 
template < typename T, typename... Ts >
struct _sigs_contains_type< T, completion_signatures< Ts... > > : _contains_type< T, Ts... >
{
};
 
template < typename U >
concept _sigs_contains_set_stopped = _sigs_contains_type< set_stopped_t(), U >::value;
 
template < typename Sigs1, typename Sigs2 >
struct _sig_is_underset_of_impl;
 
template < typename... S1, typename... S2 >
struct _sig_is_underset_of_impl< completion_signatures< S1... >, completion_signatures< S2... > >
{
        static constexpr bool value = ( _contains_type< S1, S2... >::value && ... && true );
};
 
template < typename Sigs1, typename Sigs2 >
concept _sig_is_underset_of = _sig_is_underset_of_impl< Sigs1, Sigs2 >::value;
 
 
template < typename O, typename... Ts >
struct _sigs_merge_impl;
 
template < typename... S1, typename S, typename... S2, typename... Ts >
        requires( !_contains_type< S, S1... >::value )
struct _sigs_merge_impl< completion_signatures< S1... >, completion_signatures< S, S2... >, Ts... >
  : _sigs_merge_impl< completion_signatures< S1..., S >, completion_signatures< S2... >, Ts... >
{
};
 
template < typename... S1, typename S, typename... S2, typename... Ts >
        requires( _contains_type< S, S1... >::value )
struct _sigs_merge_impl< completion_signatures< S1... >, completion_signatures< S, S2... >, Ts... >
  : _sigs_merge_impl< completion_signatures< S1... >, completion_signatures< S2... >, Ts... >
{
};
 
template < typename... S1, typename... Ts >
struct _sigs_merge_impl< completion_signatures< S1... >, completion_signatures<>, Ts... >
  : _sigs_merge_impl< completion_signatures< S1... >, Ts... >
{
};
template < typename... S1 >
struct _sigs_merge_impl< completion_signatures< S1... > >
{
        using type = completion_signatures< S1... >;
};
 
template < typename... S >
using _sigs_merge_t = typename _sigs_merge_impl< completion_signatures<>, S... >::type;
 
template < typename S, typename... Ts >
struct _sigs_append_impl;
 
template < typename... S, typename... Ts >
struct _sigs_append_impl< completion_signatures< S... >, Ts... >
{
        using type = completion_signatures< S..., Ts... >;
};
 
template < typename Sigs, typename... Ts >
using _sigs_append_t = typename _sigs_append_impl< Sigs, Ts... >::type;
 
template < typename S, typename... Ts >
struct _sigs_concat_impl;
 
template < typename... S >
struct _sigs_concat_impl< completion_signatures< S... > >
{
        using type = completion_signatures< S... >;
};
 
template < typename... S1, typename... S2, typename... Ts >
struct _sigs_concat_impl< completion_signatures< S1... >, completion_signatures< S2... >, Ts... >
  : _sigs_concat_impl< completion_signatures< S1..., S2... >, Ts... >
{
};
 
template < typename... Sigs >
using _sigs_concat_t = typename _sigs_concat_impl< completion_signatures<>, Sigs... >::type;
 
 
template < class Token >
concept stoppable_token = requires( Token const tok ) {
        { tok.stop_requested() } noexcept -> std::same_as< bool >;
        { tok.stop_possible() } noexcept -> std::same_as< bool >;
        { Token( tok ) } noexcept;
} && std::copyable< Token > && std::equality_comparable< Token > && std::swappable< Token >;
 
template < class Token >
concept unstoppable_token =
    stoppable_token< Token > && requires { requires( !Token{}.stop_possible() ); };
 
template < class Source >
concept stoppable_source = requires( Source& src, Source const csrc ) {
        { csrc.get_token() } -> stoppable_token;
        { csrc.stop_possible() } noexcept -> std::same_as< bool >;
        { csrc.stop_requested() } noexcept -> std::same_as< bool >;
        { src.request_stop() } -> std::same_as< bool >;
};
 
struct inplace_stop_token;
 
template < typename CallbackFn >
struct inplace_stop_callback;
 
struct _inplace_stop_callback_base : zll::ll_base< _inplace_stop_callback_base >
{
        virtual void _execute() = 0;
 
        virtual ~_inplace_stop_callback_base() = default;
};
 
struct inplace_stop_source
{
        inplace_stop_source() noexcept = default;
 
        inplace_stop_source( inplace_stop_source const& )            = delete;
        inplace_stop_source& operator=( inplace_stop_source const& ) = delete;
 
        [[nodiscard]] inplace_stop_token get_token() const noexcept;
 
        [[nodiscard]] bool stop_possible() const noexcept
        {
                return true;
        }
 
        [[nodiscard]] bool stop_requested() const noexcept
        {
                return _stopped;
        }
 
        bool request_stop()
        {
                if ( _stopped )
                        return false;
                _stopped = true;
 
                while ( !_callbacks.empty() ) {
                        auto& cb = _callbacks.front();
                        cb._execute();
                        if ( !_callbacks.empty() && &_callbacks.front() == &cb )
                                zll::detach( cb );
                }
 
                return true;
        }
 
private:
        bool _stopped = false;
 
        // XXX: veve oficially hates that standard requires this to be mutable
        mutable zll::ll_list< _inplace_stop_callback_base > _callbacks;
 
        template < typename CallbackFn >
        friend struct inplace_stop_callback;
};
 
struct inplace_stop_token
{
        inplace_stop_token() noexcept = default;
 
        [[nodiscard]] bool stop_requested() const noexcept
        {
                return _source && _source->stop_requested();
        }
 
        [[nodiscard]] constexpr bool stop_possible() const noexcept
        {
                return _source != nullptr;
        }
 
        friend bool
        operator==( inplace_stop_token const& lhs, inplace_stop_token const& rhs ) noexcept
        {
                return lhs._source == rhs._source;
        }
 
        friend bool
        operator!=( inplace_stop_token const& lhs, inplace_stop_token const& rhs ) noexcept
        {
                return lhs._source != rhs._source;
        }
 
        template < typename CallbackFn >
        using callback_type = inplace_stop_callback< CallbackFn >;
 
private:
        inplace_stop_source const* _source = nullptr;
 
        inplace_stop_token( inplace_stop_source const* source ) noexcept
          : _source( source )
        {
        }
 
        friend struct inplace_stop_source;
 
        template < typename CallbackFn >
        friend struct inplace_stop_callback;
};
 
inline inplace_stop_token inplace_stop_source::get_token() const noexcept
{
        return inplace_stop_token{ this };
}
 
template < typename CallbackFn >
struct inplace_stop_callback : _inplace_stop_callback_base
{
        template < typename Initializer >
        explicit inplace_stop_callback( inplace_stop_token st, Initializer&& init ) noexcept(
            std::is_nothrow_constructible_v< CallbackFn, Initializer > &&
            std::is_nothrow_invocable_v< CallbackFn > )
          : _callback_fn( (Initializer&&) init )
          , _source( st._source )
        {
                if ( _source && _source->stop_possible() ) {
                        if ( _source->stop_requested() )
                                ( (CallbackFn&&) _callback_fn )();
                        else
                                _source->_callbacks.link_back( *this );
                }
        }
 
        inplace_stop_callback( inplace_stop_callback&& )                 = delete;
        inplace_stop_callback( inplace_stop_callback const& )            = delete;
        inplace_stop_callback& operator=( inplace_stop_callback&& )      = delete;
        inplace_stop_callback& operator=( inplace_stop_callback const& ) = delete;
 
        void _execute() override
        {
                ( (CallbackFn&&) _callback_fn )();
        }
 
private:
        CallbackFn                 _callback_fn;
        inplace_stop_source const* _source;
};
 
template < typename CallbackFn >
inplace_stop_callback( inplace_stop_token, CallbackFn ) -> inplace_stop_callback< CallbackFn >;
 
struct never_stop_token
{
        struct cb_type
        {
                explicit cb_type( never_stop_token, auto&& ) noexcept
                {
                }
        };
 
public:
        template < class >
        using callback_type = cb_type;
 
        static constexpr bool stop_requested() noexcept
        {
                return false;
        }
        static constexpr bool stop_possible() noexcept
        {
                return false;
        }
 
        bool operator==( never_stop_token const& ) const = default;
};
 
struct get_stop_token_t
{
        template < typename T >
        decltype( auto ) operator()( T const& env ) const noexcept
        {
                static constexpr bool v = requires( get_stop_token_t t ) {
                        { env.query( t ) };
                };  // namespace ecor
                if constexpr ( v )
                        return env.query( *this );
                else
                        return never_stop_token{};
        }
};
inline constexpr get_stop_token_t get_stop_token{};
 
template < typename Token = inplace_stop_token >
struct stop_token_env
{
        Token _token;
 
        [[nodiscard]] decltype( auto ) query( get_stop_token_t ) const noexcept
        {
                return _token;
        }
};
 
 
template < signature... S >
struct _ll_entry : _vtable_mixin< S... >, zll::ll_base< _ll_entry< S... > >
{
        template < typename D >
        _ll_entry( _tag< D > ) noexcept
          : _vtable_mixin< S... >( _tag< D >{} )
        {
        }
};
 
template < typename R, signature... S >
struct _ll_op : _ll_entry< S... >, R
{
        using operation_state_concept = operation_state_t;
 
        _ll_op( auto& list, R receiver ) noexcept( std::is_nothrow_move_constructible_v< R > )
          : _ll_entry< S... >( _tag< _ll_op >{} )
          , R( std::move( receiver ) )
          , _list( list )
        {
        }
 
        void start() noexcept
        {
                _list.link_back( *this );
        }
 
private:
        zll::ll_list< _ll_entry< S... > >& _list;
};
 
template < typename K, signature... S >
struct _sh_entry : _vtable_mixin< S... >, zll::sh_base< _sh_entry< K, S... > >
{
        K key;
 
        template < typename D >
        _sh_entry( K k, _tag< D > ) noexcept( std::is_nothrow_move_constructible_v< K > )
          : _vtable_mixin< S... >{ _tag< D >{} }
          , key( std::move( k ) )
        {
        }
 
        constexpr bool operator<( _sh_entry const& o ) const
            noexcept( noexcept( std::declval< K const& >() < std::declval< K const& >() ) )
        {
                return key < o.key;
        }
};
 
template < typename R, typename K, signature... S >
struct _sh_op : _sh_entry< K, S... >, R
{
        using operation_state_concept = operation_state_t;
 
        _sh_op( auto& heap, K key, R receiver ) noexcept(
            std::is_nothrow_move_constructible_v< R > && std::is_nothrow_move_constructible_v< K > )
          : _sh_entry< K, S... >( key, _tag< _sh_op >{} )
          , R( std::move( receiver ) )
          , _heap( heap )
        {
        }
 
        void start() noexcept
        {
                _heap.link( *this );
        }
 
private:
        zll::sh_heap< _sh_entry< K, S... > >& _heap;
};
 
template < signature... S >
struct _ll_sender
{
        using sender_concept = sender_t;
 
        _ll_sender( zll::ll_list< _ll_entry< S... > >& ll ) noexcept
          : _ll( ll )
        {
        }
 
        using completion_signatures = ecor::completion_signatures< S... >;
 
        template < receiver R >
        _ll_op< R, S... > connect( R receiver ) noexcept(
            noexcept( _ll_op< R, S... >{ _ll, std::move( receiver ) } ) )
        {
                static_assert(
                    receiver_for_sigs< R, S... >,
                    "Receiver does not satisfy the requirements for the sender's completion signatures" );
                return { _ll, std::move( receiver ) };
        }
 
private:
        zll::ll_list< _ll_entry< S... > >& _ll;
};
 
template < typename K, signature... S >
struct _sh_sender
{
        using sender_concept = sender_t;
 
        _sh_sender( K key, zll::sh_heap< _sh_entry< K, S... > >& sh ) noexcept(
            std::is_nothrow_move_constructible_v< K > )
          : _key( key )
          , _sh( sh )
        {
        }
 
        using completion_signatures = ecor::completion_signatures< S... >;
 
        template < receiver R >
        _sh_op< R, K, S... > connect( R receiver ) && noexcept(
            noexcept( _sh_op< R, K, S... >{ _sh, std::move( _key ), std::move( receiver ) } ) )
        {
                static_assert(
                    receiver_for_sigs< R, S... >,
                    "Receiver does not satisfy the requirements for the sender's completion signatures" );
                return { _sh, std::move( _key ), std::move( receiver ) };
        }
 
        template < receiver R >
        _sh_op< R, K, S... > connect( R receiver ) const& noexcept(
            noexcept( _sh_op< R, K, S... >{ _sh, _key, std::move( receiver ) } ) )
        {
                static_assert(
                    receiver_for_sigs< R, S... >,
                    "Receiver does not satisfy the requirements for the sender's completion signatures" );
                return { _sh, _key, std::move( receiver ) };
        }
 
private:
        K                                     _key;
        zll::sh_heap< _sh_entry< K, S... > >& _sh;
};
 
 
template < signature... S >
struct broadcast_source
{
        using sender_type = _ll_sender< S... >;
 
        _ll_sender< S... > schedule() noexcept
        {
                return ( _list );
        }
 
        template < typename... Args >
        void set_value( Args&&... args )
        {
                static_assert(
                    _vtable_can_call_value< _sig_vtable< S... >, Args... >,
                    "Completion signatures do not contain set_value_t(Args...)" );
                for_each( [&]( auto& n ) {
                        n._set_value( args... );
                } );
        }
 
        template < typename E >
        void set_error( E&& err )
        {
                static_assert(
                    _vtable_can_call_error< _sig_vtable< S... >, E >,
                    "Completion signatures do not contain set_error_t(E)" );
                for_each( [&]( auto& n ) {
                        n._set_error( err );
                } );
        }
 
        void set_stopped()
        {
                static_assert(
                    _vtable_can_call_stopped< _sig_vtable< S... > >,
                    "Completion signatures do not contain set_stopped_t()" );
                for_each( [&]( auto& n ) {
                        n._set_stopped();
                } );
        }
 
private:
        void for_each( auto&& f )
        {
                auto l = std::move( _list );
                if ( l.empty() )
                        return;
 
                auto& n   = l.front();
                using Acc = typename _ll_entry< S... >::access;
                for ( _ll_entry< S... >* m = &n; m; m = _node( Acc::get( *m ).next ) )
                        f( *m );
        }
 
        zll::ll_list< _ll_entry< S... > > _list;
};
 
 
template < signature... S >
struct fifo_source
{
        using completion_sigs = completion_signatures< S... >;
        using sender_type     = _ll_sender< S... >;
 
        _ll_sender< S... > schedule() noexcept
        {
                return ( _ll );
        }
 
        template < typename... V >
        void set_value( V&&... value )
        {
                static_assert(
                    _vtable_can_call_value< _sig_vtable< S... >, V... >,
                    "Completion signatures do not contain set_value_t(Args...)" );
                do_f( [&]( auto& n ) {
                        n._set_value( (V&&) value... );
                } );
        }
 
        template < typename E1 >
        void set_error( E1&& err )
        {
                static_assert(
                    _vtable_can_call_error< _sig_vtable< S... >, E1 >,
                    "Completion signatures do not contain set_error_t(E1)" );
                do_f( [&]( auto& n ) {
                        n._set_error( (E1&&) err );
                } );
        }
 
        void set_stopped()
        {
                static_assert(
                    _vtable_can_call_stopped< _sig_vtable< S... > >,
                    "Completion signatures do not contain set_stopped_t()" );
                do_f( [&]( auto& n ) {
                        n._set_stopped();
                } );
        }
 
private:
        void do_f( auto f )
        {
                if ( _ll.empty() )
                        return;
 
                auto& n = _ll.take_front();
                f( n );
        }
 
 
        zll::ll_list< _ll_entry< S... > > _ll;
};
 
template < typename K, signature... S >
struct seq_source
{
        using sender_type = _sh_sender< K, S... >;
 
        _sh_sender< K, S... >
        schedule( K key ) noexcept( noexcept( _sh_sender< K, S... >{ key, _sh } ) )
        {
                return { key, _sh };
        }
 
        template < typename... V >
        void set_value( V&&... value )
        {
                static_assert(
                    _vtable_can_call_value< _sig_vtable< S... >, V... >,
                    "Completion signatures do not contain set_value_t(V)" );
                do_f( [&]( auto& n ) {
                        n._set_value( (V&&) value... );
                } );
        }
 
        template < typename E1 >
        void set_error( E1&& err )
        {
                static_assert(
                    _vtable_can_call_error< _sig_vtable< S... >, E1 >,
                    "Completion signatures do not contain set_error_t(E1)" );
                // XXX: should this treat all entries or just the front one?
                do_f( [&]( auto& n ) {
                        n._set_error( (E1&&) err );
                } );
        }
 
        void set_stopped()
        {
                static_assert(
                    _vtable_can_call_stopped< _sig_vtable< S... > >,
                    "Completion signatures do not contain set_stopped_t()" );
                // XXX: should this treat all entries or just the front one?
                do_f( [&]( auto& n ) {
                        n._set_stopped();
                } );
        }
 
        [[nodiscard]] bool empty() const
        {
                return _sh.empty();
        }
 
        [[nodiscard]] _sh_entry< K, S... > const& front() const
        {
                return *_sh.top;
        }
 
private:
        void do_f( auto f )
        {
                if ( _sh.empty() )
                        return;
 
                auto& n = _sh.take();
                f( n );
        }
 
        zll::sh_heap< _sh_entry< K, S... > > _sh;
};
 
// ------------------------------------------------------------------------------
 
 
template < typename T >
struct _just_error
{
        using sender_concept = sender_t;
 
        T err;
 
        template < receiver R >
        struct _op
        {
                using operation_state_concept = operation_state_t;
 
                T err;
                R receiver;
 
                void start()
                {
                        receiver.set_error( std::move( err ) );
                }
        };
 
        template < receiver R >
        auto connect( R receiver ) && noexcept(
            std::is_nothrow_move_constructible_v< T > && std::is_nothrow_move_constructible_v< R > )
        {
                static_assert(
                    receiver_for< R, _just_error< T > >,
                    "Receiver does not satisfy the requirements for the sender's completion signatures" );
                return _op{ std::move( err ), std::move( receiver ) };
        }
 
        using completion_signatures = ecor::completion_signatures< set_error_t( T ) >;
};
 
template < typename T >
_just_error< T > just_error( T err )
{
        return _just_error< T >{ std::move( err ) };
}
 
// ------------------------------------------------------------------------------
 
enum class _awaitable_state_e : uint8_t
{
        empty,
        value
};
 
struct _schedulable : zll::ll_base< _schedulable >
{
        virtual void resume() = 0;
};
 
struct task_core
{
        bool run_once()
        {
                if ( _ready_tasks.empty() )
                        return false;
                auto& t = _ready_tasks.front();
                _ready_tasks.detach_front();
                t.resume();
                return true;
        }
 
        void run_n( std::size_t n )
        {
                for ( std::size_t i = 0; i < n; ++i )
                        if ( !run_once() )
                                break;
        }
 
        void reschedule( _schedulable& op ) noexcept
        {
                _ready_tasks.link_back( op );
        }
 
private:
        zll::ll_list< _schedulable > _ready_tasks;
};
 
template < typename T >
struct _awaitable_expected
{
        _awaitable_expected() noexcept
        {
        }
 
        _awaitable_state_e state = _awaitable_state_e::empty;
        union
        {
                T val;
        };
 
        template < typename... Args >
        void set_value( Args&&... args ) noexcept( std::is_nothrow_constructible_v< T, Args... > )
        {
                new ( (void*) &val ) T( (Args&&) args... );
                state = _awaitable_state_e::value;
        }
 
        ~_awaitable_expected() noexcept
        {
                if ( state == _awaitable_state_e::value )
                        val.~T();
        }
};
 
template <>
struct _awaitable_expected< void >
{
        _awaitable_expected() noexcept = default;
 
        _awaitable_state_e state = _awaitable_state_e::empty;
 
        void set_value() noexcept
        {
                state = _awaitable_state_e::value;
        }
};
 
template < typename T >
struct _awaitable_extract_type;
 
template < typename... U >
struct _awaitable_extract_type< std::variant< std::tuple< U... > > >
{
        using type = std::tuple< U... >;
};
 
template < typename U >
struct _awaitable_extract_type< std::variant< std::tuple< U > > >
{
        using type = U;
};
 
template <>
struct _awaitable_extract_type< std::variant< std::tuple<> > >
{
        using type = void;
};
 
template <>
struct _awaitable_extract_type< std::variant<> >
{
        using type = void;
};
 
template < typename PromiseType, typename S >
struct _task_awaitable
{
        _task_awaitable( S sender )
          : _exp()
          , _op( std::move( sender ).connect( _receiver{ ._self = this } ) )
        {
        }
 
        [[nodiscard]] bool await_ready() const noexcept
        {
                return false;
        }
 
        void await_suspend( std::coroutine_handle< PromiseType > ch ) noexcept
        {
                _promise = &ch.promise();
                _op.start();
        }
 
        decltype( auto ) await_resume() noexcept
        {
                ECOR_ASSERT( _exp.state == _awaitable_state_e::value );
                if constexpr ( std::same_as< value_type, void > )
                        return;
                else
                        return std::move( _exp.val );
        }
 
        using _env         = env_type< PromiseType >;
        using _completions = _sender_completions_t< std::remove_cvref_t< S >, _env >;
        using _values      = _filter_values_as_variant_tuple_t< _completions >;
        static_assert(
            std::variant_size_v< _values > <= 1,
            "Multiple set_value completions in awaitable not supported" );
 
        using value_type = typename _awaitable_extract_type< _values >::type;
 
        struct _receiver
        {
                using receiver_concept = receiver_t;
 
                _task_awaitable* _self;
 
                template < typename... Ts >
                void set_value( Ts&&... vals ) noexcept(
                    std::is_nothrow_constructible_v< value_type, Ts... > ||
                    std::same_as< value_type, void > )
                {
                        _self->_exp.set_value( (Ts&&) vals... );
                        _self->_promise->core.reschedule( *_self->_promise );
                }
 
                template < typename E >
                void set_error( E&& err )
                {
                        _self->_promise->invoke_set_error( (E&&) err );
                }
 
                void set_stopped()
                {
                        _self->_promise->invoke_set_stopped();
                }
 
                [[nodiscard]] decltype( auto ) get_env() const noexcept
                {
                        return _self->_promise->get_env();
                }
        };
 
 
private:
        PromiseType*                      _promise;
        _awaitable_expected< value_type > _exp;
 
        using _op_t = connect_type< S, _receiver >;
        _op_t _op;
};
 
template < class Alloc >
concept simple_allocator = requires( Alloc alloc, size_t n ) {
        { *alloc.allocate( n ) } -> std::same_as< typename Alloc::value_type& >;
        { alloc.deallocate( alloc.allocate( n ), n ) };
} && std::copy_constructible< Alloc > && std::equality_comparable< Alloc >;
 
 
struct task_memory_resource
{
        template < typename M >
        task_memory_resource( M& m ) noexcept
          : mem( &m )
        {
                alloc = +[]( void* mem, std::size_t const sz, std::size_t const align ) {
                        return ecor::allocate( *( (M*) mem ), sz, align );
                };
                dealloc = +[]( void* mem, void* p, std::size_t const sz, std::size_t const align ) {
                        ecor::deallocate( *( (M*) mem ), p, sz, align );
                };
        }
 
        [[nodiscard]] void* allocate( std::size_t const sz, std::size_t const align ) const
        {
                return alloc( mem, sz, align );
        }
 
        void deallocate( void* p, std::size_t const sz, std::size_t const align ) const
        {
                dealloc( mem, p, sz, align );
        }
 
        void* ( *alloc )( void*, std::size_t const, std::size_t const )         = nullptr;
        void ( *dealloc )( void*, void*, std::size_t const, std::size_t const ) = nullptr;
        void* mem                                                               = nullptr;
};
 
struct get_task_core_t
{
        template < typename T >
        decltype( auto ) operator()( T& t ) const noexcept
        {
                return t.query( get_task_core_t{} );
        }
};
inline constexpr get_task_core_t get_task_core{};
 
struct get_memory_resource_t
{
        template < typename T >
        decltype( auto ) operator()( T& t ) const noexcept
        {
                return t.query( get_memory_resource_t{} );
        }
};
inline constexpr get_memory_resource_t get_memory_resource{};
 
struct task_ctx
{
        task_memory_resource alloc;
        task_core            core;
 
        task_ctx( auto& mem )
          : alloc( mem )
        {
        }
 
        auto& query( get_task_core_t ) noexcept
        {
                return core;
        }
 
        auto& query( get_memory_resource_t ) noexcept
        {
                return alloc;
        }
};
 
template < typename T >
concept task_context = requires( T t ) {
        { get_task_core( t ) } -> std::same_as< task_core& >;
        { get_memory_resource( t ) } -> std::same_as< task_memory_resource& >;
};
 
struct task_default_cfg
{
        using extra_error_signatures = completion_signatures<>;
};
 
template < typename T >
concept task_config = requires() { typename T::extra_error_signatures; };
 
template < typename T, task_config CFG = task_default_cfg >
struct task;
 
enum class task_error : uint8_t
{
        none                     = 1,  
        task_unfinished          = 2,  
        task_allocation_failure  = 3,  
        task_already_started     = 4,  
        task_unhandled_exception = 5,  
        task_missing             = 6,  
};
 
struct _promise_base : _schedulable
{
        static constexpr std::size_t align   = alignof( std::max_align_t );
        static constexpr std::size_t spacing = align > sizeof( void* ) ? align : sizeof( void* );
 
        static void* _alloc( std::size_t sz, task_memory_resource& mem )
        {
                sz += spacing;
                void* const vp = allocate( mem, sz, align );
                if ( !vp )
                        return nullptr;
                auto* pmem = &mem;
                std::memcpy( vp, (void const*) &pmem, sizeof( void* ) );
                return ( (char*) vp ) + spacing;
        }
 
        static void _dealloc( void* const ptr, std::size_t const sz )
        {
                void*                 beg = ( (char*) ptr ) - spacing;
                task_memory_resource* mem = nullptr;
                std::memcpy( (void*) &mem, beg, sizeof( void* ) );
                deallocate( *mem, beg, sz + spacing, align );
        }
 
        void* operator new( std::size_t const sz, task_context auto&& ctx, auto&&... ) noexcept
        {
                task_memory_resource& a = get_memory_resource( ctx );
                return _alloc( sz, a );
        }
 
        void operator delete( void* const ptr, std::size_t const sz ) noexcept
        {
                _dealloc( ptr, sz );
        }
 
        _promise_base( _promise_base const& )            = delete;
        _promise_base& operator=( _promise_base const& ) = delete;
 
        _promise_base( task_context auto&& ctx, auto&&... )
          : core( get_task_core( ctx ) )
          , token()
        {
        }
 
        std::suspend_always initial_suspend() noexcept
        {
                return {};
        }
 
        task_core&         core;
        inplace_stop_token token;
 
        using _env = stop_token_env< inplace_stop_token& >;
 
        _env get_env() noexcept
        {
                return { token };
        }
};
 
template < typename Task, typename CFG >
struct _promise_type;
 
template < typename Task, typename CFG, typename Val = typename Task::value_type >
struct _promise_return_mixin
{
        void return_value( Val v )
        {
                static_cast< _promise_type< Task, CFG >* >( this )->invoke_set_value(
                    std::move( v ) );
        }
};
 
template < typename Task, typename CFG >
struct _promise_return_mixin< Task, CFG, void >
{
        void return_void()
        {
                static_cast< _promise_type< Task, CFG >* >( this )->invoke_set_value();
        }
};
 
template < typename Task, typename CFG >
struct _promise_type : _promise_base, _promise_return_mixin< Task, CFG, typename Task::value_type >
{
        using value_type = typename Task::value_type;
        using _promise_base::_promise_base;
        using vtable = _apply_to_sigs_t< _sig_vtable, typename Task::completion_signatures >;
 
        _promise_type( _promise_type const& )            = delete;
        _promise_type& operator=( _promise_type const& ) = delete;
 
        static Task get_return_object_on_allocation_failure()
        {
                return {
                    std::coroutine_handle< _promise_type >{}, task_error::task_allocation_failure };
        }
 
        Task get_return_object()
        {
                return { std::coroutine_handle< _promise_type >::from_promise( *this ) };
        }
 
        template < typename E >
                requires( _vtable_can_call_error< vtable, E > )
        auto yield_value( with_error< E > error )
        {
                return await_transform( just_error( std::move( error.error ) ) );
        }
 
        template < typename T >
        decltype( auto ) await_transform( T&& x ) noexcept
        {
                using U = std::remove_cvref_t< T >;
                if constexpr ( sender< U > ) {
                        using compls = _sender_completions_t< U, typename _promise_base::_env >;
                        using errs   = _filter_errors_of_t< compls >;
                        static_assert(
                            _sig_is_underset_of< errs, typename Task::_error_completions >,
                            "Error signatures are not compatible" );
                        using vals = _filter_values_as_variant_tuple_t< compls >;
                        static_assert(
                            std::variant_size_v< vals > <= 1,
                            "Sender used in co_await must only complete with a single set_value signature" );
                        return _task_awaitable< _promise_type, T >{ (T&&) x };
                } else {
                        return (T&&) x;
                }
        }
 
        void unhandled_exception() noexcept
        {
                this->invoke_set_error( task_error::task_unhandled_exception );
        }
 
        void resume() override
        {
                auto h = std::coroutine_handle< _promise_type >::from_promise( *this );
                h.resume();
        }
 
        template < typename U >
        void invoke_set_error( U&& err )
        {
                ECOR_ASSERT( this->_recv );
                if ( this->_recv ) {
                        auto* r = std::exchange( this->_recv, nullptr );
                        r->invoke( set_error_t{}, _obj, (U&&) err );
                }
        }
 
        template < typename... U >
        void invoke_set_value( U&&... v )
        {
                ECOR_ASSERT( this->_recv );
                if ( this->_recv ) {
                        auto* r = std::exchange( this->_recv, nullptr );
                        r->invoke( set_value_t{}, this->_obj, (U&&) v... );
                }
        }
 
        void invoke_set_stopped()
        {
                ECOR_ASSERT( this->_recv );
                if ( this->_recv ) {
                        auto* r = std::exchange( this->_recv, nullptr );
                        r->invoke( set_stopped_t{}, this->_obj );
                }
        }
 
        std::suspend_always final_suspend() noexcept
        {
                if ( _recv ) {
                        auto* r = std::exchange( this->_recv, nullptr );
                        r->invoke( set_error_t{}, _obj, task_error::task_unfinished );
                }
                return {};
        }
 
        template < typename U >
        void setup_continuable( U& p )
        {
                this->_recv = &_vtable_of< U, U, vtable >;
                this->_obj  = static_cast< void* >( &p );
        }
 
        ~_promise_type() noexcept
        {
                if ( _recv )
                        this->_recv->invoke( set_stopped_t{}, _obj );
        }
 
        vtable const* _recv = nullptr;
        void*         _obj  = nullptr;
};
 
template < typename T, task_config CFG, typename R >
struct _task_op
{
        using promise_type            = _promise_type< task< T, CFG >, CFG >;
        using operation_state_concept = operation_state_t;
 
        _task_op() noexcept = default;
        _task_op( task_error err, std::coroutine_handle< promise_type > h, R r ) noexcept(
            std::is_nothrow_move_constructible_v< R > )
          : _h( h )
          , _recv( std::move( r ) )
          , _error( err )
        {
        }
 
        _task_op( _task_op const& )            = delete;
        _task_op& operator=( _task_op const& ) = delete;
        _task_op( _task_op&& other ) noexcept
          : _h( std::exchange( other._h, nullptr ) )
          , _recv( std::move( other._recv ) )
          , _error( std::exchange( other._error, task_error::none ) )
        {
        }
 
        _task_op& operator=( _task_op&& other ) noexcept
        {
                if ( this == &other )
                        return *this;
                _task_op cpy{ std::move( other ) };
                std::swap( _h, cpy._h );
                std::swap( _recv, cpy._recv );
                std::swap( _error, cpy._error );
                return *this;
        }
 
        void start()
        {
                if ( _error != task_error::none ) {
                        _recv.set_error( _error );
                        return;
                }
                if ( !_h ) {
                        _recv.set_error( task_error::task_missing );
                        return;
                }
                if ( _h.done() ) {
                        _recv.set_error( task_error::task_already_started );
                        return;
                }
                if constexpr ( !std::same_as<
                                   decltype( ecor::get_stop_token( ecor::get_env( _recv ) ) ),
                                   never_stop_token > )
                        _h.promise().token = ecor::get_stop_token( ecor::get_env( _recv ) );
                _h.promise().setup_continuable( _recv );
                _h.promise().core.reschedule( _h.promise() );
        }
 
        void clear()
        {
                if ( _h )
                        _h.destroy();
                _h     = {};
                _recv  = {};
                _error = task_error::none;
        }
 
        operator bool()
        {
                return this->_h && !this->_h.done();
        }
 
        ~_task_op()
        {
                if ( this->_h )
                        this->_h.destroy();
        }
 
        std::coroutine_handle< promise_type > _h;
        R                                     _recv;
        task_error                            _error = task_error::none;
};
 
template < typename T, task_config CFG >
struct task
{
        using sender_concept = sender_t;
        using value_type     = T;
        using promise_type   = _promise_type< task, CFG >;
 
        using _error_completions =
            _sigs_append_t< typename CFG::extra_error_signatures, set_error_t( task_error ) >;
 
        using completion_signatures =
            _sigs_append_t< _error_completions, _value_setter_t< T >, set_stopped_t() >;
 
        static_assert(
            alignof( promise_type ) <= alignof( std::max_align_t ),
            "Unsupported alignment" );
 
        task(
            std::coroutine_handle< promise_type > handle,
            task_error                            error = task_error::none ) noexcept
          : _h( handle )
          , _error( error )
        {
        }
 
        task( task const& )            = delete;
        task& operator=( task const& ) = delete;
        task( task&& other ) noexcept
          : _h( std::exchange( other._h, nullptr ) )
          , _error( std::exchange( other._error, task_error::none ) )
        {
        }
 
        task& operator=( task&& other ) noexcept
        {
                if ( this != &other ) {
                        task tmp = std::move( other );
                        std::swap( _h, tmp._h );
                        std::swap( _error, tmp._error );
                }
                return *this;
        }
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    receiver_for< R, task< T, CFG > >,
                    "Receiver does not satisfy the requirements for the task's completion signatures" );
                auto tmp = _h;
                _h       = nullptr;
                return _task_op< T, CFG, R >{ _error, std::move( tmp ), std::move( receiver ) };
        }
 
        ~task()
        {
                if ( _h && !_h.done() )
                        _h.destroy();
        }
 
private:
        std::coroutine_handle< promise_type > _h;
        task_error                            _error = task_error::none;
};
 
 
template < typename S1, typename S2, typename R >
struct _or_op
{
        using operation_state_concept = operation_state_t;
 
        _or_op( S1 s1, S2 s2, R r )
          : _recv( std::move( r ) )
          , _op1( std::move( s1 ).connect( _r{ ._fired = _fired, ._recv = _recv } ) )
          , _op2( std::move( s2 ).connect( _r{ ._fired = _fired, ._recv = _recv } ) )
        {
        }
 
        // XXX: cancel of the other one shall be used instead
        void start()
        {
                _op1.start();
                _op2.start();
        }
 
        struct _r
        {
                using receiver_concept = receiver_t;
 
                bool& _fired;
                R&    _recv;
 
                template < typename... Args >
                void set_value( Args&&... v )
                {
                        if ( _fired )
                                return;
                        _fired = true;
                        _recv.set_value( (Args&&) v... );
                }
 
                template < typename T >
                void set_error( T&& e )
                {
                        if ( _fired )
                                return;
                        _fired = true;
                        _recv.set_error( (T&&) e );
                }
 
                // XXX: investigate whenever this should cancel the other operation
                void set_stopped()
                {
                        if ( _fired )
                                return;
                        _fired = true;
                        _recv.set_stopped();
                }
 
                decltype( auto ) get_env() noexcept
                {
                        return ecor::get_env( _recv );
                }
        };
 
        R    _recv;
        bool _fired = false;
 
        connect_type< S1, _r > _op1{};
        connect_type< S2, _r > _op2{};
};
 
template < typename S1, typename S2 >
struct _or_sender
{
        using sender_concept = sender_t;
 
        _or_sender( S1 s1, S2 s2 )
          : _s1( std::move( s1 ) )
          , _s2( std::move( s2 ) )
        {
        }
 
        template < typename Env >
        using _completions =
            _sigs_merge_t< _sender_completions_t< S1, Env >, _sender_completions_t< S2, Env > >;
 
        template < typename Env >
        _completions< Env > get_completion_signatures( Env&& ) noexcept
        {
                return {};
        }
 
 
        template < receiver R >
        _or_op< S1, S2, R > connect( R receiver )
        {
                static_assert(
                    receiver_for< R, _or_sender >,
                    "Receiver does not satisfy the requirements for the combined completion signatures of the two senders" );
                return { std::move( _s1 ), std::move( _s2 ), std::move( receiver ) };
        }
 
private:
        S1 _s1;
        S2 _s2;
};
 
template < typename S1, typename S2 >
_or_sender< S1, S2 > operator||( S1 s1, S2 s2 )
{
        return { std::move( s1 ), std::move( s2 ) };
}
 
 
template < sender S >
struct _as_variant
{
        using sender_concept = sender_t;
 
        template < typename Env >
        using _s_completions = _sender_completions_t< S, Env >;
 
        template < typename Env >
        using _values = typename _filter_map_tag<
            std::variant<>,
            set_value_t,
            _s_completions< Env >,
            _type_identity_t >::type;
 
        template < typename Env >
        using _error = _filter_errors_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _stopped = _filter_stopped_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _completions = _sigs_concat_t<
            completion_signatures< set_value_t( _values< Env > ) >,
            _error< Env >,
            _stopped< Env > >;
 
        _as_variant( S s )
          : _s( std::move( s ) )
        {
        }
 
        template < typename Env >
        _completions< Env > get_completion_signatures( Env&& ) noexcept
        {
                return {};
        }
 
 
        template < typename R >
        struct _recv : R
        {
                using receiver_concept = receiver_t;
 
                _recv( R r )
                  : R( std::move( r ) )
                {
                }
 
                using _env       = env_type< R >;
                using value_type = _values< _env >;
 
                template < typename T >
                void set_value( T&& val ) noexcept
                {
                        R::set_value( value_type{ (T&&) val } );
                }
        };
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    is_all_singular_v< _s_completions< env_type< R > > >,
                    "All set_value signatures of sender must be singular (exactly one argument) to use as_variant" );
                return std::move( _s ).connect( _recv< R >{ std::move( receiver ) } );
        }
 
private:
        S _s;
};
 
[[maybe_unused]] static inline struct as_variant_t
{
        template < sender S >
        auto operator()( S s ) const
        {
                return _as_variant< S >{ std::move( s ) };
        }
 
} as_variant;
 
auto operator|( auto s, as_variant_t )
{
        return as_variant( std::move( s ) );
}
 
 
template < sender S >
struct _err_to_val
{
        using sender_concept = sender_t;
 
        template < typename Env >
        using _s_completions = _sender_completions_t< S, Env >;
 
        template < typename Env >
        using _s_values = _filter_values_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _stopped = _filter_stopped_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _s_errors_as_val = typename _filter_map_tag<
            completion_signatures<>,
            set_error_t,
            _s_completions< Env >,
            _sig_generator_t< set_value_t >::type >::type;
 
        template < typename Env >
        using _completions =
            _sigs_concat_t< _s_values< Env >, _s_errors_as_val< Env >, _stopped< Env > >;
 
        template < typename Env >
        _completions< Env > get_completion_signatures( Env&& ) noexcept
        {
                return {};
        }
 
        _err_to_val( S s )
          : _s( std::move( s ) )
        {
        }
 
        template < typename R >
        struct _recv : R
        {
                using receiver_concept = receiver_t;
 
                _recv( R r )
                  : R( std::move( r ) )
                {
                }
 
                template < typename... Ts >
                void set_error( Ts&&... errs )
                {
                        R::set_value( (Ts&&) errs... );
                }
        };
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    receiver_for< R, _err_to_val >,
                    "Receiver does not satisfy the requirements for the err_to_val transformation" );
                return std::move( _s ).connect( _recv< R >{ std::move( receiver ) } );
        }
 
private:
        S _s;
};
 
[[maybe_unused]] static inline struct err_to_val_t
{
        template < sender S >
        auto operator()( S s ) const
        {
                return _err_to_val< S >{ std::move( s ) };
        }
 
} err_to_val;
 
auto operator|( auto s, err_to_val_t )
{
        return err_to_val( std::move( s ) );
}
 
 
template < sender S >
struct _sink_err
{
        using sender_concept = sender_t;
 
        template < typename Env >
        using _s_completions = _sender_completions_t< S, Env >;
 
        template < typename Env >
        using _s_errors = typename _filter_map_tag<
            std::variant<>,
            set_error_t,
            _s_completions< Env >,
            _type_identity_t >::type;
 
        template < typename Env >
        using _s_errors_as_val = std::optional< _s_errors< Env > >;
 
        template < typename Env >
        using _s_values = _filter_values_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _s_stopped = _filter_stopped_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _completions = _sigs_merge_t<
            completion_signatures< set_value_t( _s_errors_as_val< Env > ) >,
            _s_stopped< Env > >;
 
        template < typename Env >
        _completions< Env > get_completion_signatures( Env&& ) noexcept
        {
                return {};
        }
 
        template < typename R >
        struct _recv : R
        {
                using receiver_concept = receiver_t;
 
                _recv( R r )
                  : R( std::move( r ) )
                {
                }
 
                using _env    = env_type< R >;
                using value_t = _s_errors_as_val< _env >;
 
                void set_value() noexcept
                {
                        R::set_value( value_t{} );
                }
 
                template < typename T >
                void set_error( T&& err ) noexcept
                {
                        R::set_value( value_t{ (T&&) err } );
                }
        };
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                using _env = env_type< R >;
                static_assert(
                    std::variant_size_v< _s_errors< _env > > > 0,
                    "Sender used with sink_err must have at least one error type" );
                static_assert(
                    std::same_as< _s_values< _env >, completion_signatures<> > ||
                        std::same_as< _s_values< _env >, completion_signatures< set_value_t() > >,
                    "Sender used with sink_err must not complete with set_value, or there has to be only one set_value of shape set_value_t()" );
                static_assert(
                    receiver_for< R, _sink_err< S > >,
                    "Receiver does not satisfy the requirements for the sink_err sender's completion signatures" );
                return std::move( _s ).connect( _recv< R >{ std::move( receiver ) } );
        }
 
        S _s;
};
 
[[maybe_unused]] static inline struct sink_err_t
{
 
        auto operator()( sender auto s ) const noexcept
        {
                return _sink_err{ std::move( s ) };
        }
 
} sink_err;
 
auto operator|( sender auto s, sink_err_t ) noexcept
{
        return _sink_err{ std::move( s ) };
}
 
 
template < typename Ret >
struct _then_value_sig_ret
{
        using type = set_value_t( Ret );
};
 
template <>
struct _then_value_sig_ret< void >
{
        using type = set_value_t();
};
 
template < typename F >
struct _then_value_sig
{
        template < typename... Args >
        using type = typename _then_value_sig_ret< std::invoke_result_t< F, Args... > >::type;
};
 
template < sender S, typename F >
struct _then
{
        using sender_concept = sender_t;
 
        template < typename Env >
        using _s_completions = _sender_completions_t< S, Env >;
 
        template < typename Env >
        using _s_errors = _filter_errors_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _s_stopped = _filter_stopped_of_t< _s_completions< Env > >;
 
        template < typename Env >
        using _s_mapped_values = typename _filter_map_tag<
            completion_signatures<>,
            set_value_t,
            _s_completions< Env >,
            _then_value_sig< F >::template type >::type;
 
        template < typename Env >
        using _completions =
            _sigs_merge_t< _s_mapped_values< Env >, _s_errors< Env >, _s_stopped< Env > >;
 
        template < typename Env >
        _completions< Env > get_completion_signatures( Env&& ) noexcept
        {
                return {};
        }
 
        template < typename R >
        struct _recv : R
        {
                using receiver_concept = receiver_t;
 
                F _f;
 
                _recv( R r, F f )
                  : R( std::move( r ) )
                  , _f( std::move( f ) )
                {
                }
 
                template < typename... Args >
                void set_value( Args&&... args )
                {
                        if constexpr ( std::is_void_v< std::invoke_result_t< F, Args... > > ) {
                                std::move( _f )( (Args&&) args... );
                                R::set_value();
                        } else {
                                R::set_value( std::move( _f )( (Args&&) args... ) );
                        }
                }
        };
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    receiver_for< R, _then >,
                    "Receiver does not satisfy the requirements for the then sender's completion signatures" );
                return std::move( _s ).connect(
                    _recv< R >{ std::move( receiver ), std::move( _f ) } );
        }
 
        S _s;
        F _f;
};
 
template < typename F >
struct _then_closure
{
        F _f;
};
 
template < sender S, typename F >
auto operator|( S s, _then_closure< F > c )
{
        return _then< S, F >{ std::move( s ), std::move( c._f ) };
}
 
[[maybe_unused]] static inline struct then_t
{
        template < typename F >
        auto operator()( F f ) const
        {
                return _then_closure< F >{ std::move( f ) };
        }
 
        template < sender S, typename F >
        auto operator()( S s, F f ) const
        {
                return _then< S, F >{ std::move( s ), std::move( f ) };
        }
 
} then;
 
 
template < task_config CFG = task_default_cfg >
struct _task_holder_base : _schedulable
{
        using _task_t = task< void, CFG >;
 
        _task_holder_base( _task_holder_base const& )            = delete;
        _task_holder_base& operator=( _task_holder_base const& ) = delete;
        _task_holder_base( _task_holder_base&& )                 = delete;
        _task_holder_base& operator=( _task_holder_base&& )      = delete;
 
        explicit _task_holder_base( task_core& core )
          : _core( core )
        {
        }
 
        void start()
        {
                _core.reschedule( *this );
        }
 
        _ll_sender< set_value_t() > stop()
        {
                _stop_src.request_stop();
                return _done_src.schedule();
        }
 
protected:
        virtual _task_t _make_task() = 0;
 
private:
        struct _recv
        {
                using receiver_concept = receiver_t;
 
                _task_holder_base* _holder;
 
                template < typename... Args >
                void set_value( Args&&... ) noexcept
                {
                        _holder->_core.reschedule( *_holder );
                }
 
                template < typename E >
                void set_error( E&& ) noexcept
                {
                        _holder->_core.reschedule( *_holder );
                }
 
                void set_stopped() noexcept
                {
                        _holder->_core.reschedule( *_holder );
                }
 
                [[nodiscard]] stop_token_env< inplace_stop_token > get_env() const noexcept
                {
                        return { _holder->_stop_src.get_token() };
                }
        };
 
        using _op_t = connect_type< _task_t, _recv >;
 
        void resume() override
        {
                if ( _op.has_value() ) {
                        _op->clear();  // explicitly free the done coroutine frame
                        _op.reset();
                }
                if ( _stop_src.stop_requested() ) {
                        _done_src.set_value();
                } else {
                        _op.emplace( _make_task().connect( _recv{ this } ) );
                        _op->start();
                }
        }
 
        task_core&                        _core;
        inplace_stop_source               _stop_src;
        broadcast_source< set_value_t() > _done_src;
        std::optional< _op_t >            _op;
};
 
template <
    task_config  CFG = task_default_cfg,
    task_context Ctx = task_ctx,
    typename Factory = void >
struct task_holder : _task_holder_base< CFG >
{
        using base = _task_holder_base< CFG >;
 
        task_holder( Ctx& ctx, Factory factory )
          : _task_holder_base< CFG >( get_task_core( ctx ) )
          , _ctx( ctx )
          , _factory( std::move( factory ) )
        {
        }
 
        using base::start;
        using base::stop;
 
private:
        task< void, CFG > _make_task() override
        {
                return _factory( _ctx );
        }
 
        Ctx&    _ctx;
        Factory _factory;
};
 
template < task_context Ctx, typename Factory >
task_holder( Ctx&, Factory ) -> task_holder< task_default_cfg, Ctx, Factory >;
 
 
struct _wait_until_stopped
{
        using sender_concept = sender_t;
 
        using completion_signatures = ecor::completion_signatures< set_value_t() >;
 
        template < typename R >
        struct _op
        {
                using operation_state_concept = operation_state_t;
 
                _op( R r )
                  : _r( std::move( r ) )
                {
                }
 
                void start()
                {
                        auto st = get_stop_token( ecor::get_env( _r ) );
                        static_assert(
                            !std::same_as< decltype( st ), never_stop_token >,
                            "Receiver must be stoppable" );
                        _cb.emplace( st, cb_t{ .op = this } );
                }
 
                R _r;
 
                struct cb_t
                {
                        void operator()() noexcept
                        {
                                op->_r.set_value();
                        }
                        _op* op;
                };
 
                std::optional< inplace_stop_callback< cb_t > > _cb;
        };
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    receiver_for< R, _wait_until_stopped >,
                    "Receiver does not satisfy the requirements for wait_until_stopped's completion signatures" );
                return _op{ std::move( receiver ) };
        }
};
 
[[maybe_unused]]
static inline _wait_until_stopped wait_until_stopped;
 
 
struct _suspend_awaiter
{
        [[nodiscard]] bool await_ready() const noexcept
        {
                return false;
        }
 
        template < typename Promise >
        void await_suspend( std::coroutine_handle< Promise > h ) noexcept
        {
                auto& p = h.promise();
                if ( p.token.stop_requested() ) {
                        p.invoke_set_stopped();
                        return;
                }
                p.core.reschedule( p );
        }
 
        void await_resume() noexcept
        {
        }
};
 
[[maybe_unused]]
static inline _suspend_awaiter suspend;
 
 
template < typename T >
concept _has_member_async_destroy = requires( T& obj ) {
        { obj.async_destroy() } -> sender;
};
 
template < typename Ctx, typename T >
concept _has_adl_async_destroy = requires( Ctx& ctx, T& obj ) {
        { async_destroy( ctx, obj ) } -> sender;
};
 
struct _async_destroy_t
{
        template < typename Ctx, typename T >
        ecor::sender auto operator()( Ctx& ctx, T& obj ) const
        {
                if constexpr ( _has_adl_async_destroy< Ctx, T > )
                        return async_destroy( ctx, obj );
                else
                        return obj.async_destroy();
        }
};
inline constexpr _async_destroy_t async_destroy{};
 
enum class _cb_state : uint8_t
{
        alive,
        queued,
        destroying,
};
 
struct _async_arena_cb_base : zll::ll_base< _async_arena_cb_base >
{
        virtual void start_destroy()  = 0;
        virtual void finish_cleanup() = 0;
 
protected:
        ~_async_arena_cb_base() = default;
};
 
struct _async_arena_core_base : _schedulable
{
        explicit _async_arena_core_base( task_core& core ) noexcept
          : _core( core )
        {
        }
 
        void _enqueue_destroy( _async_arena_cb_base& cb ) noexcept
        {
                _destroy_queue.link_back( cb );
                if ( !_active && zll::detached( static_cast< _schedulable& >( *this ) ) )
                        _core.reschedule( *this );
        }
 
        void _on_destroy_complete() noexcept
        {
                ECOR_ASSERT( _active );
                ECOR_ASSERT( zll::detached( static_cast< _schedulable& >( *this ) ) );
                _core.reschedule( *this );
        }
 
        void resume() override
        {
                if ( _active ) {
                        _active->finish_cleanup();
                        _active = nullptr;
                }
 
                if ( !_destroy_queue.empty() ) {
                        _active = &_destroy_queue.take_front();
                        _active->start_destroy();
                } else if ( _stopped && _alive_list.empty() ) {
                        _done_src.set_value();
                }
        }
 
        task_core&                           _core;
        zll::ll_list< _async_arena_cb_base > _alive_list;
        zll::ll_list< _async_arena_cb_base > _destroy_queue;
        _async_arena_cb_base*                _active  = nullptr;
        bool                                 _stopped = false;
        broadcast_source< set_value_t() >    _done_src;
 
        struct _destroy_receiver
        {
                using receiver_concept = receiver_t;
 
                _async_arena_core_base* _core;
 
                void set_value( auto&&... ) noexcept
                {
                        _core->_on_destroy_complete();
                }
 
                void set_error( auto&& ) noexcept
                {
                        _core->_on_destroy_complete();
                }
 
                void set_stopped() noexcept
                {
                        _core->_on_destroy_complete();
                }
        };
};
 
struct _async_arena_cb_counted : _async_arena_cb_base
{
        explicit _async_arena_cb_counted( _async_arena_core_base& arena ) noexcept
          : _arena( arena )
        {
        }
 
        void add_ref() noexcept
        {
                ECOR_ASSERT( _refs > 0 );
                ECOR_ASSERT( _state == _cb_state::alive );
                ++_refs;
        }
 
        void release() noexcept
        {
                ECOR_ASSERT( _refs > 0 );
                if ( --_refs == 0 ) {
                        _state = _cb_state::queued;
                        _arena._enqueue_destroy( *this );
                }
        }
 
        _async_arena_core_base& _arena;
        uint32_t                _refs  = 1;
        _cb_state               _state = _cb_state::alive;
 
protected:
        ~_async_arena_cb_counted() = default;
};
 
template < typename Ctx, typename Mem >
struct async_arena;
 
template < typename Ctx, typename Mem >
struct _async_arena_core : _async_arena_core_base
{
        _async_arena_core( Ctx& ctx, Mem& mem ) noexcept
          : _async_arena_core_base( get_task_core( ctx ) )
          , _ctx( ctx )
          , _mem( mem )
        {
        }
 
        Ctx& _ctx;
        Mem& _mem;
};
 
template < typename T, typename Ctx, typename Mem >
struct _async_arena_control_block final : _async_arena_cb_counted
{
        using _destroy_sender_t =
            decltype( ecor::async_destroy( std::declval< Ctx& >(), std::declval< T& >() ) );
 
        using _destroy_receiver = _async_arena_core_base::_destroy_receiver;
        using _destroy_op_t     = connect_type< _destroy_sender_t, _destroy_receiver >;
 
        _destroy_op_t* _destroy_op = nullptr;
        T              _object;
 
        template < typename... Args >
        _async_arena_control_block( _async_arena_core< Ctx, Mem >& arena, Args&&... a )
          : _async_arena_cb_counted( arena )
          , _object( (Args&&) a... )
        {
        }
 
        void start_destroy() override
        {
                ECOR_ASSERT( _state == _cb_state::queued );
                _state = _cb_state::destroying;
 
                auto& arena = static_cast< _async_arena_core< Ctx, Mem >& >( _arena );
 
                using op_t = _destroy_op_t;
 
                auto  s = ecor::async_destroy( arena._ctx, _object );
                void* p = ecor::allocate( arena._mem, sizeof( op_t ), alignof( op_t ) );
                ECOR_ASSERT( p );
                _destroy_op =
                    ::new ( p ) op_t( std::move( s ).connect( _destroy_receiver{ &_arena } ) );
                _destroy_op->start();
        }
 
        void finish_cleanup() override
        {
                auto& arena = static_cast< _async_arena_core< Ctx, Mem >& >( _arena );
 
                using op_t = _destroy_op_t;
                _destroy_op->~op_t();
                ecor::deallocate( arena._mem, _destroy_op, sizeof( op_t ), alignof( op_t ) );
                _destroy_op = nullptr;
                auto& mem   = arena._mem;
                this->~_async_arena_control_block();
                ecor::deallocate(
                    mem,
                    this,
                    sizeof( _async_arena_control_block ),
                    alignof( _async_arena_control_block ) );
        }
};
 
template < typename T, typename Ctx, typename Mem >
struct async_ptr
{
        using _cb_t = _async_arena_control_block< T, Ctx, Mem >;
 
        async_ptr() noexcept = default;
 
        async_ptr( std::nullptr_t ) noexcept
        {
        }
 
        async_ptr( async_ptr const& o ) noexcept
          : _cb( o._cb )
        {
                if ( _cb )
                        _cb->add_ref();
        }
 
        async_ptr( async_ptr&& o ) noexcept
          : _cb( o._cb )
        {
                o._cb = nullptr;
        }
 
        async_ptr& operator=( async_ptr const& o ) noexcept
        {
                if ( this != &o ) {
                        _reset();
                        _cb = o._cb;
                        if ( _cb )
                                _cb->add_ref();
                }
                return *this;
        }
 
        async_ptr& operator=( async_ptr&& o ) noexcept
        {
                if ( this != &o ) {
                        _reset();
                        _cb   = o._cb;
                        o._cb = nullptr;
                }
                return *this;
        }
 
        ~async_ptr()
        {
                _reset();
        }
 
        explicit operator bool() const noexcept
        {
                return _cb != nullptr;
        }
 
        T& operator*() const noexcept
        {
                ECOR_ASSERT( _cb );
                return _cb->_object;
        }
 
        T* operator->() const noexcept
        {
                ECOR_ASSERT( _cb );
                return &_cb->_object;
        }
 
        T* get() const noexcept
        {
                return _cb ? &_cb->_object : nullptr;
        }
 
        void reset() noexcept
        {
                _reset();
        }
 
        friend bool operator==( async_ptr const& a, async_ptr const& b ) noexcept
        {
                return a._cb == b._cb;
        }
 
        friend bool operator!=( async_ptr const& a, async_ptr const& b ) noexcept
        {
                return a._cb != b._cb;
        }
 
        friend bool operator==( async_ptr const& a, std::nullptr_t ) noexcept
        {
                return a._cb == nullptr;
        }
 
        friend bool operator!=( async_ptr const& a, std::nullptr_t ) noexcept
        {
                return a._cb != nullptr;
        }
 
private:
        template < typename, typename >
        friend struct async_arena;
 
        explicit async_ptr( _cb_t* cb ) noexcept
          : _cb( cb )
        {
        }
 
        void _reset() noexcept
        {
                if ( _cb ) {
                        _cb->release();
                        _cb = nullptr;
                }
        }
 
        _cb_t* _cb = nullptr;
};
 
template < typename Ctx, typename Mem >
struct async_arena
{
        async_arena( Ctx& ctx, Mem& mem ) noexcept
          : _core( ctx, mem )
        {
        }
 
        async_arena( async_arena const& )            = delete;
        async_arena& operator=( async_arena const& ) = delete;
        async_arena( async_arena&& )                 = delete;
        async_arena& operator=( async_arena&& )      = delete;
 
        template < typename T, typename... Args >
        async_ptr< T, Ctx, Mem > make( Args&&... args )
        {
                ECOR_ASSERT( !_core._stopped );
                using cb_t = _async_arena_control_block< T, Ctx, Mem >;
                void* p    = ecor::allocate( _core._mem, sizeof( cb_t ), alignof( cb_t ) );
                if ( !p )
                        return { nullptr };
                auto* cb = ::new ( p ) cb_t( _core, (Args&&) args... );
                _core._alive_list.link_back( *cb );
                return async_ptr< T, Ctx, Mem >{ cb };
        }
 
        _ll_sender< set_value_t() > async_destroy() noexcept
        {
                _core._stopped = true;
                if ( _core._alive_list.empty() && _core._destroy_queue.empty() &&
                     zll::detached( static_cast< _schedulable& >( _core ) ) && !_core._active )
                        _core._core.reschedule( _core );
                return _core._done_src.schedule();
        }
 
private:
        _async_arena_core< Ctx, Mem > _core;
};
 
 
template < typename T >
struct _trnx_vtable_mixin;
 
template < signature... S >
        requires( !_contains_type< set_stopped_t(), S... >::value )
struct _trnx_vtable_mixin< completion_signatures< S... > >
{
        using type = _vtable_mixin< S... >;
};
 
template < signature... S >
        requires( _contains_type< set_stopped_t(), S... >::value )
struct _trnx_vtable_mixin< completion_signatures< S... > >
{
        using type = _vtable_mixin< S..., _get_stopped_t() >;
};
 
template < typename T >
using _trnx_vtable_mixin_t =
    typename _trnx_vtable_mixin< _sender_completions_t< T, empty_env > >::type;
 
template < typename T >
struct trnx_entry : _trnx_vtable_mixin_t< T >, zll::ll_base< trnx_entry< T > >
{
        using base    = _trnx_vtable_mixin_t< T >;
        using _vtable = typename base::_vtable;
        static constexpr bool _has_stop_sig =
            _sigs_contains_set_stopped< _sender_completions_t< T, empty_env > >;
 
        T data;
 
        template < typename Derived >
        trnx_entry( _tag< Derived >, T data )
          : base( _tag< Derived >{} )
          , data( std::move( data ) )
        {
        }
 
        template < typename... Args >
        void set_value( Args&&... args )
        {
                this->_set_value( (Args&&) args... );
        }
 
        template < typename E >
        void set_error( E&& e )
        {
                this->_set_error( (E&&) e );
        }
 
        void set_stopped()
        {
                this->_set_stopped();
        }
 
        bool get_stopped() const noexcept
        {
                return this->get_stopped();
        }
};
template < typename T, typename R >
struct _trnx_controller_op : trnx_entry< T >, R
{
        using _vtable                       = typename trnx_entry< T >::_vtable;
        static constexpr bool _has_stop_sig = trnx_entry< T >::_has_stop_sig;
 
        using R::set_error;
        using R::set_stopped;
        using R::set_value;
 
        _trnx_controller_op( T data, R r, zll::ll_list< trnx_entry< T > >& pending )
          : trnx_entry< T >{ _tag< _trnx_controller_op >{}, std::move( data ) }
          , R( std::move( r ) )
          , _pending( pending )
        {
        }
 
        void start()
        {
                _pending.link_front( *this );
        }
 
        [[nodiscard]] bool get_stopped() const noexcept
        {
                if constexpr ( _has_stop_sig ) {
                        bool res = get_stop_token( get_env( (R&) *this ) ).stop_requested();
                        return res;
                } else
                        return false;
        }
 
private:
        zll::ll_list< trnx_entry< T > >& _pending;
};
 
template < typename T >
struct _trnx_controller_sender
{
        using sender_concept = sender_t;
 
        template < typename Env >
        constexpr auto get_completion_signatures( Env&& e ) noexcept
        {
                return ecor::get_completion_signatures( val, (Env&&) e );
        }
 
        template < receiver R >
        auto connect( R receiver ) &&
        {
                static_assert(
                    receiver_for< R, _trnx_controller_sender >,
                    "Receiver does not satisfy the requirements for the transaction sender's completion signatures" );
 
                return _trnx_controller_op< T, R >{
                    std::move( val ), std::move( receiver ), _pending };
        }
 
        T                                val;
        zll::ll_list< trnx_entry< T > >& _pending;
};
 
template < typename T, size_t N >
struct trnx_circular_buffer
{
        static_assert( std::has_single_bit( N ), "Size of circular buffer must be a power of 2" );
        static_assert(
            N < std::numeric_limits< uint16_t >::max(),
            "Size of circular buffer must be less than 65536" );
        static_assert( std::is_trivial_v< T > );
 
        bool full() const noexcept
        {
                return ( enqueue - deliver ) == N;
        }
 
        void push( T val ) noexcept
        {
                _data[enqueue++ % N] = val;
        }
 
        bool empty() const noexcept
        {
                return deliver == enqueue;
        }
 
        bool has_tx() const noexcept
        {
                return tx != enqueue;
        }
 
        T& tx_front() noexcept
        {
                return _data[tx % N];
        }
 
        void tx_done() noexcept
        {
                tx++;
        }
 
        bool has_rx() const noexcept
        {
                return rx != tx;
        }
 
        T& rx_front() noexcept
        {
                return _data[rx % N];
        }
 
        void rx_done() noexcept
        {
                rx++;
        }
 
        bool has_deliver() const noexcept
        {
                return deliver != rx;
        }
 
        T& deliver_front() noexcept
        {
                return _data[deliver % N];
        }
 
        void pop() noexcept
        {
                deliver++;
        }
 
        std::atomic< uint16_t > deliver = 0;
        std::atomic< uint16_t > rx      = 0;
        std::atomic< uint16_t > tx      = 0;
        std::atomic< uint16_t > enqueue = 0;
 
private:
        T _data[N];
};
 
template < typename T >
struct trnx_controller_source
{
        using sender_type                   = _trnx_controller_sender< T >;
        static constexpr bool _has_stop_sig = trnx_entry< T >::_has_stop_sig;
 
        sender_type schedule( T val )
        {
                return { std::move( val ), _pending_tx };
        }
 
        trnx_entry< T >* query_next_trnx()
        {
                if constexpr ( _has_stop_sig ) {
                        _pending_tx.remove_if( []( trnx_entry< T >& tx ) {
                                if ( !tx._get_stopped() )
                                        return false;
 
                                tx._set_stopped();
                                return true;
                        } );
                }
                if ( _pending_tx.empty() )
                        return nullptr;
                return &_pending_tx.take_back();
        }
 
private:
        zll::ll_list< trnx_entry< T > > _pending_tx;
};
 
}  // namespace ecor