Bold plan for a major release

[gummif]

Here are a few ideas for changes to be made for (potentially breaking) cppzmq v5.

1. Require C++11 standard support, but still support gcc 4.7 and later with partial C++11 support. This possible future requirement has been in the read me for 2 years now. Large fraction of the library now requires C++11.
2. Require libzmq 4.0+. There are no CI tests with older versions and there are no releases on github for version prior to 4.0.6. (should we maybe require 4.1 or 4.2?)
3. Remove deprecated functions that would not cause silent breakage (would cause compile time errors). Keep functions that would cause silent changes to runtime behavior if removed (e.g. greedy function overloads).

These changes are conservative in the sense that if you are using C++11 or later, libzmq v4 or later and not using any deprecated functionality, then the upgrade from the latest v4 to v5 should not require any changes to you code. If you do use deprecated functions then you might get a compile time error in the worst case.

[bluca]

Please do not break backward compatibility. This header is shipped in distributions. It would cause major, major headaches and work to downstreams.

[gummif]

After some deep though I can see that part 3 and maybe part 2 could cause some pain with system packages (system packages are generally a bad idea anyway for development). I don't know how the process is for adding new package versions to distributions, but I guess it is their responsibility to not add package versions that would break the system? But if it is possible people will eventually managed to break things.

About requiring C++11, the read me says "Some features already require a compiler supporting C++11. In the future, probably all features will require C++11". Is that something that you prefer never happens?

And regarding part2, I believe libzmq v3.x is rarely used. And it would be very unlikely that someone would use libzmq v3.x together with latest cppzmq, so in practice that would requirement should have no effect (in that event an option would be to simply update libzmq or not update cppzmq). Correct me if you can come up with a situation where that would be the case.

[bluca]

I don't know how the process is for adding new package versions to distributions, but I guess it is their responsibility to not add package versions that would break the system?

Yes, but it means there will never be a new version uploaded, which is not good.

The problem is not C++11 so much for distros - they all support it, it's just the ancient and deprecated stuff that doesn't. Not even zmq 3.x. The problem is breaking API compatibility by removing functions.

[gummif]

OK. So would you see the following would be acceptable?

1. Require C++11 standard support, but still support gcc 4.7 and later with partial C++11 support.
2. Require libzmq 4.0+.
3. Make bug fix releases for v4 if required for C++03 compatible code.

[bluca]

That would be fine, yes

[planetmarshall]

Here are a few ideas for changes to be made for (potentially breaking) cppzmq v5.

1. Require C++11 standard support, but still support gcc 4.7 and later with partial C++11 support. This possible future requirement has been in the read me for 2 years now. Large fraction of the library now requires C++11.

Skip C++11, C++14 and C++17, move to C++20 and provide an asynchronous API based on coroutines, mirroring asyncio support in the PyZMQ library.

[gummif]

I think coroutine support could be added as an addon for those that have compilers that support it. I have been thinking about it but have not yet thought of a good solution (adding an async API on top of a blocking API is not ideal).

[SidneyCogdill]

I've wrote some helper functions for Boost.Cobalt corral to asynchronously send or receive messages (works on Linux for me, probably on BSD / macOS too):

old implementation using cobalt

This old implementation has many problems:

• It repeatedly creates new coroutine stack frames in a loop and even Clang (the most capable one at doing HALO) fails optimizing them.
• I didn't really understand the multipart message until later. It's a queue mechanism rather than streaming mechanism. The implementation failed at considering that and therefore performs very inefficiently.
• By using cobalt::this_coro::executor it has an implicit dependency on asio as the primary asynchronous runtime even though it's not obvious from the signature (I think cobalt allows switching runtimes, but I'm not sure). the newer implementation on the other hand only requires the existence of asio event loop, so you can for example hook it inside Qt without much troubles.

namespace cobalt = boost::cobalt;
namespace asio = boost::asio;
namespace scope = boost::scope;

///
/// \brief The potentially slower path of [async_recv] when it
/// isn't possible to immediately retrieve a message.
///
auto async_recv_suspend(zmq::socket_t &socket,
                        zmq::message_t &message,
                        zmq::recv_flags flags = zmq::recv_flags::none)
    -> cobalt::task<zmq::recv_result_t>
{
    asio::posix::stream_descriptor sd{co_await cobalt::this_coro::executor,
                                      socket.get(zmq::sockopt::fd)};
    /// Since stream_descriptor takes the ownership of the file descriptor,
    /// it will attempt to close the descriptor on scope exit.
    /// This defer guard prevents that from happening,
    /// by releasing the fd right before exiting the scope.
    scope::defer_guard release_fd = [&] { sd.release(); };

    while (true) {
        auto events = socket.get(zmq::sockopt::events);
        if (events & ZMQ_POLLIN)
            co_return socket.recv(message, zmq::recv_flags::dontwait | flags);
        co_await sd.async_wait(asio::posix::stream_descriptor::wait_read);
    }
}

///
/// \brief Asynchronously receive a message from [zmq::socket].
/// \param [socket] The socket to retrieve message from.
/// \param [message] The message object to be filled with received message.
/// \param [flags] Accepts a [zmq::recv_flags] variable.
/// \return [zmq::recv_result_t] representing the operation result.
/// \note This function attempts to directly retrieve a message without any suspension,
/// and falls-back to asio stream descriptor monitoring if that isn't possible.
///
auto async_recv(zmq::socket_t &socket,
                zmq::message_t &message,
                zmq::recv_flags flags = zmq::recv_flags::none) -> cobalt::task<zmq::recv_result_t>
{
    auto events = socket.get(zmq::sockopt::events);
    if (events & ZMQ_POLLIN)
        co_return socket.recv(message, zmq::recv_flags::dontwait | flags);
    co_return co_await async_recv_suspend(socket, message, flags);
}

///
/// \brief Asynchronously receive a multipart message from [zmq::socket].
/// \param [socket] The socket to retrieve message from.
/// \param [multipart] The multipart message object to be cleared and filled with received messages.
/// \param [flags] Accepts a [zmq::recv_flags] variable.
/// \return `true` on success, `false` if the operation failed completely or partially.
/// \note On partial failure, the [multipart] is left with message parts that have
/// been successfully received.
///
auto async_recv(zmq::socket_t &socket,
                zmq::multipart_t &multipart,
                zmq::recv_flags flags = zmq::recv_flags::none) -> cobalt::task<bool>
{
    multipart.clear();
    bool more = true;
    /// The control flow is directly copied from the member function `zmq::multipart_t::recv`,
    /// but with the recv replaced by the async version.
    while (more) {
        zmq::message_t message;
        if (!co_await async_recv(socket, message, flags))
            co_return false;
        more = message.more();
        multipart.add(std::move(message));
    }
    co_return true;
}

///
/// \brief The potentially slower path of [async_send] when it isn't
/// possible to immediately send due to buffer saturation.
///
auto async_send_suspend(zmq::socket_t &socket,
                        zmq::message_t &message,
                        zmq::send_flags flags = zmq::send_flags::none)
    -> cobalt::task<zmq::send_result_t>
{
    asio::posix::stream_descriptor sd{co_await cobalt::this_coro::executor,
                                      socket.get(zmq::sockopt::fd)};
    /// Since stream_descriptor takes the ownership of the file descriptor,
    /// it will attempt to close the descriptor on scope exit.
    /// This defer guard prevents that from happening,
    /// by releasing the fd right before exiting the scope.
    scope::defer_guard release_fd = [&] { sd.release(); };

    while (true) {
        auto events = socket.get(zmq::sockopt::events);
        if (events & ZMQ_POLLOUT) {
            co_return socket.send(message, zmq::send_flags::dontwait | flags);
        }
        co_await sd.async_wait(asio::posix::stream_descriptor::wait_write);
    }
}

///
/// \brief Asynchronously send a message to a [zmq::socket].
/// \param [socket] The socket.
/// \param [message] The message to be sent.
/// \param [flags] Accepts a [zmq::send_flags] variable.
/// \return [zmq::send_result_t] representing the operation result.
/// \note This function attempts to directly send the message without any suspension,
/// and falls-back to asio stream descriptor monitoring if that isn't possible.
///
auto async_send(zmq::socket_t &socket,
                zmq::message_t &message,
                zmq::send_flags flags = zmq::send_flags::none) -> cobalt::task<zmq::send_result_t>
{
    auto events = socket.get(zmq::sockopt::events);
    if (events & ZMQ_POLLOUT) {
        co_return socket.send(message, zmq::send_flags::dontwait | flags);
    }
    co_return co_await async_send_suspend(socket, message, flags);
}

///
/// \brief Asynchronously send a multipart message to a [zmq::socket].
/// \param [socket] The socket.
/// \param [multipart] The multipart message object to be sent.
/// Message parts that are sent will be cleared from [multipart].
/// \param [flags] Accepts a [zmq::send_flags] variable.
/// \return `true` on success, `false` if the operation failed completely or partially.
///
auto async_send(zmq::socket_t &socket,
                zmq::multipart_t &multipart,
                zmq::send_flags flags = zmq::send_flags::none) -> cobalt::task<bool>
{
    flags = flags & ~(zmq::send_flags::sndmore);
    /// The control flow is directly copied from the member function `zmq::multipart_t::send`,
    /// but with the send replaced by the async version.
    while (bool more = !multipart.empty()) {
        zmq::message_t message = multipart.pop();
        more = !multipart.empty();
        if (!co_await async_send(socket,
                                 message,
                                 (more ? zmq::send_flags::sndmore : zmq::send_flags::none) | flags))
            co_return false;
    }
    multipart.clear();
    co_return true;
}

new implementation using corral and asio

WaitFileDescriptor.hpp

#pragma once
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <coroutine>

/// \brief An awaitable struct for suspending until a file descriptor is ready to read or write.
/// \details This struct implements the awaitable interface required by Corral's await syntax.
/// It uses Boost.Asio to asynchronously wait for the specified events on a file descriptor.
/// The await_cancel() function is a Corral extension that allows cancellation of the operation.
struct WaitFileDescriptor
{
public /* ctors */:
    struct Param;

    /// \brief Constructs a WaitFileDescriptor object for the given file descriptor.
    /// \note Refer to Param struct for parameters.
    inline WaitFileDescriptor(Param params)
        : sd(params.executor, params.handle)
        , wait_type(params.wait_type)
        , resume_filter(std::move(params.resume_filter))
    {}

    ///
    /// \brief Releases the file descriptor without closing it.
    /// stream_descriptor attempts to close the file descriptor on destruction,
    /// this prevents that from happening because the file descriptor should be
    /// managed by ZeroMQ instead.
    ///
    inline ~WaitFileDescriptor() { sd.release(); }

    struct Param
    {
        /// \brief The Boost.Asio executor to watch for the file descriptor events.
        boost::asio::any_io_executor executor;

        /// \brief The native file descriptor handle to monitor.
        boost::asio::posix::stream_descriptor::native_handle_type handle;

        /// \brief The type of events to wait for (e.g., read, write).
        boost::asio::posix::stream_descriptor::wait_type wait_type;

        /// \brief A move-only function to customize when a coroutine resumes after an event.
        /// \details This callback allows users to filter or modify the resumption of a suspended coroutine.
        /// By default, it immediately resumes the coroutine. Users can provide a custom lambda or function
        /// object to add conditional logic (e.g., ignoring certain events, logging, or delaying resumption
        /// by rescheduling).
        ///
        /// \note It's required to be `noexcept` since it's called from the event loop and
        /// there isn't a meaningful way to handle exceptions other than aborting the program.
        ///
        /// \example
        ///
        /// ```cpp
        ///
        /// // Example custom filter that resumes only if a flag is set
        /// resume_filter = [](std::coroutine_handle<> h) noexcept {
        ///     if (should_resume()) {
        ///         h.resume();
        ///     }
        /// };
        ///
        /// ```
        ///
        /// \param h - The handle to the suspended coroutine. The function can use it
        /// to resume the coroutine, or delay resumption by rescheduling with the handle.
        ///
        std::move_only_function<void(std::coroutine_handle<>) noexcept> resume_filter =
            [](std::coroutine_handle<> h) noexcept { h.resume(); };
    };

public /* awaitable */:
    inline bool await_ready() const noexcept { return false; }

    inline void await_resume() noexcept {}

    inline void await_suspend(std::coroutine_handle<> h) noexcept
    {
        sd.async_wait(wait_type, [this, h](boost::system::error_code err) {
            if (!err && !cancelled)
                this->resume_filter(h);
        });
    }

public /* corral extensions */:
    /// \brief Cancels the asynchronous wait operation (Corral extension).
    /// \return Always returns true to indicate cancellation was successful.
    /// \note refer to https://github.com/hudson-trading/corral/blob/master/doc/02_adapting.md#custom-awaitables
    inline std::true_type await_cancel(std::coroutine_handle<>) noexcept
    {
        sd.cancel();
        cancelled = true;
        return {};
    }

    [[nodiscard]] inline bool await_must_resume() const noexcept
    {
        if (!cancelled) {
            return false;
        }
        return true;
    }

private /* field */:
    boost::asio::posix::stream_descriptor sd;
    boost::asio::posix::stream_descriptor::wait_type wait_type;
    std::move_only_function<void(std::coroutine_handle<>) noexcept> resume_filter;

    /// Flag indicating whether the operation was cancelled.
    bool cancelled = false;
};

AsyncReceive.hpp

#pragma once

#include "WaitFileDescriptor.hpp"
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <coroutine>
#include <exception>
#include <zmq.hpp>

struct RecvError : public std::exception
{
public:
    inline auto what() const noexcept -> const char * override
    {
        return "Failed receiving message";
    }
};

///
/// \brief Asynchronously receive message from a ZeroMQ socket.
/// \return [zmq::message_t] - The received message.
/// \throw [RecvError] - On receive error.
///
struct [[nodiscard]] AsyncReceive
{
public /* ctor */:
    inline AsyncReceive(boost::asio::any_io_executor executor,
                        zmq::socket_ref socket,
                        zmq::recv_flags flags = zmq::recv_flags::none)
        : m_wait({
              .executor = executor,
              .handle = socket.get(zmq::sockopt::fd),
              .wait_type = boost::asio::posix::stream_descriptor::wait_read,
              // only allow resuming if there are poll-in events
              .resume_filter =
                  [this](std::coroutine_handle<> h) noexcept {
                      if (has_pollin_events()) {
                          h.resume();
                      } else {
                          /// If the file descriptor can be read but socket doesn't
                          /// have poll-in event, schedule for another awake.
                          m_wait.await_suspend(h);
                      }
                  },
          })
        , m_socket(socket)
        , m_flags(flags)
    {}

public /* awaitable */:
    ///
    /// \brief An optimization which avoids suspending the coroutine if it isn't necessary.
    /// \return true - when there are already poll-in events and `await_resume` will be
    /// immediately invoked.
    ///
    inline bool await_ready() const noexcept { return has_pollin_events(); }

    [[nodiscard]] inline auto await_resume() -> zmq::message_t
    {
        auto result = m_socket.recv(m_msg, zmq::recv_flags::dontwait | m_flags);
        if (!result) [[unlikely]]
            throw RecvError{};
        return std::move(m_msg);
    }

    /// Delegates the suspension/cancellation to the WaitFileDescriptor awaitable.
    inline void await_suspend(std::coroutine_handle<> h) noexcept { m_wait.await_suspend(h); }

public /* corral extensions */:
    inline bool await_cancel(std::coroutine_handle<> h) noexcept { return m_wait.await_cancel(h); }

    [[nodiscard]] inline bool await_must_resume() const noexcept
    {
        return m_wait.await_must_resume();
    }

private:
    inline bool has_pollin_events() const noexcept
    {
        std::puts("has_pollin_events");
        return m_socket.get(zmq::sockopt::events) & ZMQ_POLLIN;
    }

private:
    WaitFileDescriptor m_wait;
    zmq::socket_ref m_socket;
    zmq::recv_flags m_flags;
    zmq::message_t m_msg;
};

AsyncReceiveMultipart.hpp

#pragma once

#include "AsyncReceive.hpp"
#include <zmq_addon.hpp>

///
/// \brief Asynchronously receive message from a ZeroMQ socket.
/// \return [zmq::multipart_t] - The received message.
/// \throw [RecvError] - On receive error.
///
struct [[nodiscard]] AsyncReceiveMultipart
{
public /* ctor */:
    inline AsyncReceiveMultipart(boost::asio::any_io_executor executor,
                                 zmq::socket_ref socket,
                                 zmq::recv_flags flags = zmq::recv_flags::none)
        : m_wait({
              .executor = executor,
              .handle = socket.get(zmq::sockopt::fd),
              .wait_type = boost::asio::posix::stream_descriptor::wait_read,
              // only allow resuming if there are poll-in events
              .resume_filter =
                  [this](std::coroutine_handle<> h) noexcept {
                      if (this->has_pollin_events()) {
                          h.resume();
                      } else {
                          /// If the file descriptor can be read but socket doesn't
                          /// have poll-in event, schedule for another awake.
                          m_wait.await_suspend(h);
                      }
                  },
          })
        , m_socket(socket)
        , m_flags(flags)
    {}

public /* awaitable */:
    ///
    /// \brief An optimization which avoids suspending the coroutine if it isn't necessary.
    /// \return true - when there are already poll-in events and `await_resume` will be
    /// immediately invoked.
    ///
    inline bool await_ready() const noexcept { return has_pollin_events(); }

    ///
    /// \note ZeroMQ's multipart message is not equivalent to streams in other libraries.
    /// A multipart message is atomic: either it's fully received or it's not.
    /// It's like sending a std::deque in a `zmq::message_t`.
    /// Therefore it's not necessary to repeatedly suspend the coroutine in the receive loop:
    /// When poll-in event comes in, the whole multipart message has already been received by
    /// ZeroMQ.
    ///
    [[nodiscard]] inline auto await_resume() -> zmq::multipart_t
    {
        m_msg.clear();
        while (true) {
            zmq::message_t m;
            auto result = m_socket.recv(m, zmq::recv_flags::dontwait | m_flags);
            if (!result) [[unlikely]]
                throw RecvError{};
            if (m.more()) {
                m_msg.add(std::move(m));
                continue;
            } else {
                m_msg.add(std::move(m));
                break;
            }
        }
        return std::move(m_msg);
    }

    /// Delegates the suspension/cancellation to the WaitFileDescriptor awaitable.
    inline void await_suspend(std::coroutine_handle<> h) noexcept { m_wait.await_suspend(h); }

public /* corral extensions */:
    inline bool await_cancel(std::coroutine_handle<> h) noexcept { return m_wait.await_cancel(h); }

    [[nodiscard]] inline bool await_must_resume() const noexcept
    {
        return m_wait.await_must_resume();
    }

private:
    inline bool has_pollin_events() const noexcept
    {
        return m_socket.get(zmq::sockopt::events) & ZMQ_POLLIN;
    }

private:
    WaitFileDescriptor m_wait;
    zmq::socket_ref m_socket;
    zmq::recv_flags m_flags;
    zmq::multipart_t m_msg;
};

AsyncSend.hpp

#pragma once

#include "WaitFileDescriptor.hpp"
#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <coroutine>
#include <exception>
#include <zmq.hpp>

struct SendError : public std::exception
{
public:
    inline auto what() const noexcept -> const char * override { return "Failed sending message"; }
};

///
/// \brief Asynchronously send message to a ZeroMQ socket.
/// \return [zmq::message_t] - The message to be sent.
/// \throw [SendError] - On send error.
///
struct [[nodiscard]] AsyncSend
{
public /* ctor */:
    inline AsyncSend(boost::asio::any_io_executor executor,
                     zmq::socket_ref socket,
                     zmq::message_t message,
                     zmq::send_flags flags = zmq::send_flags::none)
        : m_wait({
              .executor = executor,
              .handle = socket.get(zmq::sockopt::fd),
              .wait_type = boost::asio::posix::stream_descriptor::wait_write,
              // only allow resuming if there are poll-out events
              .resume_filter =
                  [this](std::coroutine_handle<> h) noexcept {
                      if (has_pollout_events()) {
                          h.resume();
                      } else {
                          /// If the file descriptor can be written but socket doesn't
                          /// have poll-in event, schedule for another awake.
                          m_wait.await_suspend(h);
                      }
                  },
          })
        , m_socket(socket)
        , m_msg(std::move(message))
        , m_flags(flags)
    {}

public /* interface */:
    inline auto enable_flag(zmq::send_flags flag) { m_flags = m_flags | flag; }
    inline auto disable_flag(zmq::send_flags flag) { m_flags = m_flags & (~flag); }

public /* awaitable */:
    ///
    /// \brief An optimization which avoids suspending the coroutine if it isn't necessary.
    /// \return true - when there are already poll-in events and `await_resume` will be
    /// immediately invoked.
    ///
    inline bool await_ready() const noexcept { return has_pollout_events(); }

    inline void await_resume()
    {
        auto result = m_socket.send(std::move(m_msg), zmq::send_flags::dontwait | m_flags);
        if (!result) [[unlikely]]
            throw SendError{};
    }

    /// Delegates the suspension/cancellation to the WaitFileDescriptor awaitable.
    inline void await_suspend(std::coroutine_handle<> h) noexcept { m_wait.await_suspend(h); }

public /* corral extensions */:
    inline bool await_cancel(std::coroutine_handle<> h) noexcept { return m_wait.await_cancel(h); }

    inline bool await_must_resume() const noexcept { return m_wait.await_must_resume(); }

private:
    inline bool has_pollout_events() const noexcept
    {
        return m_socket.get(zmq::sockopt::events) & ZMQ_POLLOUT;
    }

private:
    WaitFileDescriptor m_wait;
    zmq::socket_ref m_socket;
    zmq::message_t m_msg;
    zmq::send_flags m_flags;
};

AsyncSendMultipart.hpp

#pragma once

#include "AsyncSend.hpp"
#include <zmq_addon.hpp>

///
/// \brief Asynchronously send message to a ZeroMQ socket.
/// \return [zmq::multipart_t] - The message to be sent.
/// \throw [SendError] - On send error.
///
struct [[nodiscard]] AsyncSendMultipart
{
public /* ctor */:
    inline AsyncSendMultipart(boost::asio::any_io_executor executor,
                              zmq::socket_ref socket,
                              zmq::multipart_t message,
                              zmq::send_flags flags = zmq::send_flags::none)
        : m_wait({
              .executor = executor,
              .handle = socket.get(zmq::sockopt::fd),
              .wait_type = boost::asio::posix::stream_descriptor::wait_write,
              // only allow resuming if there are poll-out events
              .resume_filter =
                  [this, message = std::move(message)](std::coroutine_handle<> h) mutable noexcept {
                      this->resume_filter(h, message);
                  },
          })
        , m_socket(socket)
        /// `sndmore` flag should not be controlled by the user.
        , m_flags(flags & ~(zmq::send_flags::sndmore))
    {}

public /* interface */:
    inline auto enable_flag(zmq::send_flags flag)
    {
        m_flags = m_flags | flag;
        m_flags = m_flags & ~(zmq::send_flags::sndmore);
    }
    inline auto disable_flag(zmq::send_flags flag)
    {
        m_flags = m_flags & (~flag);
        m_flags = m_flags & ~(zmq::send_flags::sndmore);
    }

public /* awaitable */:
    ///
    /// \note The early-resume optimization isn't available here:
    /// The filter must be invoked to properly check if it's OK to submit message
    /// into the send queue.
    ///
    inline bool await_ready() const noexcept { return false; }

    inline void await_resume()
    {
        if (!m_success) [[unlikely]]
            throw SendError{};
    }

    /// Delegates the suspension/cancellation to the WaitFileDescriptor awaitable.
    inline void await_suspend(std::coroutine_handle<> h) noexcept { m_wait.await_suspend(h); }

public /* corral extensions */:
    inline bool await_cancel(std::coroutine_handle<> h) noexcept { return m_wait.await_cancel(h); }

    inline bool await_must_resume() const noexcept { return m_wait.await_must_resume(); }

private:
    inline bool has_pollout_events() const noexcept
    {
        return m_socket.get(zmq::sockopt::events) & ZMQ_POLLOUT;
    }

    ///
    /// \brief This function is used as a callback. It's called when the file descriptor
    /// can be written with data.
    /// \note This is marked as const due to a quirk in C++: This function is being referred
    /// inside constructor and is technically UB. In practice this means the const qualifier
    /// is not propogated. Hence taking the conservative approach by assuming it's always const.
    ///
    inline void resume_filter(std::coroutine_handle<> h, zmq::multipart_t &msg) const noexcept
    {
        if (has_pollout_events()) {
            ///
            /// \brief This boolean is used to indicate if this function had sent anything:
            /// On sending failure, it can either be:
            ///
            /// - that the queue is full (and therefore should be suspended
            ///   and scheduled for another awake).
            /// - or that the socket is disconnected.
            ///
            bool sent_something = false;
            ///
            /// Instead of repeatedly suspending and awaking, try to fill the send queue
            /// once, and fallback to suspension if no message can be submitted to the send
            /// queue at the moment.
            ///
            while (true) {
                auto m = msg.pop();
                const bool more = !msg.empty();
                const auto result = m_socket.send(std::move(m),
                                                  (more ? zmq::send_flags::sndmore
                                                        : zmq::send_flags::none)
                                                      | zmq::send_flags::dontwait | m_flags);

                if (result) [[likely]] {
                    sent_something = true;
                } else [[unlikely]] {
                    if (sent_something)
                        goto schedule_another_awake;

                    /// Resume without setting the success flag,
                    /// to indicate that the operation has failed.
                    h.resume();
                    return;
                }

                if (!more) {
                    m_success = true;
                    h.resume();
                    return;
                }
            }
        }

    schedule_another_awake:
        /// If the above operations neither success nor fail, it means that
        /// the multipart message haven't been sent. Schedule for another awake.
        m_wait.await_suspend(h);
    }

private:
    /// \note See note of `resume_filter` on why it's `mutable`.
    mutable WaitFileDescriptor m_wait;
    mutable zmq::socket_ref m_socket;
    mutable bool m_success = false;
    zmq::send_flags m_flags;
};

corral event loop trait for asio

Because currently corral's support for asio seems to be stuck for a previous version and doesn't work with latest asio.

template<>
struct corral::EventLoopTraits<boost::asio::io_context>
{
    using T = boost::asio::io_context;
    /// Returns a value identifying the event loop.
    /// Traits for any sort of wrappers should return ID
    /// of the wrapped event loop.
    static EventLoopID eventLoopID(T &ex) { return EventLoopID{std::addressof(ex)}; }

    /// Runs the event loop.
    static void run(T &ex) { ex.run(); }

    /// Tells the event loop to exit.
    /// run() should return shortly thereafter.
    static void stop(T &ex) { ex.stop(); }
};

the to_string helper used in the following example btw

auto to_string(const zmq::multipart_t &msgs) -> std::string
{
    using namespace std::string_view_literals;

    constexpr auto list_prefix = "["sv;
    constexpr auto list_postfix = "]"sv;
    constexpr auto element_prefix = "\""sv;
    constexpr auto element_postfix = "\""sv;
    constexpr auto delimiter = ", "sv;

    auto total_size = 0;
    for (auto &e : msgs) {
        total_size += e.size();
    }
    total_size += list_prefix.size() + list_postfix.size();
    total_size += msgs.size() * (element_prefix.size() + element_postfix.size())
                  + (msgs.size() - 1) * delimiter.size();

    std::string ret;
    ret.reserve(total_size);

    ret += list_prefix;
    for (auto &e : msgs) {
        ret += element_prefix;
        ret += e.to_string();
        ret += element_postfix;
        if (&e != &msgs.at(msgs.size() - 1)) {
            ret += delimiter;
        }
    }
    ret += list_postfix;

    return ret;
}

It can be used like this:

boost::asio::io_context io{};
using namespace std::string_literals;

auto send_and_receive(zmq::context_t &ctx) -> corral::Task<void>
{
    zmq::socket_t socket{ctx, zmq::socket_type::req};
    socket.connect("tcp://127.0.0.1:5555");

    zmq::multipart_t msg;
    msg.add(zmq::message_t{"Test"s});
    msg.add(zmq::message_t{"123"s});
    co_await AsyncSendMultipart{io.get_executor(), socket, std::move(msg)};

    zmq::multipart_t rec = co_await AsyncReceiveMultipart{io.get_executor(), socket};
    std::println("{}", to_string(rec));

    co_return;
}

On co_await, the current coroutine is suspended and any other tasks can continue execute concurrently.

Quirks:

• I don't know if it will work with anything other than TCP (Surprisingly) It seems to work with IPC (on Linux) and inproc as well.
• async_send doesn't seem to respond to cancellation, but async_recv does. those two functions are mostly just mirrored (send <-> receive). maybe the sd.async_wait can't be cancelled with wait_write? Am I missing something? asio quirk? Turns out the cancellation worked correctly, it's just zmq::context_t got stuck at destructor after sending to an unreachable endpoint (I've encountered the exact issue with PyZMQ before). Ideally there should be a way to make the send state visible to the caller side so the coroutine can be suspended at the async_send (rather than returning immediately, pretending that it had sent successfully but actually isn't, then got stuck at close as it attempts to send the message in the buffer)
• Hard dependency on asio
• Not sure how to make it work on Windows yet since I don't have access to Windows machine right away.

[SidneyCogdill]

An official coroutine support would be great (albeit hard, as zmq doesn't have callbacks like other pre-C++20 asynchronous libraries, and C++20 coroutines are basically encapsulation of callbacks). the helper functions for cobalt inherently relies on asio to watch the file descriptor. I don't think it will work for any other protocol than TCP.

[SidneyCogdill]

FYI I've implemented Sender & Receiver support (over 1000 LoC). I haven't wrote test cases for it yet. Unlike the corral coroutine one, this is thread safe (at the cost of some atomic<bool> fetching). It still depends on Asio to listen on the file descriptor events. It also depends on stdexec (but this will change once C++26 standard library is implemented by major vendors).

There is a performance hazard which I haven't decided the solution yet: When many actors are attempting to send/receive multipart messages on the same socket, with the current implementation every operation is async_wait-ed and will cause excessive wakeup. In other words it doesn't scale linearly with the amount of pending multipart message operations (every pending operations are awakened on every file descriptor events but only one of them gets into the critical section).

I can address this by adding a callback queue and only awake one operation per file descriptor event, which will make the performance scale linearly. However this also makes all multipart operations perform slightly slower as there is now the additional cost of the callback queue. More importantly, operating multipart message simultaneously from multiple actors on the same socket is just not a good idea in general as it's not designed to be thread safe, and I don't know if it's worth adding that overhead to improve performance for the case you want to avoid if you actually care about performance.

[SidneyCogdill]

Another hazard is when both the recv (single message) and recv_multipart are executing in parallel, theoretically there is a small window for race condition where recv enters the critical section but recv_multipart isn't aware of that. Then recv might accidentally take a message part from the multipart message, breaking the atomic property of multipart message. This can be trivially mitigated by making recv also set the socket-wide critical section flag, however again at the cost of slowing down all recv operations (2 exchange + 1 store + 1 load for each recv, where it originally only requires 1 load + 1 exchange).