Re: [Boost-users] Multithread event handling

Subject: Re: [Boost-users] Multithread event handling
From: Matthias Vallentin (vallentin_at_[hidden])
Date: 2010-01-25 02:06:25

On Fri, Jan 22, 2010 at 09:00:42PM -0600, Alessandro Bellina wrote:
> Can somebody share their experiences on libraries or patterns used to
> implement an event handling mechanism?

Below is code for an experimental scheduler that I am currently writing,
based on boost::asio::io_service object with a bunch of threads. I am
also using the new boost::packaged_task and boost::shared_future classes
that arrived with 1.41. There are a few
issues, though:

    1) I don't know why the mutex in the handler (func) does not work.
    2) No structure (e.g., tree-like) on the tasks as in TBB, no priorities.

#include <memory>
#include <iostream>
#include <boost/asio/io_service.hpp>
#include <boost/thread.hpp>
#include "boost/utility/result_of.hpp"

namespace core {

template <typename R>
class task
{
public:
    typedef typename boost::unique_future<R> unique_future;

private:
    typedef boost::packaged_task<R> task_type;

public:
    template <typename F>
    explicit task(const F& f)
      : pt_(new task_type(f))
    {
    }

    void operator()()
    {
        (*pt_)();
    }

    unique_future future()
    {
        return pt_->get_future();
    }

private:
    boost::shared_ptr<task_type> pt_;
};

class scheduler
{
public:
    template <typename T>
    typename T::unique_future put(T& t)
    {
        io_service_.post(boost::bind(&T::operator(), t));

        return t.future();
    }

    template <typename F>
    typename task<typename boost::result_of<F()>::type>::unique_future
    schedule(const F& f)
    {
        task<typename boost::result_of<F()>::type> t(f);

        return put(t);
    }

    void run(unsigned n = boost::thread::hardware_concurrency())
    {
        sentinel_.reset(new boost::asio::io_service::work(io_service_));

        for (unsigned i = 0; i < n; ++i)
            threads_.add_thread(new boost::thread(
                        boost::bind(&boost::asio::io_service::run, &io_service_)));
    }        

    void finish()
    {
        sentinel_.reset();
    }

    void interrupt()
    {
        io_service_.stop();
    }

private:
    boost::thread_group threads_;
    boost::asio::io_service io_service_;
    std::auto_ptr<boost::asio::io_service::work> sentinel_;
};

} // namespace core

static boost::mutex mtx;

void func(int i)
{
    boost::mutex::scoped_lock(mtx);
    std::cout << "i = " << i << std::endl;
}

int main()
{
    core::scheduler s;
    s.run();

    core::task<void> task(boost::bind(&func, 42));
    core::task<void>::unique_future f = s.put(task);
    core::task<void>::unique_future g = s.schedule(boost::bind(&func, 43));
    core::task<void>::unique_future h = s.schedule(boost::bind(&func, 44));

    h.wait();
    g.wait();
    f.wait();

    return 0;
}

   Matthias
-- 
Matthias Vallentin
vallentin_at_[hidden]
http://www.icir.org/matthias