c++ - 如何在销毁 boost::asio 实体时消除崩溃？

Question

笔记!!!问题是针对 boost::asio library 领域的专家的人。 .不幸的是，我不能让代码更紧凑，它包含描述问题的最少数量。该代码是示例，人工创建的。它崩溃的地方已知并在评论中描述，它旨在说明崩溃! NO need调试代码的任何帮助...
问题是关于如何设计 asio 服务器，而不是关于 - 它在哪里崩溃!!!
此示例接近官方 boost::asio 文档中的“聊天服务器”设计。但是，与官方示例不同，只有连接类的对象是动态创建/销毁的，在我的示例中，服务器及其连接类实体都是动态创建/销毁的......我确信这种模式的实现应该在asio爱好者中广为人知，下面描述的问题应该已经有人解决了......
请看代码。
在这里，CAsioServer 和 CAsioConnection 的实体是动态创建和销毁的。

#include <map>
#include <array>
#include <set>
#include <vector>
#include <deque>
#include <thread>
#include <iostream>
#include <asio.hpp>
#include <iomanip>


class CAsioConnection
    : public std::enable_shared_from_this<CAsioConnection>
{
public:
    using PtrType = std::shared_ptr<CAsioConnection>;

    CAsioConnection(asio::ip::tcp::socket socket, std::set<CAsioConnection::PtrType>& connections)
        : socket_(std::move(socket)), connections_(connections)
    {
        std::cout << "-- CAsioConnection is creating, socket: " << socket_.native_handle() << "\n";
    }

    virtual ~CAsioConnection()
    {
        std::cout << "-- CAsioConnection is destroying , socket: " << socket_.native_handle() << "\n";
    }

    void read() { do_read(); }

private:
    void do_read(void)
    {
        uint8_t buff[3];

        asio::async_read(socket_, asio::buffer(buff,3),
            [this](std::error_code ec, std::size_t  /*length*/) {
            if (!ec)
            {
                do_read();
            }
            else
            {
                std::cout << "-- CAsioConnection::do_read() error : " << ec.message() << "\n";
                // Here is the crash N2
                connections_.erase(shared_from_this());
                // Crash may be fixed by the code below
                //if (ec.value() != 1236) // (winerror.h) #define ERROR_CONNECTION_ABORTED 1236L
                //  connections_.erase(shared_from_this());
            }
        });
    }

    asio::ip::tcp::socket socket_;
    std::set<CAsioConnection::PtrType>& connections_;
};

class CAsioServer
    : public std::enable_shared_from_this<CAsioServer>
{
public:
    using PtrType = std::shared_ptr<CAsioServer>;

    CAsioServer(int port, asio::io_context& io, const asio::ip::tcp::endpoint& endpoint)
        : port_(port), acceptor_(io, endpoint)
    {
        std::cout << "-- CAsioServer is creating, port: " << port_ << "\n";
    }

    virtual ~CAsioServer()
    {
        std::cout << "-- CAsioServer is destroying , port: " << port_ << "\n";
    }

    int port(void) { return port_; }

    void accept(void) { do_accept(); }
private:
    void do_accept()
    {
        acceptor_.async_accept([this](std::error_code ec, asio::ip::tcp::socket socket) {
            if (!ec)
            {
                std::cout << "-- CAsioServer::do_accept() connection to socket: " << socket.native_handle() << "\n";
                auto c = std::make_shared<CAsioConnection>(std::move(socket), connections_);
                connections_.insert(c);
                c->read();
            }
            else
            {
                // Here is the crash N1
                std::cout << "-- CAsioServer::do_accept() error : " << ec.message() << "\n";
                // Crash may be fixed by the code below
                //if (ec.value() == 995) // (winerror.h) #define ERROR_OPERATION_ABORTED 995L
                //  return;
            }
            // Actually here is the crash N1 )), but the fix is above...
            do_accept();
        });
    }

    int port_;
    asio::ip::tcp::acceptor acceptor_;
    std::set<CAsioConnection::PtrType> connections_;
};

//*****************************************************************************

class CTcpBase
{
public:
    CTcpBase()
    {
        // heart beat timer to keep it alive
        do_heart_beat();
        t_ = std::thread([this] {
            std::cout << "-- io context is RUNNING!!!\n";
            io_.run();
            std::cout << "-- io context has been STOPED!!!\n";
        });
    }

    virtual ~CTcpBase()
    {
        io_.stop();

        if (t_.joinable())
            t_.join();
    }

    void add_server(int port)
    {
        io_.post([this, port] 
        {
            for (auto s : servers_)
                if (port == s->port())
                    return;

            auto endpoint = asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port);
            auto s = std::make_shared<CAsioServer>(port, io_, endpoint);
            s->accept();
            servers_.insert(s);
        });
    }

    void remove_server(int port)
    {
        io_.post([this, port] 
        {
            for (auto s : servers_)
                if (port == s->port())
                    { servers_.erase(s); return; }
        });
    }

private:

    void do_heart_beat(void)
    {
        std::cout << "-- beat\n";
        auto timer = std::make_shared<asio::steady_timer>(io_, asio::chrono::milliseconds(3000));
        timer->async_wait([timer, this](const asio::error_code& ec) {
            do_heart_beat();
        });
    }

    asio::io_context io_;
    std::thread t_;
    std::set<CAsioServer::PtrType> servers_;
};

//*****************************************************************************

int main(void)
{
    CTcpBase tcp_base;

    std::cout << "CONNECT the server to port 502\n";
    tcp_base.add_server(502);

    std::this_thread::sleep_for(std::chrono::seconds(20));
    
    std::cout << "REMOVE the server from port 502\n";
    tcp_base.remove_server(502);

    std::this_thread::sleep_for(std::chrono::seconds(10));

    return 0;
}

它假设 CTcpBase::add_server()和 CTcpBase::remove_server()将由来自不同线程的外部客户端调用。 asio 上下文在它自己的线程中处理它。
让我们考虑两种情况:

启动应用程序并等待半分钟。
崩溃发生在 CAsioServer::do_accept()请参阅下面的输出。
Debug Console Output

开始申请。任何外部客户端连接到端口 502 并等待不到 20 秒。
崩溃发生在 CAsioConnection::do_read()请参阅下面的输出。
Debug Console Output

似乎 asio 框架调用推迟了 asio::async_read()和 acceptor_.async_accept()当其类的实体已经被破坏时处理程序。
我已经通过错误检查修复了处理程序，但该解决方案似乎并不可靠。谁知道可能还有什么其他错误和场景……有时，当客户端断开连接时，我需要清理 connection_设置为 asio::async_read() ，我如何确定服务器或连接对象仍然存在？...
有什么方法可以询问 boost::asio 框架以防止为已经被销毁的对象调用延迟的处理程序？或者如何识别 (be 100% sure)对象已经被销毁的错误代码？或者我在 asio 的范围内还有其他解决方案或设计模式 - 如何在一个正在运行的线程中处理动态创建/销毁的服务器及其连接，而无需互斥锁和其他东西......

Answer 1

首先检查您的 io_service严格单线程运行。这在代码中是不可见的。如果不是，那么共享状态(如 connections_ )需要同步访问。

In fact you can have a logical strand in the form of the accept loop, but to take advantage of this you should make all accesses to connections_ happen there, see e.g.

• here where we have the session list hold the sessions directly, no shared-pointer necessary at all: How to pass a boost asio tcp socket to a thread for sending heartbeat to client or server
• or here where we do have shared pointers and we store weak pointers in the sessions list, that can be "garbage collected" from inside the accept loop: ASIO - How to stop simple coroutine based server?

更新

buff是一个局部变量，这会导致未定义的行为，因为它在 async_read 操作的整个时间内都无效。

一般来说，拥有 shared_from_this 并没有什么意义。成语 和 还保留一个已经决定生命周期的共享指针容器。
您的问题似乎是有时 CAsioServer被简单地销毁，这意味着 connections_ 的所有元素被释放，当时他们的CAsioConnection对象可能被破坏。它还会破坏CAsioServer .
每当一个 Asio 对象被破坏时，任何挂起的异步操作都会失败并返回 asio::error:operation_aborted ，这确实意味着你有回应。但是，当调用完成处理程序时，对象已经变得无效。
在 my comment我刚刚注意到缺少一个关键成分:您永远不会捕获/绑定(bind)指向 CAsioConnection 的共享指针在任何完成处理程序中 .
这是非常不习惯的。
相反，您使用共享指针来管理生命周期。如果您还需要一个连接列表，则将其设为弱指针列表，以便它只观察生命周期。

变化点:

无需使服务器 enable_shared_from_this

connections_应该持有弱指针甚至非拥有指针。弱指针显然在这里更安全。事实上，您可以选择丢弃该容器，因为似乎没有任何东西在使用它。在下面的示例中，我选择保留它，以便您可以看到它的运行情况。

捕获shared_from_this在完成处理程序中确保对象在触发时仍然有效:

 asio::async_read(socket_, asio::buffer(buff,3),
     [this, self=shared_from_this()](error_code ec, std::size_t  /*length*/) {

简化
备注 我选择了 std::list因为它消除了对等式/排序的需要(参见 std::owner_less<> )，因为在 CAsioConnection 中存储对容器的引用的方式变得丑陋类 - 使其循环依赖(在实例化 CAsioConnection 类之前， owner_less<> 类型尚未完成)。我只是选择了(不需要的？)复杂性。
Live On Coliru

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

namespace asio = boost::asio;
using error_code = boost::system::error_code; // compat

class CAsioConnection : public std::enable_shared_from_this<CAsioConnection> {
  public:
    using PtrType = std::shared_ptr<CAsioConnection>;

    CAsioConnection(asio::ip::tcp::socket socket) : socket_(std::move(socket)) {
        log(__FUNCTION__);
    }

    ~CAsioConnection() { log(__FUNCTION__); }

    void read() { do_read(); }

  private:
    void log(std::string_view msg) const {
        error_code ec;
        std::clog << msg << ", socket: " << socket_.remote_endpoint(ec) << "\n";
    }

    uint8_t buff[256];
    void do_read() {
        asio::async_read(socket_, asio::buffer(buff),
             [this, self = shared_from_this()](error_code ec, std::size_t length) {
                 if (!ec) {
                     log(__FUNCTION__ + (" length: " + std::to_string(length)));
                     do_read();
                 } else {
                     log(__FUNCTION__ + (" error: " + ec.message()));
                 }
             });
    }

    asio::ip::tcp::socket socket_;
};

class CAsioServer {
  public:
    CAsioServer(asio::io_context& io, const asio::ip::tcp::endpoint& endpoint)
            : acceptor_(io, endpoint) { log(__FUNCTION__); }

    ~CAsioServer() { log(__FUNCTION__); }
    int port() const { return acceptor_.local_endpoint().port(); }
    void accept() { do_accept(); }

  private:
    void do_accept() {
        acceptor_.async_accept([this](error_code ec,
                                      asio::ip::tcp::socket socket) {
            if (!ec) {
                auto c = std::make_shared<CAsioConnection>(std::move(socket));
                connections_.push_back(c);
                c->read();
            } else {
                log(__FUNCTION__ + (" error: " + ec.message()));
            }

            connections_.remove_if(std::mem_fn(&WeakPtr::expired));

            if (acceptor_.is_open())
                do_accept();
        });
    }

    void log(std::string_view msg) const {
        std::clog << msg << ", port: " << port() << "\n";
    }

    asio::ip::tcp::acceptor acceptor_;
    using WeakPtr = std::weak_ptr<CAsioConnection>;
    std::list<WeakPtr> connections_;
};

int main() {
    boost::asio::io_context io;

    CAsioServer server(io, { {}, 7878 });
    server.accept();

    io.run_for(std::chrono::seconds(10));
}

输出:

./a.out& sleep 1; nc -w 1 127.0.0.1 7878 < main.cpp
CAsioServer, port: 7878
CAsioConnection, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() length: 256, socket: 127.0.0.1:50628
operator() error: End of file, socket: 127.0.0.1:50628
~CAsioConnection, socket: 127.0.0.1:50628
~CAsioServer, port: 7878