Channel.hpp

#pragma once
 
#include "util/async/Concepts.hpp"
 
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/experimental/channel.hpp>
#include <boost/asio/experimental/concurrent_channel.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/system/detail/error_code.hpp>
 
#include <concepts>
#include <cstddef>
#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
 
namespace util {
 
#ifdef __clang__
namespace detail {
// Forward declaration for compile-time check
template <typename T>
struct ChannelInstantiated;
}  // namespace detail
#endif

enum class ProducerType {
    Single, 
    Multi   
};

enum class ConsumerType {
    Single, 
    Multi 
};

template <typename T, ProducerType P = ProducerType::Multi, ConsumerType C = ConsumerType::Multi>
class Channel {
    static constexpr bool kIS_MULTI_PRODUCER = (P == ProducerType::Multi);
    static constexpr bool kIS_MULTI_CONSUMER = (C == ConsumerType::Multi);
 
private:
    class ControlBlock {
        using InternalChannelType =
            boost::asio::experimental::concurrent_channel<void(boost::system::error_code, T)>;
        boost::asio::any_io_executor executor_;
        InternalChannelType ch_;
 
    public:
        template <typename ContextType>
            requires(not async::SomeExecutionContext<ContextType>)
        ControlBlock(ContextType&& context, std::size_t capacity)
            : executor_(context.get_executor()), ch_(context, capacity)
        {
        }
 
        template <async::SomeExecutionContext ContextType>
        ControlBlock(ContextType&& context, std::size_t capacity)
            : executor_(context.getExecutor().get_executor()), ch_(context.getExecutor(), capacity)
        {
        }
 
        [[nodiscard]] InternalChannelType&
        channel()
        {
            return ch_;
        }
 
        void
        close()
        {
            if (not isClosed()) {
                ch_.close();
                // Workaround for Boost bug: close() alone doesn't cancel pending async operations.
                // We must call cancel() to unblock them. The bug also causes cancel() to return
                // error_code 0 instead of channel_cancelled, so async operations must check
                // isClosed() to detect this case.
                // https://github.com/chriskohlhoff/asio/issues/1575
                ch_.cancel();
            }
        }
 
        [[nodiscard]] bool
        isClosed() const
        {
            return not ch_.is_open();
        }
    };

    struct Guard {
        std::shared_ptr<ControlBlock> shared;
 
        ~Guard()
        {
            shared->close();
        }
    };
 
public:
    class Sender {
        std::shared_ptr<ControlBlock> shared_;
        std::conditional_t<kIS_MULTI_PRODUCER, std::shared_ptr<Guard>, Guard> guard_;
 
        friend class Channel<T, P, C>;

        Sender(std::shared_ptr<ControlBlock> shared)
            : shared_(shared), guard_([shared = std::move(shared)]() {
                if constexpr (kIS_MULTI_PRODUCER) {
                    return std::make_shared<Guard>(std::move(shared));
                } else {
                    return Guard{std::move(shared)};
                }
            }())
        {
        }
 
    public:
        Sender(Sender&&) = default;
        Sender(Sender const&)
            requires kIS_MULTI_PRODUCER
        = default;
        Sender(Sender const&)
            requires(!kIS_MULTI_PRODUCER)
        = delete;
 
        Sender&
        operator=(Sender&&) = default;
        Sender&
        operator=(Sender const&)
            requires kIS_MULTI_PRODUCER
        = default;
        Sender&
        operator=(Sender const&)
            requires(!kIS_MULTI_PRODUCER)
        = delete;

        template <typename D>
        bool
        asyncSend(D&& data, boost::asio::yield_context yield)
            requires(std::convertible_to<std::remove_cvref_t<D>, std::remove_cvref_t<T>>)
        {
            boost::system::error_code const ecIn;
            boost::system::error_code ecOut;
            shared_->channel().async_send(ecIn, std::forward<D>(data), yield[ecOut]);
 
            // Workaround: asio channels bug returns ec=0 on cancel, check isClosed() instead
            if (not ecOut and shared_->isClosed())
                return false;
 
            return not ecOut;
        }

        template <typename D>
        void
        asyncSend(D&& data, std::invocable<bool> auto&& fn)
            requires(std::convertible_to<std::remove_cvref_t<D>, std::remove_cvref_t<T>>)
        {
            boost::system::error_code const ecIn;
            shared_->channel().async_send(
                ecIn,
                std::forward<D>(data),
                [fn = std::forward<decltype(fn)>(fn),
                 shared = shared_](boost::system::error_code ec) mutable {
                    // Workaround: asio channels bug returns ec=0 on cancel, check isClosed()
                    // instead
                    if (not ec and shared->isClosed()) {
                        fn(false);
                        return;
                    }
 
                    fn(not ec);
                }
            );
        }

        template <typename D>
        bool
        trySend(D&& data)
            requires(std::convertible_to<std::remove_cvref_t<D>, std::remove_cvref_t<T>>)
        {
            boost::system::error_code ec;
            return shared_->channel().try_send(ec, std::forward<D>(data));
        }
    };

    class Receiver {
        std::shared_ptr<ControlBlock> shared_;
        std::conditional_t<kIS_MULTI_CONSUMER, std::shared_ptr<Guard>, Guard> guard_;
 
        friend class Channel<T, P, C>;

        Receiver(std::shared_ptr<ControlBlock> shared)
            : shared_(shared), guard_([shared = std::move(shared)]() {
                if constexpr (kIS_MULTI_CONSUMER) {
                    return std::make_shared<Guard>(std::move(shared));
                } else {
                    return Guard{std::move(shared)};
                }
            }())
        {
        }
 
    public:
        Receiver(Receiver&&) = default;
        Receiver(Receiver const&)
            requires kIS_MULTI_CONSUMER
        = default;
        Receiver(Receiver const&)
            requires(!kIS_MULTI_CONSUMER)
        = delete;
 
        Receiver&
        operator=(Receiver&&) = default;
        Receiver&
        operator=(Receiver const&)
            requires kIS_MULTI_CONSUMER
        = default;
        Receiver&
        operator=(Receiver const&)
            requires(!kIS_MULTI_CONSUMER)
        = delete;

        std::optional<T>
        tryReceive()
        {
            std::optional<T> result;
            shared_->channel().try_receive([&result](boost::system::error_code ec, auto&& value) {
                if (not ec)
                    result = std::forward<decltype(value)>(value);
            });
 
            return result;
        }

        [[nodiscard]] std::optional<T>
        asyncReceive(boost::asio::yield_context yield)
        {
            boost::system::error_code ec;
            auto value = shared_->channel().async_receive(yield[ec]);
 
            if (ec)
                return std::nullopt;
 
            return value;
        }

        void
        asyncReceive(std::invocable<std::optional<std::remove_cvref_t<T>>> auto&& fn)
        {
            shared_->channel().async_receive([fn = std::forward<decltype(fn)>(fn)](
                                                 boost::system::error_code ec, T&& value
                                             ) mutable {
                if (ec) {
                    fn(std::optional<T>(std::nullopt));
                    return;
                }
 
                fn(std::make_optional<T>(std::move(value)));
            });
        }

        [[nodiscard]] bool
        isClosed() const
        {
            return shared_->isClosed();
        }
    };

    static std::pair<Sender, Receiver>
    create(auto&& context, std::size_t capacity)
    {
#ifdef __clang__
        static_assert(
            util::detail::ChannelInstantiated<T>::value,
            "When using Channel<T> with Clang, you must add INSTANTIATE_CHANNEL_FOR_CLANG(T) "
            "to one .cpp file. See documentation at the bottom of Channel.hpp for details."
        );
#endif
        auto shared =
            std::make_shared<ControlBlock>(std::forward<decltype(context)>(context), capacity);
        auto sender = Sender{shared};
        auto receiver = Receiver{std::move(shared)};
 
        return {std::move(sender), std::move(receiver)};
    }
};
 
}  // namespace util
 
// ================================================================================================
// Clang/Apple Clang Workaround for Boost.Asio Experimental Channels
// ================================================================================================
//
// IMPORTANT: When using Channel<T> with Clang or Apple Clang, you MUST add the following line
// to ONE .cpp file that uses Channel<T>:
//
//     INSTANTIATE_CHANNEL_FOR_CLANG(YourType)
//
// Example:
//     // In ChannelTests.cpp or any .cpp file that uses Channel<int>:
//     #include "util/Channel.hpp"
//     INSTANTIATE_CHANNEL_FOR_CLANG(int)
//
// Why this is needed:
// Boost.Asio's experimental concurrent_channel has a bug where close() doesn't properly cancel
// pending async operations. When using cancellation signals (which we do in our workaround),
// Clang generates vtable references for internal cancellation_handler types but Boost.Asio
// doesn't provide the definitions, causing linker errors:
//
//   Undefined symbols for architecture arm64:
//     "boost::asio::detail::cancellation_handler<...>::call(boost::asio::cancellation_type)"
//     "boost::asio::detail::cancellation_handler<...>::destroy()"
//
// This macro explicitly instantiates the required template specializations.
//
// See: https://github.com/chriskohlhoff/asio/issues/1575
//
#ifdef __clang__
 
#include <boost/asio/cancellation_signal.hpp>
#include <boost/asio/experimental/channel_traits.hpp>
#include <boost/asio/experimental/detail/channel_service.hpp>
 
namespace util::detail {
// Tag type used to verify that INSTANTIATE_CHANNEL_FOR_CLANG was called for a given type
template <typename T>
struct ChannelInstantiated : std::false_type {};
}  // namespace util::detail
 
#define INSTANTIATE_CHANNEL_FOR_CLANG(T)                                                       \
    /* NOLINTNEXTLINE(cppcoreguidelines-virtual-class-destructor) */                           \
    template class boost::asio::detail::cancellation_handler<                                  \
        boost::asio::experimental::detail::channel_service<boost::asio::detail::posix_mutex>:: \
            op_cancellation<                                                                   \
                boost::asio::experimental::channel_traits<>,                                   \
                void(boost::system::error_code, T)>>;                                          \
    namespace util::detail {                                                                   \
    template <>                                                                                \
    struct ChannelInstantiated<T> : std::true_type {};                                         \
    }
 
#else
 
// No workaround needed for non-Clang compilers
#define INSTANTIATE_CHANNEL_FOR_CLANG(T)
 
#endif