On Fri, Jun 17, 2011 at 10:12:47AM +0200, Ovanes Markarian wrote:

On Fri, Jun 17, 2011 at 9:05 AM, Alessandro Candini wrote:

...

I have different threads which have to work on completely different input and output data (non critical sections): an atomic operation per thread, each one with different time execution but everyone with an intense use of the CPU. Let's say I have 10 operations to perform (10 threads): I would like to run concurrently only 2 threads because of resource consumption.

My problem is that when a threads ends its execution, I would like to suddenly start another thread performing operation 3, in order to have constantly 2 threads working, and so on until the end of operations.

How can I achieve this? I thought to insert my threads into a vector...but I have no idea on how start and join them in order to obtain what described above.

Can anyone post me a little example?

Thanks in advance.

I think you should use a thread pool. You can extend your thread pool with more threads, without impacting other application parts. Take a look at this thread pool implementation: http://threadpool.sourceforge.net/tutorial/intro.html

Or you could stick to the things that are in Boost, like Boost.Asio. Create an asio::io_service; create an asio::io_service::work to keep the workers alive; start N threads, each running asio::io_service::run(); enqueue work by invoking asio::io_service::post() with your tasks. When you're done, destroy the 'work' object and the run() calls will terminate when they're done. For managing the N threads, you can use a thread_group. -- Lars Viklund | zao@acc.umu.se

On Fri, Jun 17, 2011 at 12:20:31PM +0200, Ovanes Markarian wrote:

On Fri, Jun 17, 2011 at 11:10 AM, Lars Viklund wrote:

...

On Fri, Jun 17, 2011 at 10:12:47AM +0200, Ovanes Markarian wrote:

...

On Fri, Jun 17, 2011 at 9:05 AM, Alessandro Candini wrote: [...]

For managing the N threads, you can use a thread_group.

Yes, but than you must implement and debug your own thread pool pattern. Thread group only creates a number of threads, but you need to implement the producer/consumer problem to retrieve work tasks from each thread from the queue. Now there is a choise to either use a ready thread pool or implement it yourself... AFAIK Boost has no official thread pool implementation, which is not in internal impl namespace and guarantees the interface.

We have thread_group, we have Asio. One needs nothing else. I think you're misunderstanding what I suggested. I do not consider doing this remotely bothersome, particularly as you only need features present in Boost itself. It should need no debugging, as it's practically impossible to get wrong. If you for some reason need completion notifications from your workers, have that as part of your task, or wrap your functor in something. ---8<--- asio::io_service io; scoped_ptrasio::io_service::work work(new asio::io_service::work(io)); boost::thread_group tg; auto N = boost::thread::hardware_concurrency(); for (unsigned i = 0; i < N; ++i) tg.create_thread(boost::bind(&asio::io_service::run, boost::ref(io)); io.post(some_task); io.post(some_other_task); // invoke io.post whenever you want to enqueue something. // time passes, shutdown time has arrived work.reset(); tg.join_all(); ---8<--- -- Lars Viklund | zao@acc.umu.se

Comments inline. On Fri, Jun 17, 2011 at 01:33:02PM +0200, Alessandro Candini wrote:

On 17/06/11 12:39, Lars Viklund wrote:

...

---8<--- asio::io_service io; scoped_ptrasio::io_service::work work(new asio::io_service::work(io));

An io_service acts as the hub and dispatcher of completion handlers, typically used for network communication but can also be used for invoking functions on any thread that services the io_service. The 'work' object is needed to pretend that there are pending operations that the io_service is not aware of. If an io_service is out of work, it returns from the run() functions, which we do not want to do until we're done.

...

boost::thread_group tg; auto N = boost::thread::hardware_concurrency();

A thread_group is part of Boost.Thread and lets you create and join groups of threads. hardware_concurrency() simply counts the number of physical processors you have, so it's a good guess to base the number of worker threads on.

...

for (unsigned i = 0; i< N; ++i) tg.create_thread(boost::bind(&asio::io_service::run, boost::ref(io));

io_service::run() blocks until the service is out of work, and can be run in as many threads as you want. There's also single-shot variants and polling variants of it.

...

io.post(some_task); io.post(some_other_task);

io_service::post() takes something callable with the signature `void ()' and invokes it eventually on any thread that is servicing the io_service and isn't busy.

...

// invoke io.post whenever you want to enqueue something.

// time passes, shutdown time has arrived

work.reset(); tg.join_all(); ---8<---

When we're out of pending handlers, and have destroyed the last work item, the functions will start to return, and we can join all the threads.

Could you please give me a working example as little as possible, to better understand all this stuff?

io.post([when_done, other_data] { auto result = compute(other_data); when_done(result); }); This is the form most of my tasks have in the applications where I have a similar setup. You post a task that when completed informs whoever cares via a callback. As Jeroen mentioned on IRC, which I forgot to mention in my first message, this assumes that tasks are largely independent. If they're not and you have more concurrent tasks than there's workers, you might end up with blocking everything, as a handler runs to completion before returning control to run(). Normally, that's not horribly limiting, as the things you tend to block on tend to be asynchronous operations that Asio provides, like reading/writing to sockets, waiting on timers, etc. I recommend that you read the Asio docs, particularly the prose and background bits, and take a look at the examples. Michael Caisse's Boostcon presentation on an Asio Flash XML Server is quite enlightening as well. -- Lars Viklund | zao@acc.umu.se