boost::asio::ip::tcp::acceptor 在使用 async_accept 接收连接请求时终止应用程序

Posted 2023-02-21

【中文标题】boost::asio::ip::tcp::acceptor 在使用 async_accept 接收连接请求时终止应用程序

【英文标题】：boost::asio::ip::tcp::acceptor terminates application when receiving connection request using async_accept

【发布时间】：2021-01-23 10:26:18

【问题描述】：

我想制作这个简单的服务器，它可以监听传入的连接请求、建立连接并发送一些数据。当我启动这个接受器时，它看起来工作正常，它等待那些传入的连接请求，但是当我的客户端尝试连接到这个接受器时，它会自动崩溃。我什至无法用catch(...) 捕获任何异常

当我启动这个程序时，它在终端中看起来像这样

但是当我尝试连接时

客户端应用收到这种错误代码

我的my_acceptor 课程是否存在根本问题？

class my_acceptor
public:
    my_acceptor(asio::io_context& ios, unsigned short port_num) :
        m_ios(ios),
        portport_num,
        m_acceptorios

    //start accepting incoming connection requests
    void Start()
    
        std::cout << "Acceptor Start" << std::endl;
        boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), port);
        m_acceptor.open(endpoint.protocol());
        m_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        m_acceptor.bind(endpoint);
        m_acceptor.listen();

        InitAccept();
    

    void Stop()

private:
    void InitAccept()
    
        std::cout << "Acceptor InitAccept" << std::endl;

        std::shared_ptr<asio::ip::tcp::socket> socknew asio::ip::tcp::socket(m_ios);

        m_acceptor.async_accept(*sock.get(),
            [this, sock](const boost::system::error_code& error)
            
                onAccept(error, sock);
            );
    

    void onAccept(const boost::system::error_code& ec, std::shared_ptr<asio::ip::tcp::socket> sock)
    
        std::cout << "Acceptor onAccept" << std::endl;
    

private:
    unsigned short port;
    asio::io_context& m_ios;
    asio::ip::tcp::acceptor m_acceptor;

;

以防万一这是包装my_acceptor的Server代码

class Server
public:
    Server()

    //start the server
    void Start(unsigned short port_num, unsigned int thread_pool_size)
    
        assert(thread_pool_size > 0);

        //create specified number of threads and add them to the pool
        for(unsigned int i = 0; i < thread_pool_size; ++i)
        
            std::unique_ptr<std::thread> th(
                new std::thread([this]()
                
                    m_ios.run();
                ));

            m_thread_pool.push_back(std::move(th));
        

        //create and start acceptor
        acc.reset(new my_acceptor(m_ios, port_num));
        acc->Start();
    

    //stop the server
    void Stop()
    
        work_guard.reset();
        acc->Stop();
        m_ios.stop();


        for(auto& th : m_thread_pool)
        
            th->join();
        
    

private:
    asio::io_context m_ios;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard = boost::asio::make_work_guard(m_ios);
    std::unique_ptr<my_acceptor> acc;
    std::vector<std::unique_ptr<std::thread>> m_thread_pool;
;

【问题讨论】：

请使您的代码独立。正如我的回答所证明的那样，这不是很多工作，而且我不需要猜测。

@sehe 谢谢你的回答，我刚醒来，会经历它:)

【参考方案1】：

至少有一个线程错误。 tcp::acceptor 不是线程安全的，您（可能）运行多个线程。因此，您需要从一个链中完成接受器访问。

my_acceptor(asio::io_context& ios, unsigned short port_num) :
    m_ios(ios),
    portport_num,
    m_acceptormake_strand(ios)

然后任何涉及它的操作都必须在该链上。例如，缺少的 Stop() 代码应如下所示：

void Stop()
    post(m_acceptor.get_executor(), [this]  m_acceptor.cancel(); );

我将最初的接受保持原样，因为此时不涉及多个线程。

同样，在 Start() 和 Stop() 中，您应该检查 acc 是否为空，因为 acc->Stop() 会抛出并且仅替换正在运行的 acc 会导致 Undefined Behaviour 由于删除仍然具有的实例进行中的异步操作。

在旁注中，如果您停止正在运行的接受器，则不需要m_ios.stop()。将来您可能必须发出信号停止任何客户端连接，以便线程自然加入。

这是我完成接受循环的方式：

void onAccept(error_code ec, std::shared_ptr<tcp::socket> sock)

    std::cout << "Acceptor onAccept " << ec.message() << " " << sock.get() << std::endl;
    if (!ec) 
        InitAccept();
    

请注意，除非套接字被取消（或其他错误），否则我们将继续接受。

我认为线程问题可能是您的大问题。我的建议工作后的结果：

Live On Coliru

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

#include <thread>

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

class my_acceptor 
public:
    my_acceptor(asio::io_context& ios, unsigned short port_num) :
        m_ios(ios),
        portport_num,
        m_acceptormake_strand(ios)

    //start accepting incoming connection requests
    void Start()
    
        std::cout << "Acceptor Start" << std::endl;
        tcp::endpoint endpoint(tcp::v4(), port);
        m_acceptor.open(endpoint.protocol());
        m_acceptor.set_option(tcp::acceptor::reuse_address(true));
        m_acceptor.bind(endpoint);
        m_acceptor.listen();

        InitAccept();
    

    void Stop()
        post(m_acceptor.get_executor(), [this]  m_acceptor.cancel(); );
    

private:
    void InitAccept()
    
        std::cout << "Acceptor InitAccept" << std::endl;

        auto sock = std::make_shared<tcp::socket>(m_ios);

        m_acceptor.async_accept(*sock,
            [this, sock](error_code error)  onAccept(error, sock); );
    

    void onAccept(error_code ec, const std::shared_ptr<tcp::socket>& sock)
    
        std::cout << "Acceptor onAccept " << ec.message() << " " << sock.get() << std::endl;
        if (!ec) 
            InitAccept();
        
    

private:
    asio::io_context& m_ios;
    unsigned short port;
    tcp::acceptor m_acceptor;
;

class Server
public:
    Server() = default;

    //start the server
    void Start(unsigned short port_num, unsigned int thread_pool_size)
    
        assert(!acc); // otherwise UB results
        assert(thread_pool_size > 0);

        //create specified number of threads and add them to the pool
        for(unsigned int i = 0; i < thread_pool_size; ++i)
        
            std::unique_ptr<std::thread> th(
                new std::thread([this]()  m_ios.run(); ));

            m_thread_pool.push_back(std::move(th));
        
        //create and start acceptor
        acc = std::make_unique<my_acceptor>(m_ios, port_num);
        acc->Start();
    

    //stop the server
    void Stop()
    
        work_guard.reset();
        if (acc) 
            acc->Stop();
        
        //m_ios.stop();

        for(auto& th : m_thread_pool) 
            th->join();
        
        acc.reset();
    

private:
    asio::io_context m_ios;
    asio::executor_work_guard<asio::io_context::executor_type>
        work_guard = make_work_guard(m_ios);
    std::unique_ptr<my_acceptor> acc;
    std::vector<std::unique_ptr<std::thread>> m_thread_pool;
;

int main() 
    Server s;
    s.Start(6868, 1);
    std::this_thread::sleep_for(10s);
    s.Stop();

以 netcat 作为客户端进行测试：

for msg in one two three; do
    sleep 1
    nc 127.0.0.1 6868 <<< "$msg"
done

打印

Acceptor Start
Acceptor InitAccept
Acceptor onAccept Success 0x1f26960
Acceptor InitAccept
Acceptor onAccept Success 0x7f59f80009d0
Acceptor InitAccept
Acceptor onAccept Success 0x7f59f8000a50
Acceptor InitAccept
Acceptor onAccept Operation canceled 0x7f59f80009d0

【讨论】：

兄弟非常感谢您的努力 :) 这对我来说意义重大