Being a total beginner to Boost.Asio, I am confused with io_service::run()
. I would appreciate it if someone could explain to me when this method blocks/unblocks. The documentations states:
The
run()
function blocks until all work has finished and there are no more handlers to be dispatched, or until theio_service
has been stopped.Multiple threads may call the
run()
function to set up a pool of threads from which theio_service
may execute handlers. All threads that are waiting in the pool are equivalent and theio_service
may choose any one of them to invoke a handler.A normal exit from the
run()
function implies that theio_service
object is stopped (thestopped()
function returns true). Subsequent calls torun()
,run_one()
,poll()
orpoll_one()
will return immediately unless there is a prior call toreset()
.
What does the following statement mean?
[...] no more handlers to be dispatched [...]
While trying to understand the behavior of io_service::run()
, I came across this example (example 3a). Within it, I observe that io_service->run()
blocks and waits for work orders.
// WorkerThread invines io_service->run()
void WorkerThread(boost::shared_ptr<boost::asio::io_service> io_service);
void CalculateFib(size_t);
boost::shared_ptr<boost::asio::io_service> io_service(
new boost::asio::io_service);
boost::shared_ptr<boost::asio::io_service::work> work(
new boost::asio::io_service::work(*io_service));
// ...
boost::thread_group worker_threads;
for(int x = 0; x < 2; ++x)
{
worker_threads.create_thread(boost::bind(&WorkerThread, io_service));
}
io_service->post( boost::bind(CalculateFib, 3));
io_service->post( boost::bind(CalculateFib, 4));
io_service->post( boost::bind(CalculateFib, 5));
work.reset();
worker_threads.join_all();
However, in the following code that I was working on, the client connects using TCP/IP and the run method blocks until data is asynchronously received.
typedef boost::asio::ip::tcp tcp;
boost::shared_ptr<boost::asio::io_service> io_service(
new boost::asio::io_service);
boost::shared_ptr<tcp::socket> socket(new tcp::socket(*io_service));
// Connect to 127.0.0.1:9100.
tcp::resolver resolver(*io_service);
tcp::resolver::query query("127.0.0.1",
boost::lexical_cast< std::string >(9100));
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
socket->connect(endpoint_iterator->endpoint());
// Just blocks here until a message is received.
socket->async_receive(boost::asio::buffer(buf_client, 3000), 0,
ClientReceiveEvent);
io_service->run();
// Write response.
boost::system::error_code ignored_error;
std::cout << "Sending message \n";
boost::asio::write(*socket, boost::asio::buffer("some data"), ignored_error);
Any explanation of run()
that describes its behavior in the two examples below would be appreciated.
Lets start with a simplified example and examine the relevant Boost.Asio pieces:
void handle_async_receive(...) { ... }
void print() { ... }
...
boost::asio::io_service io_service;
boost::asio::ip::tcp::socket socket(io_service);
...
io_service.post(&print); // 1
socket.connect(endpoint); // 2
socket.async_receive(buffer, &handle_async_receive); // 3
io_service.post(&print); // 4
io_service.run(); // 5
A handler is nothing more than a callback. In the example code, there are 3 handlers:
print
handler (1).handle_async_receive
handler (3).print
handler (4).Even though the same print()
function is used twice, each use is considered to create its own uniquely identifiable handler. Handlers can come in many shapes and sizes, ranging from basic functions like the ones above to more complex constructs such as functors generated from boost::bind()
and lambdas. Regardless of the complexity, the handler still remains nothing more than a callback.
Work is some processing that Boost.Asio has been requested to do on behalf of the application code. Sometimes Boost.Asio may start some of the work as soon as it has been told about it, and other times it may wait to do the work at a later point in time. Once it has finished the work, Boost.Asio will inform the application by invoking the supplied handler.
Boost.Asio guarantees that handlers will only run within a thread that is currently calling run()
, run_one()
, poll()
, or poll_one()
. These are the threads that will do work and call handlers. Therefore, in above example, print()
is not invoked when it is posted into the io_service
(1). Instead, it is added to the io_service
and will be invoked at a later point in time. In this case, it within io_service.run()
(5).
An asynchronous operation creates work and Boost.Asio will invoke a handler to inform the application when the work has completed. Asynchronous operations are created by calling a function that has a name with the prefix async_
. These functions are also known as initiating functions.
Asynchronous operations can be decomposed into three unique steps:
io_service
that works needs to be done. The async_receive
operation (3) informs the io_service
that it will need to asynchronously read data from the socket, then async_receive
returns immediately.socket
receives data, bytes will be read and copied into buffer
. The actual work will be done in either:
io_service
(5).handle_async_receive
ReadHandler. Once again, handlers are only invoked within threads running the io_service
. Thus, regardless of when the work is done (3 or 5), it is guaranteed that handle_async_receive()
will only be invoked within io_service.run()
(5).The separation in time and space between these three steps is known as control flow inversion. It is one of the complexities that makes asynchronous programming difficult. However, there are techniques that can help mitigate this, such as by using coroutines.
io_service.run()
Do?When a thread calls io_service.run()
, work and handlers will be invoked from within this thread. In the above example, io_service.run()
(5) will block until either:
print
handlers, the receive operation completes with success or failure, and its handle_async_receive
handler has been invoked and returned.io_service
is explicitly stopped via io_service::stop()
.One potential psuedo-ish flow could be described as the following:
create io_service create socket add print handler to io_service (1) wait for socket to connect (2) add an asynchronous read work request to the io_service (3) add print handler to io_service (4) run the io_service (5) is there work or handlers? yes, there is 1 work and 2 handlers does socket have data? no, do nothing run print handler (1) is there work or handlers? yes, there is 1 work and 1 handler does socket have data? no, do nothing run print handler (4) is there work or handlers? yes, there is 1 work does socket have data? no, continue waiting -- socket receives data -- socket has data, read it into buffer add handle_async_receive handler to io_service is there work or handlers? yes, there is 1 handler run handle_async_receive handler (3) is there work or handlers? no, set io_service as stopped and return
Notice how when the read finished, it added another handler to the io_service
. This subtle detail is an important feature of asynchronous programming. It allows for handlers to be chained together. For instance, if handle_async_receive
did not get all the data it expected, then its implementation could post another asynchronous read operation, resulting in io_service
having more work, and thus not returning from io_service.run()
.
Do note that when the io_service
has ran out of work, the application must reset()
the io_service
before running it again.
Now, lets examine the two pieces of code referenced in the question.
socket->async_receive
adds work to the io_service
. Thus, io_service->run()
will block until the read operation completes with success or error, and ClientReceiveEvent
has either finished running or throws an exception.
In hopes of making it easier to understand, here is a smaller annotated Example 3a:
void CalculateFib(std::size_t n);
int main()
{
boost::asio::io_service io_service;
boost::optional<boost::asio::io_service::work> work = // '. 1
boost::in_place(boost::ref(io_service)); // .'
boost::thread_group worker_threads; // -.
for(int x = 0; x < 2; ++x) // :
{ // '.
worker_threads.create_thread( // :- 2
boost::bind(&boost::asio::io_service::run, &io_service) // .'
); // :
} // -'
io_service.post(boost::bind(CalculateFib, 3)); // '.
io_service.post(boost::bind(CalculateFib, 4)); // :- 3
io_service.post(boost::bind(CalculateFib, 5)); // .'
work = boost::none; // 4
worker_threads.join_all(); // 5
}
At a high-level, the program will create 2 threads that will process the io_service
's event loop (2). This results in a simple thread pool that will calculate Fibonacci numbers (3).
The one major difference between the Question Code and this code is that this code invokes io_service::run()
(2) before actual work and handlers are added to the io_service
(3). To prevent the io_service::run()
from returning immediately, an io_service::work
object is created (1). This object prevents the io_service
from running out of work; therefore, io_service::run()
will not return as a result of no work.
The overall flow is as follows:
io_service::work
object added to the io_service
.io_service::run()
. These worker threads will not return from io_service
because of the io_service::work
object.io_service
, and return immediately. The worker threads, not the main thread, may start running these handlers immediately.io_service::work
object.io_service
neither has handlers nor work.The code could be written differently, in the same manner as the Original Code, where handlers are added to the io_service
, and then the io_service
event loop is processed. This removes the need to use io_service::work
, and results in the following code:
int main()
{
boost::asio::io_service io_service;
io_service.post(boost::bind(CalculateFib, 3)); // '.
io_service.post(boost::bind(CalculateFib, 4)); // :- 3
io_service.post(boost::bind(CalculateFib, 5)); // .'
boost::thread_group worker_threads; // -.
for(int x = 0; x < 2; ++x) // :
{ // '.
worker_threads.create_thread( // :- 2
boost::bind(&boost::asio::io_service::run, &io_service) // .'
); // :
} // -'
worker_threads.join_all(); // 5
}
Although the code in the question is using an asynchronous operation, it is effectively functioning synchronously, as it is waiting for the asynchronous operation to complete:
socket.async_receive(buffer, handler)
io_service.run();
is equivalent to:
boost::asio::error_code error;
std::size_t bytes_transferred = socket.receive(buffer, 0, error);
handler(error, bytes_transferred);
As a general rule of thumb, try to avoid mixing synchronous and asynchronous operations. Often times, it can turn a complex system into a complicated system. This answer highlights advantages of asynchronous programming, some of which are also covered in the Boost.Asio documentation.