I am attempting to make a fairly simple client-server program with boost asio. The server class is implemented as follows:
template<class RequestHandler, class RequestClass>
class Server {
public:
typedef std::map<std::string, RequestHandler> CommandMap;
Server(short port, CommandMap commands, RequestClass *request_class_inst)
: acceptor_(io_context_, tcp::endpoint(tcp::v4(), port))
, commands_(std::move(commands))
, request_class_inst_(request_class_inst)
{
DoAccept();
}
~Server()
{
}
void Run()
{
io_context_.run();
}
void RunInBackground()
{
std::thread t( [this]{ Run(); });
t.detach();
}
void Kill()
{
acceptor_.close();
}
private:
boost::asio::io_context io_context_;
tcp::acceptor acceptor_;
CommandMap commands_;
RequestClass *request_class_inst_;
void DoAccept()
{
acceptor_.async_accept(
[this](boost::system::error_code ec, tcp::socket socket) {
if (!ec)
std::make_shared<Session<RequestHandler, RequestClass>>
(std::move(socket), commands_, request_class_inst_)->Run();
DoAccept();
});
}
};
In addition to the server class, I implement a basic Client class thusly:
class Client {
public:
/**
* Constructor, initializes JSON parser and serializer.
*/
Client()
: reader_((new Json::CharReaderBuilder)->newCharReader())
{}
Json::Value MakeRequest(const std::string &ip_addr, unsigned short port,
const Json::Value &request)
{
boost::asio::io_context io_context;
std::string serialized_req = Json::writeString(writer_, request);
tcp::socket s(io_context);
tcp::resolver resolver(io_context);
s.connect({ boost::asio::ip::address::from_string(ip_addr), port });
boost::asio::write(s, boost::asio::buffer(serialized_req));
s.shutdown(tcp::socket::shutdown_send);
error_code ec;
char reply[2048];
size_t reply_length = boost::asio::read(s, boost::asio::buffer(reply),
ec);
std::cout << std::string(reply).substr(0, reply_length) << std::endl;
Json::Value json_resp;
JSONCPP_STRING parse_err;
std::string resp_str(reply);
if (reader_->parse(resp_str.c_str(), resp_str.c_str() + resp_str.length(),
&json_resp, &parse_err))
return json_resp;
throw std::runtime_error("Error parsing response.");
}
bool IsAlive(const std::string &ip_addr, unsigned short port)
{
boost::asio::io_context io_context;
tcp::socket s(io_context);
tcp::resolver resolver(io_context);
try {
s.connect({boost::asio::ip::address::from_string(ip_addr), port});
} catch(const boost::wrapexcept<boost::system::system_error> &err) {
s.close();
return false;
}
s.close();
return true;
}
private:
/// Reads JSON.
const std::unique_ptr<Json::CharReader> reader_;
/// Writes JSON.
Json::StreamWriterBuilder writer_;
};
I have implemented a small example to test Client::IsAlive:
int main()
{
auto *request_inst = new RequestClass(1);
std::map<std::string, RequestClassMethod> commands {
{"ADD_1", std::mem_fn(&RequestClass::add_n)},
{"SUB_1", std::mem_fn(&RequestClass::sub_n)}
};
Server<RequestClassMethod, RequestClass> s1(5000, commands, request_inst);
s1.RunInBackground();
std::vector<Client*> clients(6, new Client());
s1.Kill();
// Should output "0" to console.
std::cout << clients.at(1)->IsAlive("127.0.0.1", 5000);
return 0;
}
However, when I attempt to run this, the output varies. About half the time, I receive the correct value and the program exits with code 0, but, on other occasions, the program will either: (1) exit with code 139 (SEGFAULT) before outputting 0 to the console, (2) output 0 to the console and subsequently exit with code 139, (3) output 0 to the console and subsequently hang, or (4) hang before writing anything to the console.
I am uncertain as to what has caused these errors. I expect that it has to do with the destruction of Server::io_context_ and implementation of Server::Kill. Could this pertain to how I am storing Server::io_context_ as a data member?
A minimum reproducible example is shown below:
#define BOOST_ASIO_HAS_MOVE
#include <cstdlib>
#include <iostream>
#include <memory>
#include <utility>
#include <boost/asio.hpp>
#include <boost/system/error_code.hpp>
#include <json/json.h>
using boost::asio::ip::tcp;
using boost::system::error_code;
/// NOTE: This class exists exclusively for unit testing.
class RequestClass {
public:
/**
* Initialize class with value n to add sub from input values.
*
* @param n Value to add/sub from input values.
*/
explicit RequestClass(int n) : n_(n) {}
/// Value to add/sub from
int n_;
/**
* Add n to value in JSON request.
*
* @param request JSON request with field "value".
* @return JSON response containing modified field "value" = [original_value] + n.
*/
[[nodiscard]] Json::Value add_n(const Json::Value &request) const
{
Json::Value resp;
resp["SUCCESS"] = true;
// If value is present in request, return value + 1, else return error.
if (request.get("VALUE", NULL) != NULL) {
resp["VALUE"] = request["VALUE"].asInt() + this->n_;
} else {
resp["SUCCESS"] = false;
resp["ERRORS"] = "Invalid value.";
}
return resp;
}
/**
* Sun n from value in JSON request.
*
* @param request JSON request with field "value".
* @return JSON response containing modified field "value" = [original_value] - n.
*/
[[nodiscard]] Json::Value sub_n(const Json::Value &request) const
{
Json::Value resp, value;
resp["SUCCESS"] = true;
// If value is present in request, return value + 1, else return error.
if (request.get("VALUE", NULL) != NULL) {
resp["VALUE"] = request["VALUE"].asInt() - this->n_;
} else {
resp["SUCCESS"] = false;
resp["ERRORS"] = "Invalid value.";
}
return resp;
}
};
typedef std::function<Json::Value(RequestClass, const Json::Value &)> RequestClassMethod;
template<class RequestHandler, class RequestClass>
class Session :
public std::enable_shared_from_this<Session<RequestHandler,
RequestClass>>
{
public:
typedef std::map<std::string, RequestHandler> CommandMap;
Session(tcp::socket socket, CommandMap commands,
RequestClass *request_class_inst)
: socket_(std::move(socket))
, commands_(std::move(commands))
, request_class_inst_(request_class_inst)
, reader_((new Json::CharReaderBuilder)->newCharReader())
{}
void Run()
{
DoRead();
}
void Kill()
{
continue_ = false;
}
private:
tcp::socket socket_;
RequestClass *request_class_inst_;
CommandMap commands_;
/// Reads JSON.
const std::unique_ptr<Json::CharReader> reader_;
/// Writes JSON.
Json::StreamWriterBuilder writer_;
bool continue_ = true;
char data_[2048];
std::string resp_;
void DoRead()
{
auto self(this->shared_from_this());
socket_.async_read_some(boost::asio::buffer(data_),
[this, self](error_code ec, std::size_t length)
{
if (!ec)
DoWrite(length);
});
}
void DoWrite(std::size_t length)
{
JSONCPP_STRING parse_err;
Json::Value json_req, json_resp;
std::string client_req_str(data_);
if (reader_->parse(client_req_str.c_str(),
client_req_str.c_str() +
client_req_str.length(),
&json_req, &parse_err))
{
try {
// Get JSON response.
json_resp = ProcessRequest(json_req);
json_resp["SUCCESS"] = true;
} catch (const std::exception &ex) {
// If json parsing failed.
json_resp["SUCCESS"] = false;
json_resp["ERRORS"] = std::string(ex.what());
}
} else {
// If json parsing failed.
json_resp["SUCCESS"] = false;
json_resp["ERRORS"] = std::string(parse_err);
}
resp_ = Json::writeString(writer_, json_resp);
auto self(this->shared_from_this());
boost::asio::async_write(socket_,
boost::asio::buffer(resp_),
[this, self]
(boost::system::error_code ec,
std::size_t bytes_xfered) {
if (!ec) DoRead();
});
}
Json::Value ProcessRequest(Json::Value request)
{
Json::Value response;
std::string command = request["COMMAND"].asString();
// If command is not valid, give a response with an error.
if(commands_.find(command) == commands_.end()) {
response["SUCCESS"] = false;
response["ERRORS"] = "Invalid command.";
}
// Otherwise, run the relevant handler.
else {
RequestHandler handler = commands_.at(command);
response = handler(*request_class_inst_, request);
}
return response;
}
};
template<class RequestHandler, class RequestClass>
class Server {
public:
typedef std::map<std::string, RequestHandler> CommandMap;
Server(short port, CommandMap commands, RequestClass *request_class_inst)
: acceptor_(io_context_, tcp::endpoint(tcp::v4(), port))
, commands_(std::move(commands))
, request_class_inst_(request_class_inst)
{
DoAccept();
}
~Server()
{
}
void Run()
{
io_context_.run();
}
void RunInBackground()
{
std::thread t( [this]{ Run(); });
t.detach();
}
void Kill()
{
acceptor_.close();
}
private:
boost::asio::io_context io_context_;
tcp::acceptor acceptor_;
CommandMap commands_;
RequestClass *request_class_inst_;
void DoAccept()
{
acceptor_.async_accept(
[this](boost::system::error_code ec, tcp::socket socket) {
if (!ec)
std::make_shared<Session<RequestHandler, RequestClass>>
(std::move(socket), commands_, request_class_inst_)->Run();
DoAccept();
});
}
};
class Client {
public:
/**
* Constructor, initializes JSON parser and serializer.
*/
Client()
: reader_((new Json::CharReaderBuilder)->newCharReader())
{}
Json::Value MakeRequest(const std::string &ip_addr, unsigned short port,
const Json::Value &request)
{
boost::asio::io_context io_context;
std::string serialized_req = Json::writeString(writer_, request);
tcp::socket s(io_context);
tcp::resolver resolver(io_context);
s.connect({ boost::asio::ip::address::from_string(ip_addr), port });
boost::asio::write(s, boost::asio::buffer(serialized_req));
s.shutdown(tcp::socket::shutdown_send);
error_code ec;
char reply[2048];
size_t reply_length = boost::asio::read(s, boost::asio::buffer(reply),
ec);
std::cout << std::string(reply).substr(0, reply_length) << std::endl;
Json::Value json_resp;
JSONCPP_STRING parse_err;
std::string resp_str(reply);
if (reader_->parse(resp_str.c_str(), resp_str.c_str() + resp_str.length(),
&json_resp, &parse_err))
return json_resp;
throw std::runtime_error("Error parsing response.");
}
bool IsAlive(const std::string &ip_addr, unsigned short port)
{
boost::asio::io_context io_context;
tcp::socket s(io_context);
tcp::resolver resolver(io_context);
try {
s.connect({boost::asio::ip::address::from_string(ip_addr), port});
} catch(const boost::wrapexcept<boost::system::system_error> &err) {
s.close();
return false;
}
s.close();
return true;
}
private:
/// Reads JSON.
const std::unique_ptr<Json::CharReader> reader_;
/// Writes JSON.
Json::StreamWriterBuilder writer_;
};
int main()
{
auto *request_inst = new RequestClass(1);
std::map<std::string, RequestClassMethod> commands {
{"ADD_1", std::mem_fn(&RequestClass::add_n)},
{"SUB_1", std::mem_fn(&RequestClass::sub_n)}
};
Server<RequestClassMethod, RequestClass> s1(5000, commands, request_inst);
s1.RunInBackground();
std::vector<Client*> clients(6, new Client());
Json::Value sub_one_req;
sub_one_req["COMMAND"] = "SUB_1";
sub_one_req["VALUE"] = 1;
s1.Kill();
std::cout << clients.at(1)->IsAlive("127.0.0.1", 5000);
return 0;
}
Using ASAN (-fsanitize=addess) on that shows
false
=================================================================
==31232==ERROR: AddressSanitizer: heap-use-after-free on address 0x6110000002c0 at pc 0x561409ca2ea3 bp 0x7efcf
bbfdc60 sp 0x7efcfbbfdc50
READ of size 8 at 0x6110000002c0 thread T1
=================================================================
#0 0x561409ca2ea2 in boost::asio::detail::epoll_reactor::run(long, boost::asio::detail::op_queue<boost::asi
o::detail::scheduler_operation>&) /home/sehe/custom/boost_1_76_0/boost/asio/detail/impl/epoll_reactor.ipp:504
==31232==ERROR: LeakSanitizer: detected memory leaks
#1 0x561409cb442c in boost::asio::detail::scheduler::do_run_one(boost::asio::detail::conditionally_enabled_
mutex::scoped_lock&, boost::asio::detail::scheduler_thread_info&, boost::system::error_code const&) /home/sehe/
custom/boost_1_76_0/boost/asio/detail/impl/scheduler.ipp:470
Direct leak of 4 byte(s) in 1 object(s) allocated from:
#0 0x7efd08fca717 in operator new(unsigned long) (/usr/lib/x86_64-linux-gnu/libasan.so.6+0xb4717)
#2 0x561409cf2792 in boost::asio::detail::scheduler::run(boost::system::error_code&) /home/sehe/custom/boos
t_1_76_0/boost/asio/detail/impl/scheduler.ipp:204
#1 0x561409bc62b5 in main /home/sehe/Projects/stackoverflow/test.cpp:229
SUMMARY: AddressSanitizer: 4 byte(s) leaked in 1 allocation(s).
Or on another run:
It already tells you "everything" you need to know. Coincidentally, it was the bug I referred to in my previous answer. To do graceful shutdown you have to synchronize on the thread. Detaching it ruins your chances forever. So, let's not detach it:
void RunInBackground()
{
if (!t_.joinable()) {
t_ = std::thread([this] { Run(); });
}
}
As you can see,
thisis captured, so you can never allow the thread to run past the destruction of theServerobject.
And then in the destructor join it:
~Server()
{
if (t_.joinable()) {
t_.join();
}
}
Now, let's be thorough. We have two threads. They share objects. io_context is thread-safe, so that's fine. But tcp::acceptor is not. Neither might request_class_inst_. You need to synchronize more:
void Kill()
{
post(io_context_, [this] { acceptor_.close(); });
}
Now, note that this is NOT enough! .close() causes .cancel() on the acceptor, but that just makes the completion handler be invoked with error::operation_aborted. So, you need to prevent initiating DoAccept again in that case:
void DoAccept()
{
acceptor_.async_accept(
[this](boost::system::error_code ec, tcp::socket socket) {
if (ec) {
std::cout << "Accept loop: " << ec.message() << std::endl;
} else {
std::make_shared<Session<RequestHandler, RequestClass>>(
std::move(socket), commands_, request_class_inst_)
->Run();
DoAccept();
}
});
}
I took the liberty of aborting on /any/ error. Err on the safe side: you prefer processes to exit instead of being stuck in unresponsive state of high-CPU loops.
Regardless of this, you should be aware of the race condition between server startup/shutdown and your test client:
s1.RunInBackground();
// unspecified, race condition!
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(0).IsAlive("127.0.0.1", 5000) << std::endl;
sleep_for(10ms); // likely enough for acceptor to start
// true:
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(1).IsAlive("127.0.0.1", 5000) << std::endl;
std::cout << "MakeRequest: " << clients.at(2).MakeRequest(
"127.0.0.1", 5000, {{"COMMAND", "MUL_2"}, {"VALUE", "21"}})
<< std::endl;
s1.Kill();
// unspecified, race condition!
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(3).IsAlive("127.0.0.1", 5000) << std::endl;
sleep_for(10ms); // likely enough for acceptor to be closed
// false:
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(4).IsAlive("127.0.0.1", 5000) << std::endl;
Prints
IsAlive(240): true
IsAlive(245): true
MakeRequest: {"SUCCESS":false,"ERRORS":"not an int64"}
{"SUCCESS":false,"ERRORS":"not an int64"}
IsAlive(252): CLOSING
Accept loop: Operation canceled
THREAD EXIT
false
IsAlive(256): false
Note that this also fixed the unnecessary leak of the RequestClass instance. You were already assuming copy-ability (because you were passing it by value in various places).
Also note that in MakeRequest we now no longer swallow any errors except EOF.
Like last time, I employ Boost Json for simplicity and to make the sample self-contained for StackOverflow.
Address sanitizer (ASan) and UBSan are silent. Life is good.
#include <boost/asio.hpp>
#include <boost/json.hpp>
#include <boost/json/src.hpp>
#include <iostream>
#include <deque>
using boost::asio::ip::tcp;
using boost::system::error_code;
namespace json = boost::json;
using Value = json::object;
using namespace std::chrono_literals;
static auto sleep_for(auto delay) { return std::this_thread::sleep_for(delay); }
/// NOTE: This class exists exclusively for unit testing.
struct RequestClass {
int n_;
Value add_n(Value const& request) const { return impl(std::plus<>{}, request); }
Value sub_n(Value const& request) const { return impl(std::minus<>{}, request); }
Value mul_n(Value const& request) const { return impl(std::multiplies<>{}, request); }
Value div_n(Value const& request) const { return impl(std::divides<>{}, request); }
private:
template <typename Op> Value impl(Op op, Value const& req) const {
return (req.contains("VALUE"))
? Value{{"VALUE", op(req.at("VALUE").as_int64(), n_)},
{"SUCCESS", true}}
: Value{{"ERRORS", "Invalid value."}, {"SUCCESS", false}};
}
};
using RequestClassMethod =
std::function<Value(RequestClass const&, Value const&)>;
template <class RequestHandler, class RequestClass>
class Session
: public std::enable_shared_from_this<
Session<RequestHandler, RequestClass>> {
public:
using CommandMap = std::map<std::string, RequestHandler>;
Session(tcp::socket socket, CommandMap commands,
RequestClass request_class_inst)
: socket_(std::move(socket))
, commands_(std::move(commands))
, request_class_inst_(std::move(request_class_inst))
{
}
void Run() { DoRead(); }
void Kill() { continue_ = false; }
private:
tcp::socket socket_;
CommandMap commands_;
RequestClass request_class_inst_;
bool continue_ = true;
char data_[2048];
std::string resp_;
void DoRead()
{
socket_.async_read_some(
boost::asio::buffer(data_),
[this, self = this->shared_from_this()](error_code ec, std::size_t length) {
if (!ec) {
DoWrite(length);
}
});
}
void DoWrite(std::size_t length)
{
Value json_resp;
try {
auto json_req = json::parse({data_, length}).as_object();
json_resp = ProcessRequest(json_req);
json_resp["SUCCESS"] = true;
} catch (std::exception const& ex) {
json_resp = {{"SUCCESS", false}, {"ERRORS", ex.what()}};
}
resp_ = json::serialize(json_resp);
boost::asio::async_write(socket_, boost::asio::buffer(resp_),
[this, self = this->shared_from_this()](
error_code ec, size_t bytes_xfered) {
if (!ec)
DoRead();
});
}
Value ProcessRequest(Value request)
{
auto command = request.contains("COMMAND")
? request["COMMAND"].as_string() //
: "";
std::string cmdstr(command.data(), command.size());
// If command is not valid, give a response with an error.
return commands_.contains(cmdstr)
? commands_.at(cmdstr)(request_class_inst_, request)
: Value{{"SUCCESS", false}, {"ERRORS", "Invalid command."}};
}
};
template <class RequestHandler, class RequestClass> class Server {
public:
using CommandMap = std::map<std::string, RequestHandler>;
Server(uint16_t port, CommandMap commands, RequestClass request_class_inst)
: acceptor_(io_context_, tcp::endpoint(tcp::v4(), port))
, commands_(std::move(commands))
, request_class_inst_(std::move(request_class_inst))
{
DoAccept();
}
~Server()
{
if (t_.joinable()) {
t_.join();
}
assert(not t_.joinable());
}
void Run()
{
io_context_.run();
}
void RunInBackground()
{
if (!t_.joinable()) {
t_ = std::thread([this] {
Run();
std::cout << "THREAD EXIT" << std::endl;
});
}
}
void Kill()
{
post(io_context_, [this] {
std::cout << "CLOSING" << std::endl;
acceptor_.close(); // causes .cancel() as well
});
}
private:
boost::asio::io_context io_context_;
tcp::acceptor acceptor_;
CommandMap commands_;
RequestClass request_class_inst_;
std::thread t_;
void DoAccept()
{
acceptor_.async_accept(
[this](boost::system::error_code ec, tcp::socket socket) {
if (ec) {
std::cout << "Accept loop: " << ec.message() << std::endl;
} else {
std::make_shared<Session<RequestHandler, RequestClass>>(
std::move(socket), commands_, request_class_inst_)
->Run();
DoAccept();
}
});
}
};
class Client {
public:
/**
* Constructor, initializes JSON parser and serializer.
*/
Client() {}
Value MakeRequest(std::string const& ip_addr, uint16_t port,
Value const& request)
{
boost::asio::io_context io_context;
std::string serialized_req = serialize(request);
tcp::socket s(io_context);
s.connect({boost::asio::ip::address::from_string(ip_addr), port});
boost::asio::write(s, boost::asio::buffer(serialized_req));
s.shutdown(tcp::socket::shutdown_send);
char reply[2048];
error_code ec;
size_t reply_length = read(s, boost::asio::buffer(reply), ec);
if (ec && ec != boost::asio::error::eof) {
throw boost::system::system_error(ec);
}
// safe method:
std::string_view resp_str(reply, reply_length);
Value res = json::parse({reply, reply_length}).as_object();
std::cout << res << std::endl;
return res;
}
bool IsAlive(std::string const& ip_addr, unsigned short port)
{
boost::asio::io_context io_context;
tcp::socket s(io_context);
error_code ec;
s.connect({boost::asio::ip::address::from_string(ip_addr), port}, ec);
return not ec.failed();
}
};
int main()
{
std::cout << std::boolalpha;
std::deque<Client> clients(6);
Server<RequestClassMethod, RequestClass> s1(
5000,
{
{"ADD_2", std::mem_fn(&RequestClass::add_n)},
{"SUB_2", std::mem_fn(&RequestClass::sub_n)},
{"MUL_2", std::mem_fn(&RequestClass::mul_n)},
{"DIV_2", std::mem_fn(&RequestClass::div_n)},
},
RequestClass{1});
s1.RunInBackground();
// unspecified, race condition!
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(0).IsAlive("127.0.0.1", 5000) << std::endl;
sleep_for(10ms); // likely enough for acceptor to start
// true:
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(1).IsAlive("127.0.0.1", 5000) << std::endl;
std::cout << "MakeRequest: " << clients.at(2).MakeRequest(
"127.0.0.1", 5000, {{"COMMAND", "MUL_2"}, {"VALUE", "21"}})
<< std::endl;
s1.Kill();
// unspecified, race condition!
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(3).IsAlive("127.0.0.1", 5000) << std::endl;
sleep_for(10ms); // likely enough for acceptor to be closed
// false:
std::cout << "IsAlive(" << __LINE__ << "): " << clients.at(4).IsAlive("127.0.0.1", 5000) << std::endl;
}