c++ - async_connect 不调用 TCP 客户端类中的处理程序

Question

我正在尝试为我的项目从 boost TCP 客户端示例创建一个客户端类，并且我注意到有时在连接到不存在的主机时不会调用 handle_connect。

我在堆栈上读过类似的问题，人们忘记运行 io_service 或在发布任何任务之前调用它，但我不认为这是我的情况，因为我在调用后立即启动 io_service.run() 线程async_connect、成功连接、网络无法访问以及我测试过的其他一些情况都工作得很好。

以下是完整列表:

tcp_client.hpp

#ifndef TCP_CLIENT_HPP
#define TCP_CLIENT_HPP

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/chrono.hpp>
#include <boost/thread/thread.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/make_shared.hpp>
#include <mutex>
#include <iostream>
#include <iomanip>

namespace com {

using boost::asio::ip::tcp;
using namespace std;

class client : public boost::enable_shared_from_this<client> {

private:

  std::mutex mx_;
  bool stopped_ = 1;
  boost::asio::streambuf ibuf_;
  boost::shared_ptr<boost::asio::io_service> io_service_;
  boost::shared_ptr<boost::asio::ip::tcp::socket> sock_;
  boost::shared_ptr<tcp::resolver::iterator> ei_;
  std::vector<std::string> inbound_;
  std::string host_, port_;

public:

  client() {}

  void connect( std::string host, std::string port ) {
    if (!stopped_) stop();
    host_ = host; port_ = port;
    io_service_.reset(new boost::asio::io_service);
    sock_.reset(new boost::asio::ip::tcp::socket(*io_service_));
    ei_.reset(new tcp::resolver::iterator);
    tcp::resolver r(*io_service_);
    ei_ = boost::make_shared<tcp::resolver::iterator>( r.resolve(tcp::resolver::query(host_, port_)) );
    stopped_ = 0;
    start_connect();
    boost::thread work( boost::bind(&client::work, shared_from_this()) );
    return;
  }

  bool is_running() {
    return !stopped_;
  }

  void stop() {
    stopped_ = 1;
    sock_->close();
    return;
  }

  void send(std::string str) {
    if (stopped_) return;
    auto msg = boost::asio::buffer(str, str.size());
    boost::asio::async_write( (*sock_), msg, boost::bind(&client::handle_write, shared_from_this(), _1) );
    return;
  }

  std::string pull() {
    std::lock_guard<std::mutex> lock(mx_);
    std::string msg;
    if (inbound_.size()>0) {
      msg = inbound_.at(0);
      inbound_.erase(inbound_.begin());
    }
    return msg;
  }

  int size() {
    std::lock_guard<std::mutex> lock(mx_);
    return inbound_.size();
  }

  void clear() {
    std::lock_guard<std::mutex> lock(mx_);
    inbound_.clear();
    return;
  }

private:

  void work() {
    if (stopped_) return;
    std::cout<<"work in"<<std::endl;
    io_service_->run();
    std::cout<<"work out"<<std::endl;
    return;
  }

  void start_connect() {
    if ((*ei_) != tcp::resolver::iterator()) {
      std::cout<<"Trying "<<(*ei_)->endpoint()<<std::endl;
      sock_->async_connect( (*ei_)->endpoint(), boost::bind(&client::handle_connect, shared_from_this(), boost::asio::placeholders::error) );
    } else {
      stop();
    }
    return;
  }

  void handle_connect(const boost::system::error_code& ec) {
    if (stopped_) return;

    if (!sock_->is_open()) {
      std::cout<<"Socket closed"<<std::endl;
      (*ei_)++;
      start_connect();
    } else if (ec) {
      std::cout<<"Connect error: "<<ec.message()<<std::endl;
      sock_->close();
      (*ei_)++;
      start_connect();
    } else {
      std::cout<<"Connected to "<<(*ei_)->endpoint()<<std::endl;
      start_read();
    }

    return;
  }

  void start_read() {
    if (stopped_) return;
    boost::asio::async_read_until((*sock_), ibuf_, "", boost::bind(&client::handle_read, shared_from_this(), boost::asio::placeholders::error));
    return;
  }

  void handle_read(const boost::system::error_code& ec) {
    std::lock_guard<std::mutex> lock(mx_);
    if (stopped_) return;
    if (ec) {
      std::cout<<"Read error: "<<ec.message()<<std::endl;
      stop();
      return;
    }

    std::string line;
    std::istream is(&ibuf_);
    std::getline(is, line);
    if (!line.empty() && inbound_.size()<1000) inbound_.push_back(line);

    start_read();
    return;
  }

private:

  void handle_write(const boost::system::error_code& ec) {
    if (stopped_) return;
    if (ec) {
      std::cout<<"Write error: "<<ec.message()<<std::endl;
      stop();
      return;
    }
    return;
  }

};

};

和tcp_test.cpp

#include "tcp_client.hpp"

int main(int argc, char* argv[]) {
  auto tcp_client = boost::shared_ptr<com::client>(new com::client);

  try {
    tcp_client->connect("192.168.1.15", "50000");
    boost::this_thread::sleep_for(boost::chrono::milliseconds(1000));
    tcp_client->connect("192.168.1.20", "50000");
  } catch (std::exception& e) {
    std::cerr<<"Exception: "<<e.what()<<std::endl;
  }

  int cnt=0;
  while (cnt<5) {
    std::cout<<cnt<<std::endl;
    cnt++;
    tcp_client->send("<test>");
    boost::this_thread::sleep_for(boost::chrono::milliseconds(500));
  }

  tcp_client->stop();

  while (tcp_client->size()>0) std::cout<<tcp_client->pull()<<std::endl;

  return 0;
}

我得到的输出是连接到环回服务器时的:

Trying 192.168.1.15:50000
work in
work out
Trying 192.168.1.20:50000
0
work in
Connected to 192.168.1.20:50000
1
2
3
4
work out
<test>
<test>
<test>
<test>
<test>

如您所见，192.168.1.20 正常工作。 192.168.1.15 不存在，但我预计它会抛出某种错误。相反 io_service.run() 立即返回，就像 async_connect 从未发布回调任务一样。也许它与端点迭代器有关，而不是 async_connect？

谁能解释一下为什么会这样？

然后我尝试隔离此代码中的问题:

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/chrono.hpp>
#include <boost/thread/thread.hpp>

boost::asio::io_service io_svc;
boost::asio::ip::tcp::socket sock(io_svc);
boost::asio::ip::tcp::resolver::iterator ei;

void work() {
  std::cout<<"work in"<<std::endl;
  io_svc.run();
  std::cout<<"work out"<<std::endl;
  return;
}

void stop() {
  sock.close();
  return;
}

void start_connect();

void handle_connect(const boost::system::error_code& ec) {
  if (!sock.is_open()) {
    std::cout<<"Socket closed"<<std::endl;
    ei++;
    start_connect();
  } else if (ec) {
    std::cout<<"Connect error: "<<ec.message()<<std::endl;
    sock.close();
    ei++;
    start_connect();
  } else {
    std::cout<<"Connected to "<<ei->endpoint()<<std::endl;
  }
  return;
}

void start_connect() {
  if (ei != boost::asio::ip::tcp::resolver::iterator()) {
    std::cout<<"Trying "<<ei->endpoint()<<std::endl;
    sock.async_connect( ei->endpoint(), boost::bind(handle_connect, boost::asio::placeholders::error) );
  } else {
    stop();
  }
  return;
}

int main(int argc, char* argv[]) {

  std::string host="192.168.1.15", port="50000";

  boost::asio::ip::tcp::resolver r(io_svc);
  ei = r.resolve(boost::asio::ip::tcp::resolver::query(host, port));
  start_connect();
  boost::thread* thr = new boost::thread(work);

  boost::this_thread::sleep_for(boost::chrono::milliseconds(2000));

  return 0;
}

但我得到了完全不同的结果。当我尝试连接到不存在的主机时，大多数情况下是:

Trying 192.168.1.15:50000
work in

有时是:

Trying 192.168.1.15:50000
work in
Connect error: Operation canceled
Connect error: Operation canceled

而且很少是:

Trying 192.168.1.15:50000
work in
Segmentation fault

“work out”永远不会被打印，所以我猜测这个例子中的 io_service 正在做一些事情，但这与以前的代码有什么不同，为什么我有时会收到“操作取消”错误？

Answer 1

在后台线程中运行的客户端应该看起来像这样。

请注意，我已记录了诸如连接超时之类的内容。为此，您希望有一个与 async_connect 并行运行的截止时间计时器。然后，您必须正确处理交叉情况(提示:在成功连接时取消截止计时器，并从 async_wait 中丢弃随之而来的错误)。

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/chrono.hpp>
#include <thread>
#include <functional>

boost::asio::io_service io_svc;

struct client
    : std::enable_shared_from_this<client>
{
    using protocol = boost::asio::ip::tcp;
    using resolver = protocol::resolver;
    using socket = protocol::socket;

    using error_code = boost::system::error_code;

    client(boost::asio::io_service& ios)
        : ios_(ios) {}

    void start(std::string const& host, std::string const& service)
    {
        auto presolver = std::make_shared<resolver>(get_io_service());

        presolver->async_resolve(protocol::resolver::query(host, service),
                                 strand_.wrap([self = shared_from_this(), presolver](auto&& ec, auto iter)
                                              {
                                                  self->handle_resolve(ec, presolver, iter);
                                              }));

    }

private:
    void
    handle_resolve(boost::system::error_code const& ec, std::shared_ptr<resolver> presolver, resolver::iterator iter)
    {
        if (ec) {
            std::cerr << "error resolving: " << ec.message() << std::endl;
        }
        else {
            boost::asio::async_connect(sock, iter, strand_.wrap([self = shared_from_this(),
                                                                    presolver]
                                                                    (auto&& ec, auto iter)
                                                                {
                                                                    self->handle_connect(ec, iter);
                                                                    // note - we're dropping presolver here - we don't need it any more
                                                                }));
        }
    }

    void handle_connect(error_code const& ec, resolver::iterator iter)
    {
        if (ec) {
            std::cerr << "failed to connect: " << ec.message() << std::endl;
        }
        else {
            auto payload = std::make_shared<std::string>("Hello");

            boost::asio::async_write(sock, boost::asio::buffer(*payload),
                                     strand_.wrap([self = shared_from_this(),
                                                      payload] // note! capture the payload so it continues to exist during async send
                                                      (auto&& ec, auto size)
                                                  {
                                                      self->handle_send(ec, size);
                                                  }));
        }
    }

    void handle_send(error_code const& ec, std::size_t size)
    {
        if (ec) {
            std::cerr << "send failed after " << size << " butes : " << ec.message() << std::endl;
        }
        else {
            // send something else?
        }
    }

    boost::asio::io_service& get_io_service()
    {
        return ios_;
    }

private:

    boost::asio::io_service& ios_;
    boost::asio::strand strand_{get_io_service()};
    socket              sock{get_io_service()};

};

void work()
{
    std::cout << "work in" << std::endl;
    io_svc.run();
    std::cout << "work out" << std::endl;
    return;
}

int main(int argc, char *argv[])
{

    auto        pclient = std::make_shared<client>(io_svc);
    std::string host    = "192.168.1.15", port = "50000";
    pclient->start(host, port);

    auto run_thread = std::thread(work);
    if (run_thread.joinable())
        run_thread.join();

    return 0;
}

示例输出:

work in
    <time passes>...
failed to connect: Operation timed out
work out