Ускорить синхронизацию Asio между TCP-клиентом и сервером

std::async не так явно говорит о многопоточности, как мне бы хотелось. Я бы предпочел просто std::thread.

Тем не менее, есть еще одна проблема, которая может вас сбить с толку: asio::read не только блокирует, но если, как вы сделали, вы не ограничите буфер до точного количества байтов, которые должны быть получены, вызов будет блокироваться «на неопределенный срок», пока либо

• (А) буфер заполнен
• (B) произошла ошибка (например, сервер прерывает соединение)

B никогда не произойдет в вашем тесте. Здесь вы либо используете read_some, либо сервер заранее закрывает сеансы.

И последнее, но не менее важное: есть ли гонки, в которых сервер может не слушать в то время, когда клиент connects? Я не думаю, что это так, потому что конструктор EchoServer открывает акцептор и инициирует асинхронное принятие. Даже если служба еще не запущена, это означает, что ядро ​​должно ставить в очередь входящие IP-пакеты для этого сокета.

Проверенное объявление

В этом листинге было много улучшений (например, печать только полученных байтов вместо всего буфера, их цитирование, трассировка с отметками времени, идентификаторами потоков и эксклюзивным доступом к консоли, отчеты об ошибках и обработка (например, EOF), использование loopback() вместо. .. синтаксический анализ строки «127.0.0.1», создание порта без знака, явное использование потока над std::async, отсутствие ручного переопределения размеров буфера в asio::buffer() вызовах, отмена слушателя вместо жесткой остановки контекста ввода-вывода, возможно, еще что-то, о чем я забыл) .

Жить на Колиру

#include <boost/asio.hpp>
#include <iomanip>
#include <iostream>

namespace asio = boost::asio;
using boost::asio::ip::tcp;
using boost::system::error_code;

namespace {
    static auto timestamp() {
        auto now = std::chrono::high_resolution_clock::now;
        using namespace std::chrono_literals;
        static auto start = now();
        auto        t     = now();

        return (t - start) / 1.ms;
    }

    static std::atomic_int tid_gen = 0;
    thread_local int       tid     = tid_gen++;
    std::mutex             console_mx;

    void trace(auto const&... args) {
        std::lock_guard lk(console_mx);
        ((std::cout << "T:" << tid << std::right << std::setw(10) << timestamp() << "ms ") << ... << args)
            << std::endl;
    }
} // namespace

class Session : public std::enable_shared_from_this<Session> {
  public:
    Session(tcp::socket socket) : socket_(std::move(socket)) {}
    void start() { do_read(); }

  private:
    void do_read() {
        trace("Server reads.");
        auto self = shared_from_this();
        socket_.async_read_some(asio::buffer(data_), [this, self](error_code ec, size_t length) {
            trace("Session read: ", ec.message(), " ", length);
            if (!ec || (length && ec == asio::error::eof))
                do_write(length);
        });
    }

    void do_write(size_t length) {
        trace("Server writes: ", quoted(std::string_view(data_.data(), length)));
        auto self = shared_from_this();
        async_write(socket_, asio::buffer(data_, length), [this, self](error_code ec, size_t length) {
            trace("Session write: ", ec.message(), " ", length);
            if (!ec)
                do_read();
        });
    }

    tcp::socket socket_;
    std::array<char, 1024> data_;
};

class EchoServer {
  public:
    EchoServer(asio::any_io_executor ex, asio::ip::address ipAddress, short port)
        : acceptor_(ex, tcp::endpoint(ipAddress, port)) {
        do_accept();
    }

    void stop() {
        post(acceptor_.get_executor(), [this] {
            acceptor_.cancel(); /* or close() */
        });
    }

  private:
    void do_accept() {
        acceptor_.async_accept(                    //
            make_strand(acceptor_.get_executor()), //
            [this](error_code ec, tcp::socket socket) {
                if (ec)
                    trace("Accept: ", ec.message());
                if (!ec) {
                    trace("Accepted ", socket.remote_endpoint());
                    std::make_shared<Session>(std::move(socket))->start();

                    do_accept();
                }
            });
    }

    tcp::acceptor acceptor_;
};

int main() {
    std::cout << std::fixed << std::setprecision(5);

    trace("Main start");
    auto const     ipAddress = asio::ip::address_v4::loopback();
    uint16_t const port      = 8080;

    asio::io_context io_context;
    EchoServer       echoServer(io_context.get_executor(), ipAddress, port);

    auto io_thread = std::thread([&]() {
        io_context.run();
        trace("Service has finished");
    });

    {
        // client code:
        tcp::socket socket(io_context);

        tcp::resolver resolver(io_context);
        connect(socket, resolver.resolve(ipAddress.to_string(), std::to_string(port)));

        {
            std::string const request{"test123"};
            write(socket, asio::buffer(request));
            trace("Main client wrote request: ", quoted(request));
        }

        {
            std::array<char, 1024> reply;
            size_t n = socket.read_some(asio::buffer(reply));
            trace("Main reply is: ", quoted(std::string_view(reply.data(), n)));
        }
    } // destructor closes connection

    echoServer.stop();
    trace("Main service stopped, waiting");
    io_thread.join();
    trace("Main bye");
}

Отпечатки, например.

T:0   0.00171ms Main start
T:0   0.36595ms Main client wrote request: "test123"
T:1   0.42401ms Accepted 127.0.0.1:48852
T:1   0.47220ms Server reads.
T:1   0.53624ms Session read: Success 7
T:1   0.55286ms Server writes: "test123"
T:1   0.60766ms Session write: Success 7
T:1   0.63068ms Server reads.
T:0   0.65522ms Main reply is: "test123"
T:0   0.70585ms Main service stopped, waiting
T:1   0.73496ms Session read: End of file 0
T:1   0.80807ms Accept: Operation canceled
T:1   0.82992ms Service has finished
T:0   0.87840ms Main bye

Или много работает локально:

Дайте мне знать, можете ли вы воспроизвести поведение, которое не описали полностью.