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

Subject: Re: [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-04 04:33:44

Hi Nat

I think we're on the same page here - but I'm looking for a more generalised
form. Who owns "obj" or T that_ in my example below is unimportant - I guess
we should enforce using a shared_ptr to "obj" so that it is unimportant who
owns it.

So this is what I came up with before your reply... and although I think
it's not really the right way to do it, I think it will serve to illustrate
what I'm trying to do. (this code works in conjunction with the EventQueue
code I pasted in my original question (see below)

// all callbacks will be of the form void T::mem_fn(...)
// this for the callback of form T::mem_fn()
template <typename T>
class event_0
{
    EventQueue &queue_;         // the event queue which will process the
job
    boost::shared_ptr<T> that_; // the object whose member function the
event queue will call

    typedef void (T::*CB_t)();  // the member function of class T which we
will callback to

    boost::function<void()> job_;

public:
    event_0(boost::shared_ptr<T> that, CB_t cb, EventQueue &queue) :
        queue_(queue),
        that_(that),
        job_(boost::bind(cb, that)) // bind the class instance and its
callback function together for later invocation
    {}

    // the function the user calls which posts the job to the queue. The
thread associated with the queue picks up the
    // job off the queue and calls it (ergo - calls that_->CB_t(); )
    void post() { queue_.post(job_); }
};

// So in my application I have this class:

struct foo0
    : boost::enable_shared_from_this<foo0>
{
    event_0<foo0> ev;

    foo0(EventQueue &queue) : ev(boost::shared_ptr<foo0>(this), &foo0::bar,
queue) { }

    void bar()
    {
        printf("%p foo::bar()\n", this);
    }

    ~foo0()
    {
        std::cout << __func__ << " - Thread id is: " <<
boost::this_thread::get_id() << std::endl;
    }

};

*// and I create an event which is associated with a queue. This is most
likely a class member, where the class is, for eg, the producer, and the
queue being passed to the event, the queue for the consumer.*

boost::shared_ptr<foo0> job_3(new foo0(queue));

*// and then some time later, when the producer has created some data...*

job_3->ev.post(); // resulting in void foo0::bar() function being called on
the consumer thread

*// so - now we want to be able to post a shared_ptr to the consumer - ie:
the producer creates some data (maybe receives it from a socket) and wants
to pass it to the consumer. So we want to be able to go **
job_4->ev.post(data);*

// this for the callback of form T::mem_fn(A0)
template <typename T, typename A0>
class event_1
{
    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 void (T::*CB_t)(A0);
    typedef boost::function<void()> Job;
    Job job_;

public:
    event_1(boost::shared_ptr<T> that, CB_t cb, EventQueue &queue) :
        queue_(queue), that_(that), job_(boost::bind(cb, that, _1))
    {}

    void post(A0 data) { queue_.post(boost::bind(job_, data)); }
};

In other words, I hide the details of the bind from the user, and all he has
to do is create some data and post it to the event, which sorts out the
binding, queueing, etc - We could make it requisite that all classes which
have a callback function used by the event must be derived from shared_ptr
(or be shared_ptrs, or whatever), and all data posted by events (
event_1::post(A0) ) must be shared - to ensure thread safety.

Ideally I'd like to be able to generalise the above somehow, and be able to
cater for post(A0), post(A0, A1), post(A0, A1, A2) etc.

Do you sorta see what I'm trying to do? If I can attempt to explain it
better... I'm trying to create some library code which will handle
inter-thread messaging in an elegant way, and hide the implementation
details from the user. The idea is the EventQueue class will actually be
part of an EventThread class - you create an EventThread, and all it does is
endlessly loop, picking up Event objects from its EventQueue, and calling
their functors (which in fact wrap up a class and member callback function
and has a shared_ptr to some data which is the argument to the member
callback function) - So the creation of the thread, the creation of the
queue etc is hidden - all the user does is create events and call
event::post(data).

Do you think this is even a good design? Would you suggest a different
design philosophy?

TIA
Steve

PS: My EventQueue code is here:

#include <boost/enable_shared_from_this.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <stdio.h>
#include <list>
#include <iostream>

// cross-thread event queue. multiple threads can post jobs, one or many
threads can execute
// jobs.
class EventQueue
{
public:
    typedef boost::function<void()> Job;

private:

    boost::mutex mtx_;
    boost::condition cnd_;
    typedef std::list<Job> Jobs;
    Jobs jobs_;
    int stop_;

public:

    // 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); }
};
//-----------------------------------------------------------------------

// all callbacks will be of the form void T::mem_fn(...)
// this for the callback of form T::mem_fn()
template <typename T>
class event_0
{
    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 void (T::*CB_t)();
    typedef boost::function<void()> Job;
    Job job_;

public:
    event_0(boost::shared_ptr<T> that, CB_t cb, EventQueue &queue) :
        queue_(queue), that_(that), job_(boost::bind(cb, that))
    {}

    void post() { queue_.post(job_); }
};
//-----------------------------------------------------------------------

struct foo0
    : boost::enable_shared_from_this<foo0>
{
    event_0<foo0> ev;

    foo0(EventQueue &queue) : ev(boost::shared_ptr<foo0>(this), &foo0::bar,
queue) { }

    void bar()
    {
        printf("%p foo::bar()\n", this);
    }

    ~foo0()
    {
        std::cout << __func__ << " - Thread id is: " <<
boost::this_thread::get_id() << std::endl;
    }

};
//-----------------------------------------------------------------------

void anotherThread(EventQueue* queue)
{
    std::cout << __func__ << " - Thread id is: " <<
boost::this_thread::get_id() << std::endl;
    EventQueue::Job job;
    // wait and execute jobs till stopped
    while(queue->pull(&job))
        job(); // execute the job
}
//-----------------------------------------------------------------------

int main()
{
    std::cout << __func__ << " - Thread id is: " <<
boost::this_thread::get_id() << std::endl;

    EventQueue queue;
    // start another thread and pass an argument
    boost::thread another_thread(boost::bind(anotherThread, &queue));

    boost::shared_ptr<foo0> job_3(new foo0(queue));
    job_3->ev.post();

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

    return 0;
}
//-----------------------------------------------------------------------

On 3 June 2010 22:11, Nat Goodspeed <nat_at_[hidden]> wrote:

it's not really clear to me whether 'obj' is owned by the consumer thread or
> is in the hands of the producer object. Let's consider both cases.
>
> ==== Either way:
>
> class Data { ... };
> typedef boost::shared_ptr<Data> DataPtr;
>
> ==== If 'obj' is the ConsumerClass instance:
>
> class ConsumerClass
> {
> public:
>    ...
>    void run();
>    void doSomething(DataPtr data);
>    void somethingElse(DataPtr data);
>    ...
> };
>
> typedef (ConsumerClass::*ConsumerMethod)(DataPtr);
>
> typedef boost::function<void(ConsumerClass*)> QueueItem;
> typedef YourFavoriteThreadSafeQueue<QueueItem> Queue;
>
> void ConsumerClass::run()
> {
>    for (;;)
>    {
>        QueueItem item(queue.get());
>        item(this);
>    }
> }
>
> void ProducerClass::post(ConsumerMethod method, DataPtr data)
> {
>    queue.put(boost::bind(method, _1, data));
> }
>
>