Сопрограммы C++20 читают/записывают веб-сокет

Я хочу, чтобы веб-сокет работал в одном потоке и обслуживал операции чтения и записи по мере их поступления. Если делать нечего - проверьте на работу без сжигания циклов

Вам повезло, именно в этом и заключается асинхронное завершение.

Если нечего делать, ничего не произойдет, и поток(и) в пуле фактически перейдут в спящий режим, а это означает, что другие процессы в системе могут получить ход. Разница в том, что вместо «голого сна» используется «умный сон», например: сон просыпается, как только появляется какое-либо соответствующее событие ввода-вывода. Это может быть любое событие, поддерживаемое службой Asio, например файлы, каналы, UNIX или интернет-сокеты, последовательный порт, сигналы асинхронного процесса¹.

Что касается кода, позвольте мне сначала отметить, что ожидание появления:

    co_await asio::co_spawn(
        _ws.get_executor(), do_write() || do_read(),
        asio::use_awaitable);

это потенциально более дорогостоящий способ просто написать

    co_await (do_write() || do_read()); // idiomatic full-duplex!

И

    co_await std::suspend_always{}; // I want to switch context but ERROR

Это также всегда неоптимально, но если вы настаиваете, это может быть

    co_await post(asio::deferred);

Как подойти

Я бы сделал так, чтобы цикл записи запускался только тогда, когда что-то стоит в очереди. Это сразу упрощает синхронизацию доступа к очереди с помощью цепочки.

В качестве альтернативы посмотрите на channel как замену очереди. Поток записи может асинхронно получать данные из канала. Это также дает вам контроль над емкостью очереди: https://www.boost.org/doc/libs/1_84_0/doc/html/boost_asio/overview/channels.html

Каналы кажутся здесь лучшей идеей, потому что они также позволяют вам писать интерфейс Read естественно и оптимально.

Демо

Использование каналов кажется подходящим вариантом. Все сводится к такой сути:

const std::string _host, _port;
Stream            _ws;
Channel           _in{_ws.get_executor()}, _out{_ws.get_executor()};

Task do_run() {
    while (true) {
        try {
            co_await do_connect();
            co_await (do_write_loop() || do_read_loop());
        } catch (boost::system::system_error const& se) {
            std::cerr << "Error: " << se.code().message() << std::endl;
        }
    }
}

Task do_write_loop() {
    for (;;)
        co_await _ws.async_write(asio::buffer(co_await _out.async_receive()));
}

Task do_read_loop() {
    for (Message msg;; msg.clear()) {
        auto buf = asio::dynamic_buffer(msg);

        auto [ec, bytes] = co_await _ws.async_read(buf, as_tuple(asio::deferred));
        co_await _in.async_send(ec, std::move(msg));
    }
}

Добавление флага корректного завершения работы и задержки отключения в случае сбоя соединения:

Task do_run() {
    for (; !_close_requested; co_await delay(200ms)) {
        try {
            co_await do_connect();
            co_await (do_write_loop() || do_read_loop());
        } catch (boost::system::system_error const& se) {
            std::cerr << "Error: " << se.code().message() << std::endl;
        }
    }
}

Полный список

Прямой эфир на Колиру

#include <iostream>
#include <boost/asio.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/experimental/concurrent_channel.hpp>
#include <boost/beast.hpp>

namespace asio      = boost::asio;
namespace beast     = boost::beast;
namespace websocket = beast::websocket;
using namespace std::chrono_literals;
using namespace asio::experimental::awaitable_operators;
using boost::system::error_code;
using tcp = asio::ip::tcp;

struct Server {
    using Message  = std::string;
    using Task     = asio::awaitable<void>;
    using Channel  = asio::deferred_t::as_default_on_t< //
        asio::experimental::concurrent_channel<void(error_code, Message)>>;
    using Stream   = asio::deferred_t::as_default_on_t<websocket::stream<tcp::socket>>;
    using Resolver = asio::deferred_t::as_default_on_t<tcp::resolver>;
    using Opts     = websocket::stream_base;

    Server(asio::any_io_executor ex, std::string host, std::string port)
        : _host(std::move(host))
        , _port(std::move(port))
        , _ws(ex) {
        co_spawn(ex, do_run(), asio::detached);
    }

    bool Write(std::string data) { return _out.try_send(error_code{}, std::move(data)); }

    std::optional<std::string> Read() {
        Message ret;
        if (_in.try_receive([&](error_code ec, Message msg_) {
                if (ec)
                    throw boost::system::system_error(ec);
                ret = std::move(msg_);
            }))
            return ret;
        return std::nullopt;
    }

    void close() {
        _close_requested = true;
        post(_ws.get_executor(), [this] {
            if (error_code ignore; _ws.is_open())
                _ws.close({}, ignore);
        });
    }

  private:
    const std::string _host, _port;
    Stream            _ws;
    Channel           _in{_ws.get_executor(), 10}, _out{_ws.get_executor(), 10};
    std::atomic_bool  _close_requested{false};

    Task delay(auto duration_or_timepoint) {
        auto ex = co_await asio::this_coro::executor;
        co_await asio::steady_timer(ex, duration_or_timepoint).async_wait(asio::deferred);
    }

    Task do_run() {
        for (; !_close_requested; co_await delay(200ms)) {
            try {
                co_await do_connect();
                co_await (do_write_loop() || do_read_loop());
            } catch (boost::system::system_error const& se) {
                std::cerr << "Error: " << se.code().message() << std::endl;
            }
        }
    }

    Task do_write_loop() {
        for (;;)
            co_await _ws.async_write(asio::buffer(co_await _out.async_receive()));
    }

    Task do_read_loop() {
        for (Message msg;; msg.clear()) {
            auto buf = asio::dynamic_buffer(msg);

            auto [ec, bytes] = co_await _ws.async_read(buf, as_tuple(asio::deferred));
            co_await _in.async_send(ec, std::move(msg));

            if (ec)
                break;
        }
    }

    asio::awaitable<void> do_connect() {
        auto ex = co_await asio::this_coro::executor;
        if (error_code ignore; _ws.is_open()) {
            _ws.close({}, ignore);
            _in.reset();
            _out.reset();
        }

        auto eps = co_await Resolver(ex).async_resolve(_host, _port);
        co_await async_connect(beast::get_lowest_layer(_ws), eps);

        _ws.set_option(Opts::decorator([](websocket::request_type& req) {
            req.set(beast::http::field::user_agent, BOOST_BEAST_VERSION_STRING " WsConnect");
        }));

        co_await _ws.async_handshake(_host + ':' + _port, "/");
    }
};

int main() {
    asio::thread_pool ioc;
    Server s(make_strand(ioc), "localhost", "8989");

    using std::this_thread::sleep_for;

    for (auto msg : {"foo", "bar", "qux"}) {
        s.Write(msg);
        sleep_for(1s);

        while (auto response = s.Read())
            std::cout << "Received response " << quoted(*response) << std::endl;
    }

    s.close();
    ioc.join();
}

С живой демонстрацией против:

 websocketd --port 8989 -- \
      bash -c 'tee log | while read line; do echo "Responding to ($line)"; done'

¹ и некоторые особенности платформы, например порты завершения в Windows.