[Boost-users] [bind] & [function] How do I store member functions with generalised arguments for later invocation?

Subject: [Boost-users] [bind] & [function] How do I store member functions with generalised arguments for later invocation?
From: Steve Lorimer (steve.lorimer_at_[hidden])
Date: 2010-06-02 13:21:18

I am wanting to write library code that will allow a user to create an
"event" which is associated with a thread, a class instance, and a member
function of that class. The user can then "post" data (generalised number of
arguments) - which results in the class instance's member function being
called in the context of the associated thread.

For example (pseudo code):

event
{
    thread* thr;
    class &that;
    mem_fun* fun;

    void post();
    void post(a1);
    void post(a1, a2);
    void post(a1, a2, a3, ...etc);
}

mem_fun* fun stores a member function pointer which is of the form void
class::fun(...0, 1 or more arguments...)

Here is some example code which uses boost::bind to bind arguments to member
functions and posts them to be called on another thread. However, it does
not go far enough.

#include <boost/function.hpp>
> #include <boost/bind.hpp>
> #include <boost/enable_shared_from_this.hpp>
> #include <boost/thread/thread.hpp>
> #include <boost/thread/mutex.hpp>
> #include <boost/thread/condition.hpp>
> #include <list>
> #include <stdio.h>
>
> // cross-thread event queue. multiple threads can post jobs, one or many
> threads can execute
> // jobs.
> class EventQueue
> {
> public:
>     typedef boost::function<void()> Job;
>
>     // puts a jon into the queue
>     void post(Job job)
>     {
>         boost::mutex::scoped_lock lock(mtx_);
>         jobs_.push_back(job);
>         cnd_.notify_one();
>     }
>
>     // pulls one job from the queue, returns false when stopped
>     bool pull(Job* job)
>     {
>         boost::mutex::scoped_lock lock(mtx_);
>         for(;;) { // handle spurious wake-ups
>             while(!stop_ && jobs_.empty())
>                 cnd_.wait(lock);
>             if(!jobs_.empty() && 2 != stop_) {
>                 job->swap(jobs_.front()); // move efficiently, avoiding
> *job = jobs.front()
>                 jobs_.pop_front();
>                 return true;
>             }
>             else if(stop_) {
>                 return false;
>             }
>         }
>     }
>
>     // make pull() return false
>     void stop(bool cancel_jobs)
>     {
>         boost::mutex::scoped_lock lock(mtx_);
>         stop_ = 1 + cancel_jobs; // 1 - complete jobs, 2 - cancel jobs
>         cnd_.notify_all();
>     }
>
>     EventQueue() : stop_() {}
>     ~EventQueue() { this->stop(true); }
>
> private:
>     boost::mutex mtx_;
>     boost::condition cnd_;
>     typedef std::list<Job> Jobs;
>     Jobs jobs_;
>     int stop_;
> };
>
> struct JobX
> // micro-oprimization, embed the reference counter into the job to avoid an
> extra memory
> // allocation by boost::shared_ptr ctor
>     : boost::enable_shared_from_this<JobX>
> {
>     void foo()
>     {
>         printf("%p foo()\n", this);
>     }
>
>     void bar(int a)
>     {
>         printf("%p bar(%d)\n", this, a);
>     }
> };
>
> void anotherThread(EventQueue* queue)
> {
>     EventQueue::Job job;
>     // wait and execute jobs till stopped
>     while(queue->pull(&job))
>         job(); // execute the job
> }
>
> int main()
> {
>     EventQueue queue;
>     // start another thread and pass an argument
>     boost::thread another_thread(boost::bind(anotherThread, &queue));
>
>     // post jobs, allocate in this thread, deallocate in the other
>     boost::shared_ptr<JobX> job_0(new JobX);
>     queue.post(boost::bind(&JobX::foo, job_0));
>     // post several jobs to the same object, deallocated when no longer in
> use
>     boost::shared_ptr<JobX> job_1(new JobX);
>     queue.post(boost::bind(&JobX::foo, job_1));
>     queue.post(boost::bind(&JobX::bar, job_1, 1)); // pass an extra arg 1
>     queue.post(boost::bind(&JobX::bar, job_1, 2)); // pass an extra arg 2
>

>     // stop the queue and let it complete all jobs
>     queue.stop(false);
>     another_thread.join();
> }
>

And this is where I fall down. I am able to bind arguments to JobX::bar
above, at the point at which I create my boost::function<void()>, but if I
want to bind _1, _2 etc arguments for calling later, I'm not sure what to
do.

So my question is, how should I form class Event such that I can store
member functions of the form void T::fn(), void T::fn(a1), void T::fn(a1,
a2, etc...), and then allow a generalised form of post(), post(a1), post(a1,
a2, etc...) to bind the arguments given in post to the callback and post it
to the queue?

I have some code below which clearly won't compile, but should illustrate to
some degree what I'm looking for.

template <typename T>
class Event
{
    EventQueue &queue_;         // the event queue which will process the
job
    boost::shared_ptr<T> that_; // the object whose mem_fun the event queue
will call

    typedef boost::function<void (T*, void) > CB_0 ; // used for callbacks
which are of the form void T::fn();
    typedef boost::function<void (T*, _1) > CB_1 ;   // used for callbacks
which are of the form void T::fn(1 arg);
    typedef boost::function<void (T*, _1, _2) > CB_2 ;   // used for
callbacks which are of the form void T::fn(1 arg);
    CB_0 cb_0;
    CB_1 cb_1;
    CB_2 cb_2;
public:
    Event(boost::shared_ptr<T> &that, CB &cb, EventQueue &queue) :
        that_(that), cb_(cb), queue_(queue)
    {}

    void post() { queue_.post(boost::bind(&cb_0, that));

    template <typename arg_1>
    void post(arg_1 &arg1) { queue_.post(boost::bind(&cb_1, that));

    template <typename arg_1, typename arg_2>
    void post(arg_1 &arg1, arg_2 &arg2) { queue_.post(boost::bind(&cb_2,
that)); // TODO: args should be shared_ptr
};

TIA
Steve