c++ - 套接字关闭但 session 引用未被正确销毁

Question

我正在使用 boost::asio 编写一个多线程服务器。有一个带有 X 线程的池，使用异步读写(基于 this 示例)。

服务器结构看起来是这样的:

服务器

管理程序线程并启动async_accept为每个新客户端创建新 session 。

session

代表客户端本身。让 socket 引用并管理来自她的 socket(客户端)的 async_read 和 async_write。还有超时管理。

---

有时客户端(硬件设备)死机，我的服务器没有他的回答。为了解决这个问题，我阅读了有关将 async_wait 与 deadline_timer (如 this example )一起使用的信息，并将其应用于我的软件，但发生了一些奇怪的事情:

当正常断开连接发生时，async 操作被取消(达到 operation_aborted 错误)并且 Session 对象被销毁。但是当设备卡住时，套接字关闭但 Session 对象没有被销毁并且他的实例保留在内存中，即使 socket.close() 已经被调用。

我简化了代码并放在下面:

服务器.h

    class Server
    {
    private:
        boost::asio::io_service& _io_service;
        boost::asio::ip::tcp::acceptor* _acceptor;
        boost::asio::ip::tcp::endpoint* _endpoint;

        boost::asio::signal_set _signals;

        Session_SP _session;
    public:
        Server(boost::asio::io_service& io_service);
        ~Server();


        /**
         * Queues async accept action.
         */
        virtual void queueAccept();

        /**
         * Async accept handler.
         */
        virtual void handleAccept(const boost::system::error_code& error);

        /**
         * Start the server
         */
        virtual void run();

        boost::asio::io_service& getIOService();

        /**
         * Shutdown the service
         */
        virtual void shutdown();
    };

服务器.cpp

#include "server.hpp"

Server::Server(boost::asio::io_service& io_service):
    _io_service(io_service), _signals(io_service)
{
    this->_endpoint = new boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), config.getServer().port);

    this->_acceptor = new boost::asio::ip::tcp::acceptor(io_service);

    this->_acceptor->open(boost::asio::ip::tcp::v4());
    this->_acceptor->bind(*this->_endpoint);
    this->_acceptor->listen();

    this->_signals.add(SIGINT);
    this->_signals.add(SIGTERM);
#if defined(SIGQUIT)
    this->_signals_.add(SIGQUIT);
#endif // defined(SIGQUIT)

    this->_signals.async_wait(boost::bind(&Server::shutdown, this));

    this->queueAccept();
}

Server::~Server()
{
    delete this->_acceptor;
    delete this->_endpoint;
}

void Server::queueAccept()
{
    this->_session.reset(new Session(*this));

    _acceptor->async_accept(
        this->_session->getSocket(),
        boost::bind(
            &Server::handleAccept,
            this,
            boost::asio::placeholders::error
        )
    );
}

void Server::handleAccept(const boost::system::error_code& error)
{
    if (!error)
        this->_session->start();

    this->queueAccept();
}

boost::asio::io_service& Server::getIOService()
{
    return this->_io_service;
}

void Server::shutdown()
{
    this->_io_service.stop();
}

session .h

    class Session:
        public boost::enable_shared_from_this<Session>
    {
    public:
        Session(Server& server);
        ~Session();

        bool stopped() const;

        virtual void start();

        virtual boost::asio::ip::tcp::socket& getSocket();

        virtual void disconnect();

        /**
         * Async read handler
         */
        void handleRead(const boost::system::error_code& error, size_t bytesTransfered);

        /**
         * Async write handler
         */
        void handleWrite(const boost::system::error_code& error);

        /**
         * Queues write action.
         */
        void queueWrite();

        /**
         * Push a packet to be sent on queue end
         */
        void pushPacket(protocols::SendPacket &packet);

        void handleDeadlineAsyncWait(boost::asio::deadline_timer* deadline);

        void handleDeadlineAsyncWaitKillConnection(boost::asio::deadline_timer* deadline);

    private:
        Server& _server;

        boost::asio::ip::tcp::socket _socket;

        boost::asio::io_service* _ioService;

        boost::asio::io_service::strand _strand;

        boost::asio::deadline_timer _input_deadline;

        boost::asio::deadline_timer _non_empty_output_queue;

        /**
         * Queue that stores the packet to be sent.
         */
        protocols::SendPacketQueue _writeQueue;

        /**
         * Referência do pacote que será atualizado.
         */
        Protocol* _protocol;

        /**
         * Queues the async_read acction.
         */
        virtual void queueRead();

        virtual void _pushPacket(protocols::SendPacket &packet);
    };

    typedef boost::shared_ptr<Session> Session_SP;

session .cpp

#include "session.hpp"

Session::Session(Server& server):
    _server(server), _socket(server.getIOService()), _protocol(NULL),
    _ioService(&server.getIOService()), _strand(server.getIOService()),
    _input_deadline(server.getIOService()),
    _non_empty_output_queue(server.getIOService())
{

    this->_input_deadline.expires_at(boost::posix_time::pos_infin);
    this->_non_empty_output_queue.expires_at(boost::posix_time::pos_infin);
}

Session::~Session()
{
}

bool Session::stopped() const
{
    return !_socket.is_open();
}

boost::asio::ip::tcp::socket& Session::getSocket()
{
    return this->_socket;
}

void Session::disconnect()
{
    this->_input_deadline.cancel();
    this->_non_empty_output_queue.cancel();
    try
    {
        this->getSocket().close();
        LOG("Session::disconnect : close successful!");
    }
    catch (void* e)
    {
        // Never reached here!!
    }
}

void Session::queueRead()
{
    this->_input_deadline.expires_from_now(boost::posix_time::seconds(30));

    boost::asio::async_read_until(
        _socket,
        _buffer,
        "\x004", // Just a test
        this->_strand.wrap(boost::bind(
            &Session::handleRead,
            this->shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred
        ))
    );
}

void Session::start()
{
    this->queueRead();

    this->_input_deadline.async_wait(
        this->_strand.wrap(boost::bind(
            &Session::handleDeadlineAsyncWait,
            shared_from_this(),
            &this->_input_deadline
        ))
    );

    this->queueWrite();
}

void Session::handleRead(const boost::system::error_code& error, size_t bytesTransfered)
{
    if (this->stopped())
        return;

    if (!error)
    {
        // ... a lot of code here, but isn't important
    }
    else if (error != boost::asio::error::operation_aborted)
        this->disconnect();
}

void Session::handleWrite(const boost::system::error_code& error)
{
    if (this->stopped())
        return;

    if (!error)
    {
        this->_writeQueue.pop_front(); // Dequeue
        this->queueWrite();
    }
    else
    {
        if (error != boost::asio::error::operation_aborted)
            this->disconnect();
    }
}

void Session::queueWrite()
{
    if (this->stopped())
        return;

    if (this->_writeQueue.empty())
    {
        this->_non_empty_output_queue.expires_at(boost::posix_time::pos_infin);
        this->_non_empty_output_queue.async_wait(
            boost::bind(&Session::queueWrite, shared_from_this())
        );
    }
    else
    {
        this->_input_deadline.expires_from_now(boost::posix_time::seconds(this->_server.getConfig().getServer().timeout));

        boost::asio::async_write(
            this->getSocket(),
            boost::asio::buffer(
                this->_writeQueue.front().getData(),
                this->_writeQueue.front().getDataSize()
            ),
            this->_strand.wrap(boost::bind(
                &Session::handleWrite,
                this,
                boost::asio::placeholders::error
            ))
        );
    }
}

void Session::handleDeadlineAsyncWait(boost::asio::deadline_timer* deadline)
{
    if (this->stopped())
        return;

    if (deadline->expires_at() <= boost::asio::deadline_timer::traits_type::now())
    {
        boost::system::error_code sdEc;
        this->getSocket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, sdEc);

        deadline->expires_from_now(boost::posix_time::seconds(15));

        deadline->async_wait(
            this->_strand.wrap(boost::bind(
                &Session::handleDeadlineAsyncWaitKillConnection,
                shared_from_this(),
                deadline
            ))
        );

    }
    else
    {
        deadline->async_wait(
            this->_strand.wrap(boost::bind(
                &Session::handleDeadlineAsyncWait,
                shared_from_this(),
                deadline
            ))
        );
    }
}

void Session::handleDeadlineAsyncWaitKillConnection(boost::asio::deadline_timer* deadline)
{
    this->disconnect();
}

Answer 1

您的 async_wait 超时处理程序应该取消未完成的 async_read() 而不是关闭套接字，否则套接字将保持打开状态。

void Session::handleDeadlineAsyncWait(boost::asio::deadline_timer* deadline)
{
    if (this->stopped())
        return;

    if (deadline->expires_at() <= boost::asio::deadline_timer::traits_type::now())
    {
        boost::system::error_code sdEc;
        this->getSocket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, sdEc);
        this->getSocket().cancel(); // <-- add this
    }
    else
    {
        deadline->async_wait(
            this->_strand.wrap(boost::bind(
                &Session::handleDeadlineAsyncWait,
                shared_from_this(),
                deadline
            ))
        );
    }
}

此外，在您的 Session::handleRead() 处理程序中，您应该检测到 boost::asio::error::operation_aborted 错误，因为这意味着读取是已取消。