Having a single end point and receive buffer shared between the threads implies that asio waits for a handler to complete within a single thread
If you mean "when running the service with a a single thread" then this is correct.
Otherwise, this isn't the case. Instead Asio just says behaviour is "undefined" when you call operations on a single service object (i.e. the socket, not the io_service) concurrently.
That seems to negate the point of allowing multiple threads to call run in the first place.
Not unless processing takes a considerable amount of time.
The first paragraphs of the introduction of the Timer.5 sample seem like a good exposition about your topic.
Session
To separate the request-specific data (buffer and endpoint) you want some notion of a session. A popular mechanism in Asio is either bound shared_ptrs or a shared-from-this session class (boost bind supports binding to boost::shared_ptr instances directly).
Strand
To avoid concurrent, unsynchronized access to members of m_socket you can either add locks or use the strand approach as documented in the Timer.5 sample linked above.
Demo
Here for your enjoyment is the Daytime.6 asynchronous UDP daytime server, modified to work with many service IO threads.
Note that, logically, there's still only a single IO thread (the strand) so we don't violate the socket class's documented thread-safety.
However, unlike the official sample, the responses may get queued out of order, depending on the time taken by the actual processing in udp_session::handle_request.
Note the
- a
udp_session class to hold the buffers and remote endpoint per request
- a pool of threads, which are able to scale the load of actual processing (not the IO) over multiple cores.
#include <ctime>
#include <iostream>
#include <string>
#include <boost/array.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/make_shared.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
using namespace boost;
using asio::ip::udp;
using system::error_code;
std::string make_daytime_string()
{
using namespace std; // For time_t, time and ctime;
time_t now = time(0);
return ctime(&now);
}
class udp_server; // forward declaration
struct udp_session : enable_shared_from_this<udp_session> {
udp_session(udp_server* server) : server_(server) {}
void handle_request(const error_code& error);
void handle_sent(const error_code& ec, std::size_t) {
// here response has been sent
if (ec) {
std::cout << "Error sending response to " << remote_endpoint_ << ": " << ec.message() << "\n";
}
}
udp::endpoint remote_endpoint_;
array<char, 100> recv_buffer_;
std::string message;
udp_server* server_;
};
class udp_server
{
typedef shared_ptr<udp_session> shared_session;
public:
udp_server(asio::io_service& io_service)
: socket_(io_service, udp::endpoint(udp::v4(), 1313)),
strand_(io_service)
{
receive_session();
}
private:
void receive_session()
{
// our session to hold the buffer + endpoint
auto session = make_shared<udp_session>(this);
socket_.async_receive_from(
asio::buffer(session->recv_buffer_),
session->remote_endpoint_,
strand_.wrap(
bind(&udp_server::handle_receive, this,
session, // keep-alive of buffer/endpoint
asio::placeholders::error,
asio::placeholders::bytes_transferred)));
}
void handle_receive(shared_session session, const error_code& ec, std::size_t /*bytes_transferred*/) {
// now, handle the current session on any available pool thread
socket_.get_io_service().post(bind(&udp_session::handle_request, session, ec));
// immediately accept new datagrams
receive_session();
}
void enqueue_response(shared_session const& session) {
socket_.async_send_to(asio::buffer(session->message), session->remote_endpoint_,
strand_.wrap(bind(&udp_session::handle_sent,
session, // keep-alive of buffer/endpoint
asio::placeholders::error,
asio::placeholders::bytes_transferred)));
}
udp::socket socket_;
asio::strand strand_;
friend struct udp_session;
};
void udp_session::handle_request(const error_code& error)
{
if (!error || error == asio::error::message_size)
{
message = make_daytime_string(); // let's assume this might be slow
// let the server coordinate actual IO
server_->enqueue_response(shared_from_this());
}
}
int main()
{
try {
asio::io_service io_service;
udp_server server(io_service);
thread_group group;
for (unsigned i = 0; i < thread::hardware_concurrency(); ++i)
group.create_thread(bind(&asio::io_service::run, ref(io_service)));
group.join_all();
}
catch (std::exception& e) {
std::cerr << e.what() << std::endl;
}
}
Closing thoughts
Interestingly, in most cases you'll see the single-thread version performing just as well, and there's no reason to complicate the design.
Alternatively, you can use a single-threaded io_service dedicated to the IO and use an old fashioned worker pool to do the background processing of the requests if this is indeed the CPU intensive part. Firstly, this simplifies the design, secondly this might improve the throughput on the IO tasks because there is no more need to coordinate the tasks posted on the strand.
post(...)lines does that). – Heavyio_servicecould be a simple option) – Heavy