Boost Beast Async Server Failing with Assertion failed: (id != TO::id) 在多个异步调用上

Question

断言失败：(id_ != T::id)，函数 try_lock，文件 /usr/local/include/boost/beast/websocket/detail/stream_base.hpp，第 91 行。

       // Echoes back all received WebSocket messages
class session : public std::enable_shared_from_this<session>
{
    websocket::stream<tcp::socket> ws_;
    boost::asio::strand<
            boost::asio::io_context::executor_type> strand_;
    boost::beast::multi_buffer buffer_;

public:
    // Take ownership of the socket
    explicit
    session(tcp::socket socket)
            : ws_(std::move(socket))
            , strand_(ws_.get_executor())
    {
    }

    // Start the asynchronous operation
    void
    run()
    {
        // Accept the websocket handshake
        ws_.async_accept(
                boost::asio::bind_executor(
                        strand_,
                        std::bind(
                                &session::on_accept,
                                shared_from_this(),
                                std::placeholders::_1)));
    }

    void
    on_accept(boost::system::error_code ec)
    {
        std::cout << std::this_thread::get_id() <<" : DO ACCEPT" << std::endl;

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

        std::ifstream ifs("tweetsample.txt");
        if(ifs.good())
        {
            auto tweet = std::string{};
            while(std::getline(ifs,tweet))
            {
                try
                {
                    std::this_thread::sleep_for(std::chrono::seconds(1));
                    auto t = json::parse(tweet);
                    std::string tweet_text = t["text"];
                    auto n = boost::asio::buffer_copy(buffer_.prepare(tweet_text.size()), 
boost::asio::buffer(tweet_text));
                buffer_.commit(n);
                    do_write();
                }
                catch(nlohmann::detail::type_error& ex)
                {
                       std::cout << ex.what() << std::endl;
                }
            }
        }
    }

    void
    do_read()
    {
        // Read a message into our buffer
        ws_.async_read(
                buffer_,
                boost::asio::bind_executor(
                        strand_,
                        std::bind(
                                &session::on_read,
                                shared_from_this(),
                                std::placeholders::_1,
                                std::placeholders::_2)));
    }

    void do_write()
    {
        std::cout << std::this_thread::get_id() << "do_write" << std::endl;
        ws_.async_write(
                buffer_.data(),
                boost::asio::bind_executor(
                        strand_,
                        std::bind(
                                &session::on_write,
                                shared_from_this(),
                                std::placeholders::_1,
                                std::placeholders::_2)));
    }

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

        // This indicates that the session was closed
        if(ec == websocket::error::closed)
            return;

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

        std::cout << boost::beast::buffers(buffer_.data()) << std::endl;

        // Echo the message
        ws_.text(ws_.got_text());
        ws_.async_write(
                buffer_.data(),
                boost::asio::bind_executor(
                        strand_,
                        std::bind(
                                &session::on_write,
                                shared_from_this(),
                                std::placeholders::_1,
                                std::placeholders::_2)));
    }

    void
    on_write(
            boost::system::error_code ec,
            std::size_t bytes_transferred)
    {
        std::cout << "on_write" << std::endl;
        boost::ignore_unused(bytes_transferred);

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

        // Clear the buffer
        buffer_.consume(buffer_.size());

        // Do another read
        do_read();
    }
};

//------------------------------------------------------------------------------

// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
    tcp::acceptor acceptor_;
    tcp::socket socket_;

public:
    listener(
            boost::asio::io_context& ioc,
            tcp::endpoint endpoint)
            : acceptor_(ioc)
            , socket_(ioc)
    {
        boost::system::error_code ec;

        // Open the acceptor
        acceptor_.open(endpoint.protocol(), ec);
        if(ec)
        {
            fail(ec, "open");
            return;
        }

        // Allow address reuse
        acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
        if(ec)
        {
            fail(ec, "set_option");
            return;
        }

        // Bind to the server address
        acceptor_.bind(endpoint, ec);
        if(ec)
        {
            fail(ec, "bind");
            return;
        }

        // Start listening for connections
        acceptor_.listen(
                boost::asio::socket_base::max_listen_connections, ec);
        if(ec)
        {
            fail(ec, "listen");
            return;
        }
    }

    // Start accepting incoming connections
    void
    run()
    {
        if(! acceptor_.is_open())
            return;
        do_accept();
    }

    void
    do_accept()
    {

        acceptor_.async_accept(
                socket_,
                std::bind(
                        &listener::on_accept,
                        shared_from_this(),
                        std::placeholders::_1));
    }

    void
    on_accept(boost::system::error_code ec)
    {
        if(ec)
        {
            fail(ec, "accept");
        }
        else
        {
            // Create the session and run it
            std::make_shared<session>(std::move(socket_))->run();
        }

        // Accept another connection
        do_accept();
    }
};  


    int main(int argc, char* argv[])
    {
        auto const address = boost::asio::ip::make_address("XXX.XX.XX.X");
        auto const port = static_cast<unsigned short>(std::atoi("XXXX"));
        auto const threads = 1;

        // The io_context is required for all I/O
        boost::asio::io_context ioc{threads};

        // Create and launch a listening port
        std::make_shared<listener>(ioc, tcp::endpoint{address, port})->run();

        // Run the I/O service on the requested number of threads
        std::vector<std::thread> v;
        v.reserve(threads - 1);
        for(auto i = threads - 1; i > 0; --i)
            v.emplace_back(
                    [&ioc]
                    {
                        ioc.run();
                    });
        ioc.run();        
        return EXIT_SUCCESS;
    }

我正在尝试使用 websocket 进行试验，目前当我接收接受时，我想逐行发送我从文件中读取的所有数据。

发送第一行后程序崩溃。 请指教我做错了什么。

Answer 1

我遇到了同样的问题，所以在查看失败的断言时我发现了这条评论：

    // If this assert goes off it means you are attempting to
    // simultaneously initiate more than one of same asynchronous
    // operation, which is not allowed. For example, you must wait
    // for an async_read to complete before performing another
    // async_read.
    //
    BOOST_ASSERT(id_ != T::id);

Answer 2

这是一个并发问题。 答案提到了here

当你调用 websocket::stream::async_write 时，你必须等到 操作完成（您的完成处理程序被调用）之前 再次调用 async_write。这在这里解释： https://www.boost.org/doc/libs/1_67_0/libs/beast/doc/html/beast/using_websocket/notes.html#beast.using_websocket.notes.thread_safety

如果您想一次发送多条消息，则需要 实现你自己的写队列。可以在这里找到一个例子： https://github.com/vinniefalco/CppCon2018/blob/8fc8528571561d2acee5c05f4b7a51861d1d496d/websocket_session.cpp#L87

在我的例子中，我使用了同步调用：

 ws_.write(net::buffer(std::string(message)));

Answer 3

    std::cout << "do_write" << std::endl;
    // Echo the message
    ws_.async_write(boost::asio::buffer(tweet),
                    boost::asio::bind_executor(strand_, std::bind(&session::on_write, shared_from_this(),
                                                                  std::placeholders::_1, std::placeholders::_2)));

这使用 local 缓冲区执行异步操作（tweet 参数在操作开始/完成之前超出范围）。

在read 方面，您使用data_ 解决了它，它是一个成员变量。你应该考虑这样的解决方案。