[Boost-users] Asio & strand & threads

From: Zola (looseonthestreet_at_[hidden])
Date: 2006-07-19 13:46:39

Hi Boosters,

I'm trying to figure out how Asio's strand works.
My concept of strand is a bit blurry.

Asio's Timer.5 tutorial says:
"An boost::asio::strand<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00260.html#9b084e7ac93c5dd1264b5ee9d3c2afe1>guarantees
that, for those handlers that are dispatched through it, an
executing handler will be allowed to complete before the next one is
started. This is guaranteed irrespective of the number of threads that are
calling boost::asio::io_service::run()<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html#4a0b19807d4f44e1e6f3c6667d98476c>.
Of course, the handlers may still execute concurrently with other handlers
that were *not* dispatched through an
boost::asio::strand<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00260.html#9b084e7ac93c5dd1264b5ee9d3c2afe1>,
or were dispatched through a different
boost::asio::strand<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00260.html#9b084e7ac93c5dd1264b5ee9d3c2afe1>object."

Asio's documentation for boost::asio::io_service::run says:
"Run the io_service's event processing loop.
The run()<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html#4a0b19807d4f44e1e6f3c6667d98476c>function
blocks until all work has finished and there are no more handlers
to be dispatched, or until the
io_service<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html>has
been interrupted.
Multiple threads may call the
run()<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html#4a0b19807d4f44e1e6f3c6667d98476c>function
to set up a pool of threads from which the
io_service<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html>may
execute handlers."

The key here is "Multiple threads may call the
run()<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html#4a0b19807d4f44e1e6f3c6667d98476c>function
to set up a pool of threads from which the
io_service<http://asio.sourceforge.net/boost_asio_0_3_7/libs/asio/doc/reference/a00139.html>may
execute handlers." --
This implies to me that it's automatic, and the thread pool will take
whatever work it can.

I've created a little example which has ten threads calling io_service::run,
and printing out which thread is calling a handler given to the deadline
timer.
The issue here is that when I run this Test, the same thread is calling
print() every time, no matter what I do.

My understanding of this Test is as so:
io_service holds a queue of asynchronous events.
You fill up the io service with events, and then you call run on it.
If multiple threads call run, they'll do whatever work they can.
I'm calling run from 10 threads:
Each thread will see what events are there, and handle them if possible.
The asynchronous waits are calling a strand wrapper.
The strand wrapper is like a mutex, and ensures that no two threads will
call print() at the same time.

What is going on, and why is the same thread calling print() every time?
Is my understanding here completely wrong?

- Zola

// The Test:

#include <iostream>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <vector>
namespace asio = boost::asio;

int const num_threads = 10;
std::vector<boost::thread*> tv(num_threads);

class Test
{
  public:

    Test(asio::io_service& io, int n, int ms)
    :
        n(n),
        ms(ms),
        s(io),
        dt(io, boost::posix_time::milliseconds(ms)),
        f(boost::bind(&Test::print, this))
    {
        dt.async_wait(s.wrap(f));
    }

    void print()
    {
        if (n >= 0)
        {
            boost::thread t;
            for (int i=0; i<num_threads; ++i)
            {
                if (tv[i] && (*tv[i] == t))
                {
                    std::cout << "(thread=" << i << ") ";
                    break;
                }
            }

            std::cout << "Timer: " << n << std::endl;
            --n;

            dt.expires_at(dt.expires_at() +
boost::posix_time::milliseconds(ms));
            dt.async_wait(s.wrap(f));
        }
    }

  private:

    int n;
    unsigned int ms;
    asio::strand s;
    asio::deadline_timer dt;
    boost::function<void (asio::error&)> f;
};

int main()
{
    asio::io_service io;
    Test t(io, 1000, 5); // 1000 iterations, every 5 ms.

    boost::thread_group threads;
    for (int i=0; i<num_threads; ++i)
    {
        tv[i] = threads.create_thread(boost::bind(&asio::io_service::run,
&io));
    }
    threads.join_all();
}