Async operations run on the execution context.
The original code runs the execution context:
c.start(r.resolve(argv[1], argv[2]));
io_context.run();
You stop() it instead:
io_context.post([&]() { c->start(argv[1], argv[2]) });
...
io_context.stop();
work_guard_.reset();
worker_result_.wait();
The simplest thing you can do is remove the stop(), or put it at the end (where it will be redundant because io_context.run() by definition already returned when worker_result_.wait() returns.
Demo: Live On Coliru
#include <boost/asio.hpp>
#include <future>
#include <iostream>
using boost::asio::steady_timer;
using boost::asio::ip::tcp;
using std::placeholders::_1;
using std::placeholders::_2;
//
// This class manages socket timeouts by applying the concept of a deadline.
// Some asynchronous operations are given deadlines by which they must complete.
// Deadlines are enforced by an "actor" that persists for the lifetime of the
// client object:
//
// +----------------+
// | |
// | check_deadline |<---+
// | | |
// +----------------+ | async_wait()
// | |
// +---------+
//
// If the deadline actor determines that the deadline has expired, the socket
// is closed and any outstanding operations are consequently cancelled.
//
// Connection establishment involves trying each endpoint in turn until a
// connection is successful, or the available endpoints are exhausted. If the
// deadline actor closes the socket, the connect actor is woken up and moves to
// the next endpoint.
//
// +---------------+
// | |
// | start_connect |<---+
// | | |
// +---------------+ |
// | |
// async_- | +----------------+
// connect() | | |
// +--->| handle_connect |
// | |
// +----------------+
// :
// Once a connection is :
// made, the connect :
// actor forks in two - :
// :
// an actor for reading : and an actor for
// inbound messages: : sending heartbeats:
// :
// +------------+ : +-------------+
// | |<- - - - -+- - - - ->| |
// | start_read | | start_write |<---+
// | |<---+ | | |
// +------------+ | +-------------+ | async_wait()
// | | | |
// async_- | +-------------+ async_- | +--------------+
// read_- | | | write() | | |
// until() +--->| handle_read | +--->| handle_write |
// | | | |
// +-------------+ +--------------+
//
// The input actor reads messages from the socket, where messages are delimited
// by the newline character. The deadline for a complete message is 30 seconds.
//
// The heartbeat actor sends a heartbeat (a message that consists of a single
// newline character) every 10 seconds. In this example, no deadline is applied
// to message sending.
//
namespace test_async {
class tcp_client {
public:
tcp_client(boost::asio::io_context& io_context)
: socket_(io_context)
, deadline_(io_context)
, heartbeat_timer_(io_context) {}
// Called by the user of the client class to initiate the connection
// process. The endpoints will have been obtained using a tcp::resolver.
void start(std::string host, std::string service) {
std::cerr << __PRETTY_FUNCTION__ << std::endl;
tcp::resolver r(socket_.get_executor());
endpoints_ = r.resolve(host, service);
// Start the connect actor.
start_connect(endpoints_.begin());
// Start the deadline actor. You will note that we're not setting
// any particular deadline here. Instead, the connect and input
// actors will update the deadline prior to each asynchronous
// operation.
deadline_.async_wait(std::bind(&tcp_client::check_deadline, this));
}
// This function terminates all the actors to shut down the connection.
// It may be called by the user of the client class, or by the class
// itself in response to graceful termination or an unrecoverable error.
void stop() {
stopped_ = true;
boost::system::error_code ignored_error;
socket_.close(ignored_error);
deadline_.cancel();
heartbeat_timer_.cancel();
}
private:
void
start_connect(tcp::resolver::results_type::iterator endpoint_iter) {
if (endpoint_iter != endpoints_.end()) {
std::cout << "Trying " << endpoint_iter->endpoint() << "...\n";
// Set a deadline for the connect operation.
deadline_.expires_after(std::chrono::seconds(60));
// Start the asynchronous connect operation.
socket_.async_connect(endpoint_iter->endpoint(),
std::bind(&tcp_client::handle_connect,
this, _1, endpoint_iter));
} else {
// There are no more endpoints to try. Shut down the client.
stop();
}
}
void
handle_connect(const boost::system::error_code& error,
tcp::resolver::results_type::iterator endpoint_iter) {
if (stopped_)
return;
// The async_connect() function automatically opens the socket at
// the start of the asynchronous operation. If the socket is closed
// at this time then the timeout handler must have run first.
if (!socket_.is_open()) {
std::cout << "Connect timed out\n";
// Try the next available endpoint.
start_connect(++endpoint_iter);
}
// Check if the connect operation failed before the deadline
// expired.
else if (error) {
std::cout << "Connect error: " << error.message() << "\n";
// We need to close the socket used in the previous connection
// attempt before starting a new one.
socket_.close();
// Try the next available endpoint.
start_connect(++endpoint_iter);
}
// Otherwise we have successfully established a connection.
else {
std::cout << "Connected to " << endpoint_iter->endpoint()
<< "\n";
// Start the input actor.
start_read();
// Start the heartbeat actor.
start_write();
}
}
void start_read() {
// Set a deadline for the read operation.
deadline_.expires_after(std::chrono::seconds(30));
// Start an asynchronous operation to read a newline-delimited
// message.
boost::asio::async_read_until(
socket_, boost::asio::dynamic_buffer(input_buffer_), '\n',
std::bind(&tcp_client::handle_read, this, _1, _2));
}
void handle_read(const boost::system::error_code& error,
std::size_t n) {
if (stopped_)
return;
if (!error) {
// Extract the newline-delimited message from the buffer.
std::string line(input_buffer_.substr(0, n - 1));
input_buffer_.erase(0, n);
// Empty messages are heartbeats and so ignored.
if (!line.empty()) {
std::cout << "Received: " << line << "\n";
}
start_read();
} else {
std::cout << "Error on receive: " << error.message() << "\n";
stop();
}
}
void start_write() {
if (stopped_)
return;
// Start an asynchronous operation to send a heartbeat message.
boost::asio::async_write(
socket_, boost::asio::buffer("\n", 1),
std::bind(&tcp_client::handle_write, this, _1));
}
void handle_write(const boost::system::error_code& error) {
if (stopped_)
return;
if (!error) {
// Wait 10 seconds before sending the next heartbeat.
heartbeat_timer_.expires_after(std::chrono::seconds(10));
heartbeat_timer_.async_wait(
std::bind(&tcp_client::start_write, this));
} else {
std::cout << "Error on heartbeat: " << error.message() << "\n";
stop();
}
}
void check_deadline() {
if (stopped_)
return;
// Check whether the deadline has passed. We compare the deadline
// against the current time since a new asynchronous operation may
// have moved the deadline before this actor had a chance to run.
if (deadline_.expiry() <= steady_timer::clock_type::now()) {
// The deadline has passed. The socket is closed so that any
// outstanding asynchronous operations are cancelled.
socket_.close();
// There is no longer an active deadline. The expiry is set to
// the maximum time point so that the actor takes no action
// until a new deadline is set.
deadline_.expires_at(steady_timer::time_point::max());
}
// Put the actor back to sleep.
deadline_.async_wait(std::bind(&tcp_client::check_deadline, this));
}
private:
bool stopped_ = false;
tcp::resolver::results_type endpoints_;
tcp::socket socket_;
std::string input_buffer_;
steady_timer deadline_;
steady_timer heartbeat_timer_;
};
}
int main(int argc, char* argv[])
{
try
{
if (argc != 3)
{
std::cerr << "Usage: client <host> <port>\n";
return 1;
}
#if OFFICIAL_BOOST_DOCUMENTIATION
boost::asio::io_context io_context;
test_async::tcp_client c(io_context);
c.start(argv[1], argv[2]);
io_context.run();
#else
boost::asio::io_context io_context;
auto c = std::make_unique<test_async::tcp_client>(io_context);
using work_guard_type = boost::asio::executor_work_guard<
boost::asio::io_context::executor_type>;
auto work_guard_ =
std::make_unique<work_guard_type>(io_context.get_executor());
auto worker_result_ = std::async(std::launch::async, [&]() {
std::cerr << "Runner thread enter" << std::endl;
io_context.run();
std::cerr << "Runner thread exit" << std::endl;
});
io_context.post([&]() { c->start(argv[1], argv[2]); });
//...
work_guard_.reset();
worker_result_.wait();
io_context.stop();
#endif
} catch (std::exception& e) {
std::cerr << "Exception: " << e.what() << "\n";
}
return 0;
}
Prints
g++ -std=c++20 -O2 -Wall -pedantic -pthread main.cpp
(for a in {1..10}; do sleep 1; date; done) | nc -tlkp 4444 -w10 &
(sleep .5; ./a.out 127.0.0.1 4444)&
wait %1
kill %2
Runner thread enter
void test_async::tcp_client::start(std::string, std::string)
Trying 127.0.0.1:4444...
Connected to 127.0.0.1:4444
Received: Wed Sep 21 21:18:55 UTC 2022
Received: Wed Sep 21 21:18:56 UTC 2022
Received: Wed Sep 21 21:18:57 UTC 2022
Received: Wed Sep 21 21:18:58 UTC 2022
Received: Wed Sep 21 21:18:59 UTC 2022
Received: Wed Sep 21 21:19:00 UTC 2022
Received: Wed Sep 21 21:19:01 UTC 2022
Received: Wed Sep 21 21:19:02 UTC 2022
Received: Wed Sep 21 21:19:03 UTC 2022
Received: Wed Sep 21 21:19:04 UTC 2022
Error on receive: End of file
Runner thread exit
