Reputation: 3
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?
Upvotes: 0
Views: 1385
Reputation: 69892
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
Upvotes: 1