Boost Beast 异步服务器失败，断言失败:(id_ ！= T::id) 在多个 aync 调用中

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

这是一个并发问题。 答案在这里提到

当您调用 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)));

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

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