networking - boost::asio tcp async_accept 处理程序未被调用但同步连接工作正常

Question

我有一个 TCP 客户端，它可以正常调用 TCP 套接字上的常规连接。但是，对 async_connect 的调用永远不会触发处理程序。他们都使用几乎相同的代码。区别仅在于调用连接与异步连接。

标题

#ifndef TCPCLIENT_H
#define TCPCLIENT_H

#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/asio.hpp>

#include <boost/thread/thread.hpp> 
#include <boost/thread/mutex.hpp>

using boost::asio::ip::tcp;

//How about an interface for outputting errors
class BoostTCPDebugOutputInterface
{
public:
    virtual void outputError(std::string& error)=0;
    virtual void outputDebug(std::string& debug) = 0;
    virtual void outputWarning(std::string& warning) = 0;
    virtual void outputInfo(std::string& info) = 0;

};

class ShutdownCallback
{
public:
    virtual void shutdown() = 0;
};

class BoostTCPConnection 
    : public boost::enable_shared_from_this<BoostTCPConnection>
{
public:
    typedef boost::shared_ptr<BoostTCPConnection> Connection_ptr;

    static Connection_ptr create(boost::asio::io_service& io_service, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner)
    {
        return Connection_ptr(new BoostTCPConnection(io_service, output, shutdowner));
    }

    virtual ~BoostTCPConnection(void);

    void start();
    bool isRunning();
    void setRunning(bool running);

    void enqueueMessageToSend(boost::shared_ptr<std::string>& message);

    boost::asio::ip::tcp::socket& socket()
    {
        return socket_;
    }

    void getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead);

    void getMessages(std::list< boost::shared_ptr<std::string> >& messages);

protected:
    BoostTCPConnection(boost::asio::io_service& io_service, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner) : socket_(io_service), errorOutput_(output), shutdowner_(shutdowner) {}

    void send();
    void handleWrite(const boost::system::error_code& error,size_t bytes_transferred);
    void handleReceive(const boost::system::error_code& error,size_t bytes_transferred);

    void readMessage();
    void handleReceive();

    std::string hostIP_;
    unsigned int hostPort_;

    boost::asio::io_service io_service_;
    boost::asio::ip::tcp::socket socket_;

    boost::mutex runningMutex_;
    bool running_;

    boost::mutex readMutex_;
    std::list< boost::shared_ptr<std::string> > receivedMsgs_;

    boost::mutex sendMutex_;
    std::list< boost::shared_ptr<std::string> > sendMsgs_;

    BoostTCPDebugOutputInterface* errorOutput_;
    ShutdownCallback* shutdowner_;

    static const size_t HEADERSIZE = 4;
};

class TCPClient
{
public:
    TCPClient();
    virtual ~TCPClient();

    bool start(std::string& hostIP,  unsigned int& hostPort, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner);
    void stop();

    void getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead);

    void send(boost::shared_ptr<std::string>& message);
    void getMessages(std::list< boost::shared_ptr<std::string> >& messages);

protected:
    void threadAction();

    void handleConnect(const boost::system::error_code& error);


    BoostTCPConnection::Connection_ptr connection_;

    boost::thread_group worker_threads_;

    boost::asio::io_service io_service_;

    BoostTCPDebugOutputInterface* errorOutput_;
    ShutdownCallback* shutdowner_;
};

#endif

CPP文件

#include "TCPClient.h"

#include <iostream>

BoostTCPConnection::~BoostTCPConnection(void)
{
}

void BoostTCPConnection::start()
{
    setRunning(true);
    while (isRunning())
    {

        bool readData(false);
        bool wroteData(false);

        if (!socket_.is_open())
        {
            std::string info("BoostTCPConnection::start() socket is no longer open.  ");
            errorOutput_->outputError(info);
            shutdowner_->shutdown();
            //Stop this NOW!!!
        }

        //Check if there are 4 bytes for packet size 
        //If there are read the size and then do an read to get the packet
        //The handler function should put the packet on a queue.
        boost::asio::socket_base::bytes_readable command(true);

        socket_.io_control(command);
        std::size_t bytes_readable = command.get();

        if ( bytes_readable >= HEADERSIZE )
        {
            readMessage();
            readData=true;
        }

        size_t sendSize(0);
        {
            boost::mutex::scoped_lock(sendMutex_);
            sendSize = sendMsgs_.size();
        }

        if ( sendSize > 0)
        {
            send();
        }


        if ( !readData && !wroteData )
            boost::this_thread::sleep(boost::posix_time::milliseconds(5)); 
    }
}

void BoostTCPConnection::readMessage()
{
    size_t messageSize(0);

    char temp[4]="";
    std::vector<char> header(4);

    boost::system::error_code ec;

    //Read the header which is the size
    size_t read=boost::asio::read(socket_, boost::asio::buffer(header), ec);
    if (ec)
    {       
        std::string info("BoostTCPConnection::readMessage errorcode ");
        info+=ec.message();
        errorOutput_->outputError(info);
        shutdowner_->shutdown();
        //TODO Signal the GUI to stop
        return;
    }
    memcpy((void*)(&messageSize),(void*)header.data(),4);

    std::vector<char> rcvBuffer(messageSize);

    read=boost::asio::read(socket_, boost::asio::buffer(rcvBuffer),ec);
    if (ec)
    {

        std::string info("BoostTCPConnection::readMessage errorcode ");
        info+=ec.message();
        errorOutput_->outputError(info);
        shutdowner_->shutdown();
        //TODO Signal the GUI to stop
        return;
    }

    rcvBuffer.push_back('\0');

    std::string test(rcvBuffer.begin(),rcvBuffer.end());
    boost::shared_ptr<std::string> message(new std::string(rcvBuffer.begin(),rcvBuffer.end()));

    receivedMsgs_.push_back(message);

}

void BoostTCPConnection::getMessages(std::list< boost::shared_ptr<std::string> >& messages)
{

    if (messages.size() > 0)
        messages.clear();

    {
        boost::mutex::scoped_lock lock(readMutex_); 
        receivedMsgs_.swap(messages);
    }
}

void BoostTCPConnection::handleReceive(const boost::system::error_code& error, size_t bytes_transferred)
{

    if (error)
    {
        std::ostringstream oss;
        oss<< "BoostTCPConnection::handleReceive got an error Code of "<<error.value()<<" and message "<<error.message()<<" bytes_transferred = "<<bytes_transferred<<std::endl;
        errorOutput_->outputError(oss.str());
        shutdowner_->shutdown();
        return;
    }

}


bool BoostTCPConnection::isRunning()
{
    boost::mutex::scoped_lock lock(runningMutex_);
    return running_;
}

void BoostTCPConnection::setRunning(bool running)
{
    boost::mutex::scoped_lock lock(runningMutex_);
    running_=running;
}

void BoostTCPConnection::enqueueMessageToSend(boost::shared_ptr<std::string>& message)
{
    boost::mutex::scoped_lock lock(sendMutex_);
    sendMsgs_.push_back(message);
}

void BoostTCPConnection::getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead)
{
    {
        boost::mutex::scoped_lock lock(sendMutex_);
        tcpToSend=sendMsgs_.size();
    }

    {
        boost::mutex::scoped_lock lock(readMutex_);
        tcpRead=receivedMsgs_.size();
    }

}


void BoostTCPConnection::send()
{
    if (sendMsgs_.empty())
        return;

    boost::shared_ptr<std::string> message;
    {
        message=sendMsgs_.front();
        sendMsgs_.pop_front();
    }

    char temp[4];
    size_t messageSize=message->size();
    memcpy(temp,&messageSize, 4);

    message->insert(0,temp, 4);

    boost::asio::async_write(socket_, boost::asio::buffer(*message),
        boost::bind(&BoostTCPConnection::handleWrite, shared_from_this(),
        boost::asio::placeholders::error,
        boost::asio::placeholders::bytes_transferred));

}

void BoostTCPConnection::handleWrite(const boost::system::error_code& error,size_t bytes_transferred)
{
    //Success
    if (error.value() == 0) 
        return;

    std::ostringstream oss;
    oss<< "BoostTCPConnection::handleWrite  got an error Code of "<<error.value()<<" and message "<<error.message()<<" with bytes_transferred = "<<bytes_transferred<<std::endl;
    errorOutput_->outputError(oss.str());
    shutdowner_->shutdown();

}

//***************************************************
//              TCPClient
//***************************************************

TCPClient::TCPClient()
{

}

TCPClient::~TCPClient()
{

}

void TCPClient::threadAction()
{
    io_service_.run();
}

bool TCPClient::start(std::string& hostIP,  unsigned int& hostPort, BoostTCPDebugOutputInterface* output, ShutdownCallback* shutdowner)
{
    bool bResult(false);
    errorOutput_=output;
    connection_ = BoostTCPConnection::create(io_service_, output, shutdowner);
    shutdowner_=shutdowner;

    //Use multiple threads to do my bidding
    for( int x = 0; x < 3; ++x )
    {
        worker_threads_.create_thread( 
            boost::bind( &TCPClient::threadAction, this )
            );
    }

    boost::system::error_code ec;
    try
    {
        boost::asio::ip::tcp::endpoint ep( boost::asio::ip::address_v4::from_string(hostIP), hostPort);
        if (connection_)
        {


            connection_->socket().async_connect(ep,
                                                boost::bind(&TCPClient::handleConnect, 
                                                this, boost::asio::placeholders::error));

                            //Synchronous code that works fine
                            //boost::system::error_code ec;         
            //connection_->socket().connect(ep, ec);
            /*if (!ec)
            {
                worker_threads_.create_thread( 
                    boost::bind( &TCPClient::handleConnect, this, ec ));

                bResult=true;
            }
            else
            {
                std::ostringstream oss;
                oss<< "BoostTCPConnection::start has an error "<<ec.message()<<std::endl;
                errorOutput_->outputError(oss.str());

            }*/

        }
    }
    catch (std::exception& e)
    {
        std::ostringstream oss;
        oss<< "BoostTCPConnection::start received the exception "<<e.what()<<std::endl;
        errorOutput_->outputError(oss.str());
    }

    return bResult;
}

void TCPClient::handleConnect(const boost::system::error_code& error)
{

    if (error )
    {
        std::ostringstream oss;
        oss<< "BoostTCPConnection::handleConnect received the error "<<error.message()<<std::endl;
        errorOutput_->outputError(oss.str());
        shutdowner_->shutdown();
        return;
    }

    if (connection_)
        connection_->start();

}

void TCPClient::stop()
{
    if (connection_)
        connection_->setRunning(false);

    io_service_.stop();
    worker_threads_.join_all();
}

void TCPClient::getQueueSize(unsigned long& tcpToSend, unsigned long& tcpRead)
{
    if (connection_)
        connection_->getQueueSize(tcpToSend, tcpRead);
    else
    {
        tcpToSend=0;
        tcpRead=0;
    }
}

void TCPClient::send(boost::shared_ptr<std::string>& message)
{
    if (connection_)
        connection_->enqueueMessageToSend(message);
}

void TCPClient::getMessages(std::list< boost::shared_ptr<std::string> >& messages)
{ 
    if (connection_)
        connection_->getMessages(messages);
}

Answer 1

问题是你的 io_service 在启动时有一个空队列并立即返回:

void TCPClient::threadAction() {
  io_service_.run();
  assert(0); // triggers right away because there's no async operation queued
}

来自 Stopping the io_service from running out of work :

Some applications may need to prevent an io_service object's run() call from returning when there is no more work to do. For example, the io_service may be being run in a background thread that is launched prior to the application's asynchronous operations. The run() call may be kept running by creating an object of type io_service::work.