Integration with Boost ASIO + Cobalt as coroutines

[nyibbang]

Hello,

I've been trying to implement an integration of an event loop for sdbus-c++ with Boost ASIO through the use of C++20 coroutines. I've used Boost Cobalt for that.

I'm interested in having opinions and reviews on this code.

#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/error.hpp>
#include <boost/cobalt/task.hpp>
#include <boost/cobalt/race.hpp>
#include <boost/cobalt/this_coro.hpp>
#include <boost/cobalt/op.hpp>
#include <boost/system/system_error.hpp>
#include <sdbus-c++/IConnection.h>
#include <sys/poll.h>
#include <unistd.h>

boost::cobalt::task<void> dbus_connection_task(sdbus::IConnection& connection) {
    using descriptor = boost::cobalt::use_op_t::as_default_on_t<boost::asio::posix::stream_descriptor>;
    using timer = boost::cobalt::use_op_t::as_default_on_t<boost::asio::steady_timer>;
    using wait_type = descriptor::wait_type;

    try {
        auto exec = co_await boost::cobalt::this_coro::executor;

        // We might at some point need to wait for both read and write events, but ASIO does not
        // provide a mean to do that. So instead we save the last operation we polled and alternate
        // as much as possible between reads and writes when asked to poll for both.
        enum { read, write } last_operation = read;

        // A monotonic clock based timer for interrupting polls after a given timeout.
        timer timeout(exec);

        while (!co_await boost::cobalt::this_coro::cancelled) {
            // Process all pending events: the processing function returns true until there are no more
            // events pending.
            while (connection.processPendingEvent()) {}

            // Create 2 descriptor objects from file descriptors returned by
            // sdbus-c++. From the library documentation : PollData::fd is a bus
            // I/O fd. PollData::eventFd is an sdbus-c++ internal fd for
            // communicating important changes from other threads to the event
            // loop thread.
            auto poll_data = connection.getEventLoopPollData();
            descriptor connection_fd(exec, ::dup(poll_data.fd));
            descriptor loop_wakeup_fd(exec, ::dup(poll_data.eventFd));

            // Update the timer with the new timeout value and then create the
            // asynchronous waiting task.
            timeout.expires_after(poll_data.timeout);
            auto wait_timeout = timeout.async_wait();

            // From poll(2) documentation Possible values for poll events are
            // POLLIN, POLLOUT, POLLPRI, POLLERR, POLLHUP, POLLNVAL.
            //
            // Values POLLERR, POLLHUP and POLLNVAL are only returned as output
            // of poll and are ignored as input.
            //
            // So the only events that we might want to handle are POLLIN, POLLOUT and POLLPRI.
            // POLLPRI means that "there is some exceptional condition on the file descriptor".
            // POLLIN means that there is data to read and POLLOUT means that writing is possible.
            if (poll_data.events & POLLPRI) {
                // Wait for the error and then stop the loop.
                co_await boost::cobalt::race(
                    std::move(wait_timeout),
                    connection_fd.async_wait(wait_type::wait_error)
                );
                co_return;
            }

            auto event = wait_type::wait_read;
            if (poll_data.events & POLLOUT) {
                if (poll_data.events & POLLIN) {
                    // We both need to poll for read and write, but we can't do that with ASIO so
                    // just alternate from the last operation.
                    if (last_operation == read) {
                        event = wait_type::wait_write;
                        last_operation = write;
                    }
                    else {
                        assert(last_state == write);
                        event = wait_type::wait_read;
                        last_operation = read;
                    }
                }
                // Only poll for write.
                else {
                    event = wait_type::wait_write;
                    last_operation = write;
                }
            }
            // Only poll for read.
            else if (poll_data.events & POLLIN) {
                event = wait_type::wait_read;
                last_operation = read;
            }
            auto wait_event = connection_fd.async_wait(event);

            // loop_wakeup_fd is actually an eventfd descriptor, which are specified as containing
            // a single uint64 value. Reading the value should reset the state of the file
            // descriptor.
            std::array<std::uint64_t, 1> loop_wakeup_fd_value {{0}};
            auto wait_loop_wakeup = loop_wakeup_fd.async_read_some(
                boost::asio::buffer(loop_wakeup_fd_value)
            );

            co_await boost::cobalt::race(
                std::move(wait_timeout),
                std::move(wait_event),
                std::move(wait_loop_wakeup)
            );
        }
    }
    catch (const std::exception& ex) {
        // Do not rethrow an error if the task is cancelled.
        auto system_err = dynamic_cast<const boost::system::system_error*>(&ex);
        if (system_err && system_err->code() == boost::asio::error::operation_aborted) {
            co_return;
        }
        throw;
    }
}

Thank you in advance !

EDIT: Fixed errors with the code and added comments.