Boost :: Beast: сервер с конвейерной обработкой веб-сокетов - PullRequest
Новая
       28

Boost :: Beast: сервер с конвейерной обработкой веб-сокетов

4 голосов
/ 20 мая 2019

Я пишу сервер веб-сокетов c ++ с Boost Beast 1.70 и MySQL 8 C разъемом. К серверу будет подключено несколько клиентов одновременно. Особенность заключается в том, что каждый клиент будет выполнять около 100 запросов веб-сокетов подряд к серверу. Каждый запрос является «легким процессором» для моего сервера, но сервер выполняет «тяжелый по времени» SQL-запрос для каждого запроса.

Я запустил свой сервер с примером websocket_server_coro.cpp. Шаги сервера:

1) чтение веб-сокета

2) запрос sql

3) в вебсокете пишем

Проблема в том, что для данного пользователя сервер «заблокирован» на шаге 2 и не может прочитать, пока этот шаг и шаг 3 не будут завершены. Таким образом, 100 запросов решаются последовательно. Это слишком медленно для моего варианта использования.

Я прочитал, что неблокирующее чтение / запись невозможно с бустом зверя. Однако сейчас я пытаюсь выполнить async_read и async_write в сопрограмме. 

void ServerCoro::accept(websocket::stream<beast::tcp_stream> &ws) {
    beast::error_code ec;

    ws.set_option(websocket::stream_base::timeout::suggested(beast::role_type::server));

    ws.set_option(websocket::stream_base::decorator([](websocket::response_type &res) {
                res.set(http::field::server, std::string(BOOST_BEAST_VERSION_STRING) + " websocket-Server-coro");
            }));

    ws.async_accept(yield[ec]);
    if (ec) return fail(ec, "accept");

    while (!_bStop) {
        beast::flat_buffer buffer;
        ws.async_read(buffer, yield[ec]);

        if (ec == websocket::error::closed) {
            std::cout << "=> get closed" << std::endl;
            return;
        }

        if (ec) return fail(ec, "read");

        auto buffer_str = new std::string(boost::beast::buffers_to_string(buffer.cdata()));
        net::post([&, buffer_str] {

            // sql async request such as :
            // while (status == (mysql_real_query_nonblocking(this->con, sqlRequest.c_str(), sqlRequest.size()))) {
            //    ioc.poll_one(ec);
            // }
            // more sql ...

            ws.async_write(net::buffer(worker->getResponse()), yield[ec]); // this line is throwing void boost::coroutines::detail::pull_coroutine_impl<void>::pull(): Assertion `! is_running()' failed.
            if (ec) return fail(ec, "write");

        });
    }
}

Проблема в том, что строка с async_write выдает ошибку:

void boost :: coroutines :: detail :: pull_coroutine_impl :: pull (): утверждение `! is_running () 'не удалось.

Если заменить эту строку на sync_write, она работает, но сервер остается последовательным для данного пользователя. Я пытался выполнить этот код на однопоточном сервере. Я также попытался использовать одну и ту же прядь для async_read и async_write. Все еще есть ошибка утверждения.

Такой сервер невозможен с boost beast для websocket? Спасибо.

1 Ответ

2 голосов
/ 22 мая 2019

Следуя предложению Винни Фалько, я переписал код, используя в качестве примера «чат websocket» и «асинхронный сервер». Вот окончательный результат работы кода: 

void Session::on_read(beast::error_code ec, std::size_t bytes_transferred)
{
    boost::ignore_unused(bytes_transferred);

    if(ec == websocket::error::closed) return;  // This indicates that the Session was closed
    if(ec) return fail(ec, "read");

    net::post([&, that = shared_from_this(), ss = std::make_shared<std::string const>(std::move(boost::beast::buffers_to_string(_buffer.cdata())))] {
        /* Sql things that call ioc.poll_one(ec) HERE, for me the sql response go inside worker.getResponse() used below */

        net::dispatch(_wsStrand, [&, that = shared_from_this(), sss = std::make_shared < std::string const>(worker.getResponse())] {
            async_write(sss);
        });
    });
    _buffer.consume(_buffer.size()); // we remove from the buffer what we just read
    do_read(); // go for another read
}

void Session::async_write(const std::shared_ptr<std::string const> &message) {
    _writeMessages.push_back(message);

    if (_writeMessages.size() > 1) {
        BOOST_LOG_TRIVIAL(warning) << "WRITE IS LOCKED";
    } else {
        _ws.text(_ws.got_text());
            _ws.async_write(net::buffer(*_writeMessages.front()), boost::asio::bind_executor(_wsStrand, beast::bind_front_handler(
                    &Session::on_write, this)));
    }
}

void Session::on_write(beast::error_code ec, std::size_t)
{
    // Handle the error, if any
    if(ec) return fail(ec, "write");

    // Remove the string from the queue
    _writeMessages.erase(_writeMessages.begin());

    // Send the next message if any
    if(!_writeMessages.empty())
        _ws.async_write(net::buffer(*_writeMessages.front()), boost::asio::bind_executor(_wsStrand, beast::bind_front_handler(
                        &Session::on_write, this)));
}

Спасибо.

...