async_connect doesn't call handler in TCP client class

Question

I'm trying to make a client class from boost TCP client example for my projects, and I've noticed that sometimes handle_connect doesn't get called when connecting to nonexistent host.

I've read similar issues here on stack, where people forgot to run io_service or called it before any tasks were posted, but I don't think that's my case, since I launch io_service.run() thread right after calling async_connect, and successfull connect, network unreachable, and some other cases I've tested work just fine.

Here is the full listing:

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;
  }

};

};

and 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;
}

The output I get is when connecting to loopback server:

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>

The 192.168.1.20 works just as it should, as you see. The 192.168.1.15 doesnt'e exist, but I've expected it to throw some kind of error. Instead io_service.run() returns right away, like async_connect never posted callback task. Maybe it's related to endpoint iterator and not async_connect?

Can anyone please explain why is it happening like this?

Then I've tried to isolate the problem in this code:

#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;
}

But I've got a totally different result. When I try to connect to a nonexistent host, most of the time it's:

Trying 192.168.1.15:50000
work in

Sometimes it's:

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

And rarely it's:

Trying 192.168.1.15:50000
work in
Segmentation fault

"work out" is never printed, so I'm guessing io_service in this example is doing something, but how is this different from previous code, and why I get "operation canceled" error only sometimes?

Answer 1

A client running in a background thread should look something like this.

Note that I have note included things like connection timeouts. For that you'd want to have a deadline timer running in parallel with the async_connect. Then you'd have to correctly handle crossing cases (hint: cancel the deadline timer on successful connect and throw away the ensuing error from its 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;
}

example output:

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