我的服务器基于 boost spawn echo 服务器示例,并在此线程中进行了改进。真正的服务器很复杂,我做了一个更简单的服务器来显示问题:
服务器监听 12345 端口,从新连接接收 0x4000 字节数据。
客户端运行 1000 个线程,连接到服务器并发送 0x4000 字节数据。
问题:当客户端运行时,1秒后通过控制台中的Ctrl-Cio_context
终止客户端进程,然后服务器将停止,服务器进入无限循环并消耗100%的cpu。如果这没有发生,重复启动客户端并杀死它几次,它就会发生。也许几次后它会耗尽 TCP 端口,只需等待几分钟再试一次,它会在我的机器上杀死客户端 3~15 次后发生。
boost文档说是io_context.stopped()
用来判断是否停止
要么通过显式调用 stop(),要么由于工作用完
我从不调用io_context.stop()
,并使用 amake_work_guard(io_context)
来保持io_context
不停止,但为什么它仍然停止?
我的环境:Win10-64bit,boost 1.71.0
服务器代码:
#include <iostream>
using namespace std;
#include <boost/thread/thread.hpp>
#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>
using namespace boost;
using namespace boost::asio;
using namespace boost::asio::ip;
namespace ba=boost::asio;
#define SERVER_PORT 12345
#define DATA_LEN 0x4000
struct session : public std::enable_shared_from_this<session>
{
tcp::socket socket_;
boost::asio::steady_timer timer_;
boost::asio::strand<boost::asio::io_context::executor_type> strand_;
explicit session(boost::asio::io_context& io_context, tcp::socket socket)
: socket_(std::move(socket)),
timer_(io_context),
strand_(io_context.get_executor())
{ }
void go()
{
auto self(shared_from_this());
boost::asio::spawn(strand_, [this, self](boost::asio::yield_context yield)
{
spawn(yield, [this, self](ba::yield_context yield) {
timer_.expires_from_now(10s); // 10 second
while (socket_.is_open()) {
boost::system::error_code ec;
timer_.async_wait(yield[ec]);
// timeout triggered, timer was not canceled
if (ba::error::operation_aborted != ec) {
socket_.close();
}
}
});
try
{
// recv data
string packet;
// read data
boost::system::error_code ec;
ba::async_read(socket_,
ba::dynamic_buffer(packet),
ba::transfer_exactly(DATA_LEN),
yield[ec]);
if(ec) {
throw "read_fail";
}
}
catch (...)
{
cout << "exception" << endl;
}
timer_.cancel();
socket_.close();
});
}
};
struct my_server {
my_server() { }
~my_server() { }
void start() {
ba::io_context io_context;
auto worker = ba::make_work_guard(io_context);
ba::spawn(io_context, [&](ba::yield_context yield)
{
tcp::acceptor acceptor(io_context,
tcp::endpoint(tcp::v4(), SERVER_PORT));
for (;;)
{
boost::system::error_code ec;
tcp::socket socket(io_context);
acceptor.async_accept(socket, yield[ec]);
if (!ec) {
std::make_shared<session>(io_context, std::move(socket))->go();
}
}
});
// Run io_context on All CPUs
auto thread_count = std::thread::hardware_concurrency();
boost::thread_group tgroup;
for (auto i = 0; i < thread_count; ++i)
tgroup.create_thread([&] {
for (;;) {
try {
if (io_context.stopped()) { // <- this happens after killing Client process several times
cout << "io_context STOPPED, now server runs infinit loop with full cpu usage" << endl;
}
io_context.run();
}
catch(const std::exception& e) {
MessageBox(0, "This never popup", e.what(), 0);
}
catch(const boost::exception& e) {
MessageBox(0, "This never popup", boost::diagnostic_information(e).data(), 0);
}
catch(...) { MessageBox(0, "This never popup", "", 0); }
}
});
tgroup.join_all();
}
};
int main() {
my_server svr;
svr.start();
}
客户端:
#include <iostream>
#include <random>
#include <thread>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
using namespace std;
using boost::asio::ip::tcp;
namespace ba=boost::asio;
#define SERVER "127.0.0.1"
#define PORT "12345"
int main() {
boost::asio::io_context io_context;
static string data_0x4000(0x4000, 'a');
boost::thread_group tgroup;
for (auto i = 0; i < 1000; ++i)
tgroup.create_thread([&] {
for(;;) {
try {
tcp::socket s(io_context);
tcp::resolver resolver(io_context);
boost::asio::connect(s, resolver.resolve(SERVER, PORT));
ba::write(s, ba::buffer(data_0x4000));
} catch (std::exception e) {
cout << " exception: " << e.what() << endl;
} catch (...) {
cout << "unknown exception" << endl;
}
}
});
tgroup.join_all();
return 0;
}
更新解决方法:
我想和协程会发生问题io_context
,所以我尝试替换不必要spawn
的 to std::thread
,并且它有效,io_context
永远不会停止。但为什么还是会出现问题呢?
代替:
ba::spawn(io_context, [&](ba::yield_context yield)
{
tcp::acceptor acceptor(io_context,
tcp::endpoint(tcp::v4(), SERVER_PORT));
for (;;)
{
boost::system::error_code ec;
tcp::socket socket(io_context);
acceptor.async_accept(socket, yield[ec]);
if (!ec) {
std::make_shared<session>(io_context, std::move(socket))->go();
}
}
});
至:
std::thread([&]()
{
tcp::acceptor acceptor(io_context,
tcp::endpoint(tcp::v4(), SERVER_PORT));
for (;;)
{
boost::system::error_code ec;
tcp::socket socket(io_context);
acceptor.accept(socket, ec);
if (!ec) {
std::make_shared<session>(io_context, std::move(socket))->go();
}
}
}).detach();