This is a reworking of the wrapper to make any object active (that is, to
execute all its methods in its own thread of execution).
Changes since the previous post:
1. Extensible - the proxy implementation method can now be extended to any
number of parameters.
2. Future values - where the client can, if desired, postpone the wait to
acquire the result of invoking a method on an active method by explicitly
capturing a handle to the result for later inspection.
3. Policy for void proxies - the implementer of the class which wraps a non-
active class, can specify whether a client must wait for a void method
to complete exceution or not. This would be useful if a wrapped method had
side effects that a client would expect to see after invoking the method.
Comments welcome.
Matt
(Sorry this is so long, I can only post in text mode via gmane)
// test_active_object.cpp --------------------------------------------------
#include "active_object.hpp"
using std::cin;
using std::endl;
using ActiveWrapper::active; // Active-object wrapper
using ActiveWrapper::future; // Future value encapsulator
using ActiveWrapper::output; // std::cout locking object
using ActiveWrapper::sleep; // 3-second pause to make effects observable
// An example class to test active behaviour
struct Object
{
int non_void_with_param(int i)
{
{ output() << "\tInside Object::non_void_with_param" << endl; }
sleep();
return (i*i);
}
const char* non_void_without_param(void)
{
{ output() << "\tInside Object::non_void_without_param" << endl; }
sleep();
return "Hello";
}
void void_with_param(double)
{
{ output() << "\tInside Object::void_with_param" << endl; }
sleep();
}
void void_without_param(void)
{
{ output() << "\tInside Object::void_without_param" << endl; }
sleep();
}
};
// An active version of the Object class
struct ActiveObject : private active<Object>
{
ActiveObject() :
object(),
non_void_with_param(&object, &Object::non_void_with_param, this),
non_void_without_param(&object, &Object::non_void_without_param, this),
void_with_param(&object, &Object::void_with_param, this),
void_without_param(&object, &Object::void_without_param, this)
{
}
// Proxy declarations for all methods of the active object
proxy<int (int)> non_void_with_param;
proxy<const char* (void)> non_void_without_param;
proxy<void (double), true> void_with_param;
proxy<void (void)> void_without_param;
private:
Object object;
};
int main(int, char**)
{
{
ActiveObject object;
{ output() << "Invoking methods from main thread" << endl; }
// Implicit wait until task is completed by acquiring result
int i = object.non_void_with_param(5);
{ output() << "Result of non_void_with_param: " << i << endl; }
// Explicitly bypass wait for result
future<const char*> c = object.non_void_without_param();
{ output() << "Returned from non_void_without_param" << endl; }
// Block until task completion by accessing result
{ output() << "Result of non_void_without_param: " << c << endl; }
// Block until task completion, due to proxy policy parameter
object.void_with_param(3.1415927);
{ output() << "Returned from void_with_param" << endl; }
// Explicitly bypass completion-wait policy
future<void> v = object.void_with_param(1e2);
{ output() << "Not waiting for void_with_param" << endl; }
// Non-blocking call due to default policy
object.void_without_param();
{ output() << "Returned from void_without_param" << endl; }
// Block until active object thread closes on destruction
{ output() << "Waiting for completion of all tasks" << endl; }
}
{ output() << "Press <Enter> to continue..." << endl; }
cin.get();
return 0;
}
// active_object.hpp -------------------------------------------------------
#include <iomanip>
#include <iostream>
#include <queue>
#include <boost/any.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/optional.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/xtime.hpp>
#include <boost/tuple/tuple.hpp>
#include <boost/utility.hpp>
namespace ActiveWrapper {
// Lock the output stream to prevent interleaving
struct output
{
output() :
lock(mutex) {}
static unsigned count;
private:
static boost::mutex mutex;
boost::mutex::scoped_lock lock;
};
unsigned output::count;
boost::mutex output::mutex;
template <typename T>
std::ostream& operator<< (const output& out, T t)
{
++out.count;
return std::cout << std::setw(2) << std::setfill(' ')
<< out.count << ": " << t;
}
// Pause to demonstrate effect
int sleep(void)
{
boost::xtime time;
boost::xtime_get(&time, boost::TIME_UTC);
time.sec += 3;
boost::thread::sleep(time);
}
namespace detail {
// A generic thread-safe queue
template <typename T>
struct safe_queue
{
safe_queue(void) :
terminated(false) { }
bool enqueue (T t)
{
if (terminated)
return false;
boost::mutex::scoped_lock lock(mutex);
queue.push(t);
not_empty.notify_one();
return true;
}
boost::optional<T> dequeue(void)
{
boost::mutex::scoped_lock lock(mutex);
if (terminated)
return boost::optional<T>();
while (queue.empty())
{
not_empty.wait(lock);
if (terminated)
return boost::optional<T>();
}
T t = queue.front();
queue.pop();
return t;
}
void terminate(void)
{
boost::mutex::scoped_lock lock(mutex);
terminated = true;
not_empty.notify_all();
}
private:
bool terminated;
std::queue<T> queue;
boost::mutex mutex;
boost::condition not_empty;
};
// Object used to wait for the completion of a task
struct synch_type
{
synch_type(boost::optional<boost::any*> a,
boost::mutex* m,
boost::condition* c) :
result(a),
mutex(m),
condition(c),
completed(false) {}
~synch_type()
{
if (result)
delete *result;
delete mutex;
delete condition;
}
void wait_for_completion(void)
{
boost::mutex::scoped_lock lock(*mutex);
if (completed == false)
{
if (result)
{
while ((*result)->empty())
(*condition).wait(lock);
}
else
{
(*condition).wait(lock);
}
completed = true;
}
}
void notify(void)
{
boost::mutex::scoped_lock lock(*mutex);
(*condition).notify_all();
completed = true;
}
boost::optional<boost::any*> get_result(void)
{
return result;
}
private:
boost::optional<boost::any*> result;
boost::mutex* mutex;
boost::condition* condition;
bool completed;
};
// A reference-counted pointer to a synch type
typedef boost::shared_ptr<synch_type> synch_ptr;
// A type to enforce waiting policy on synchronisation
struct waiter_type
{
waiter_type(synch_ptr s, bool force) :
synch(s), force_wait(force), waited_for_completion(false) {}
~waiter_type()
{
if (force_wait)
{
wait_for_completion();
}
}
void wait_for_completion(void)
{
if (waited_for_completion == false)
{
synch->wait_for_completion();
waited_for_completion = true;
}
}
boost::optional<boost::any*> result(void)
{
waited_for_completion = true;
return synch->get_result();
}
private:
synch_ptr synch;
bool force_wait;
bool waited_for_completion;
};
}
// An encapsulation of a result that might require waiting for
template <typename T>
struct future
{
future(boost::shared_ptr<detail::waiter_type> w) :
waiter(w) {}
operator T& (void)
{
if (!value)
{
waiter->wait_for_completion();
boost::optional<boost::any*> result = waiter->result();
value = boost::any_cast<T>(*(*result));
}
return *value;
}
protected:
boost::optional<T> value;
boost::shared_ptr<detail::waiter_type> waiter;
};
template <>
struct future<void>
{
future(boost::shared_ptr<detail::waiter_type> w) :
waiter(w) {};
protected:
boost::shared_ptr<detail::waiter_type> waiter;
};
// Forward reference required by proxy_impl
template <typename base>
class active;
namespace detail {
// The type of tasks that are executed by the active object wrapper
typedef boost::function<boost::any (void)> task_type;
// A task and associated notification machinery
typedef boost::tuple<task_type, boost::shared_ptr<synch_type> >
task_descriptor;
// A thead safe queue of tasks to be performed
typedef safe_queue<task_descriptor> task_queue;
// Dispatch to task queue, returning a facility to wait for the result
template <typename result_type>
future<result_type> invoke(task_queue* queue,
bool force_wait,
task_type task)
{
synch_ptr synch(new synch_type(new boost::any(),
new boost::mutex(),
new boost::condition()));
task_descriptor td(task, synch);
queue->enqueue(td);
boost::shared_ptr<waiter_type> waiter
(new waiter_type(synch, force_wait));
return future<result_type>(waiter);
}
template <>
future<void> invoke<void>(task_queue* queue,
bool force_wait,
task_type task)
{
synch_ptr synch(new synch_type(boost::optional<boost::any*>(),
new boost::mutex(),
new boost::condition()));
task_descriptor td(task, synch);
queue->enqueue(td);
boost::shared_ptr<waiter_type> waiter
(new waiter_type(synch, force_wait));
return future<void>(waiter);
}
template <typename base, typename signature, bool force_wait>
class proxy_impl;
// Handler for non-void functions with one parameter
template <typename base, typename R, typename T1, bool force_wait>
class proxy_impl<base, R (T1), force_wait>
{
typedef R result_type;
typedef T1 arg1_type;
typedef result_type (base::*method_type)(arg1_type);
typedef boost::function<result_type (arg1_type)> wrapper_type;
typedef boost::function<future<result_type> (task_type)>
invocation_type;
static boost::any wrap(wrapper_type task, arg1_type arg1)
{
return task(arg1);
}
invocation_type invocation;
wrapper_type task;
public:
proxy_impl(base* object, method_type method, active<base>* active)
{
task = boost::bind(method, object, _1);
invocation = boost::bind(&invoke<result_type>,
active->queue(),
force_wait,
_1);
}
future<result_type> operator()(arg1_type arg1)
{
return invocation(boost::bind(&wrap, task, arg1));
}
};
// Handler for void functions with one parameter
template <typename base, typename T1, bool force_wait>
class proxy_impl<base, void (T1), force_wait>
{
typedef void result_type;
typedef T1 arg1_type;
typedef result_type (base::*method_type)(arg1_type);
typedef boost::function<result_type (arg1_type)> wrapper_type;
typedef boost::function<future<result_type> (task_type)>
invocation_type;
static boost::any wrap(wrapper_type task, arg1_type arg1)
{
task(arg1);
return boost::any();
}
invocation_type invocation;
wrapper_type task;
public:
proxy_impl(base* object, method_type method, active<base>* active)
{
task = boost::bind(method, object, _1);
invocation = boost::bind(&invoke<result_type>,
active->queue(),
force_wait,
_1);
}
future<result_type> operator()(arg1_type arg1)
{
return invocation(boost::bind(&wrap, task, arg1));
}
};
// Handler for non-void functions with no parameters
template <typename base, typename R, bool force_wait>
class proxy_impl<base, R (void), force_wait>
{
typedef R result_type;
typedef result_type (base::*method_type)(void);
typedef boost::function<result_type (void)> wrapper_type;
typedef boost::function<future<result_type> (task_type)>
invocation_type;
static boost::any wrap(wrapper_type task)
{
return task();
}
invocation_type invocation;
wrapper_type task;
public:
proxy_impl(base* object, method_type method, active<base>* active)
{
task = boost::bind(method, object);
invocation = boost::bind(&invoke<result_type>,
active->queue(),
force_wait,
_1);
}
future<result_type> operator()(void)
{
return invocation(boost::bind(&wrap, task));
}
};
// Handler for void functions without parameters
template <typename base, bool force_wait>
class proxy_impl<base, void (void), force_wait>
{
typedef void result_type;
typedef result_type (base::*method_type)(void);
typedef boost::function<result_type (void)> wrapper_type;
typedef boost::function<future<result_type> (task_type)>
invocation_type;
static boost::any wrap(wrapper_type task)
{
task();
return boost::any();
}
invocation_type invocation;
wrapper_type task;
public:
proxy_impl(base* object, method_type method, active<base>* active)
{
task = boost::bind(method, object);
invocation = boost::bind(&invoke<result_type>,
active->queue(),
force_wait,
_1);
}
future<result_type> operator()(void)
{
return invocation(boost::bind(&wrap, task));
}
};
}
// Wrapper class to supply active behaviour to other objects
template <typename base>
class active
{
void thread_function(void)
{
while (true)
{
boost::optional<detail::task_descriptor> td = tasks.dequeue();
if (!td)
break;
detail::task_type task = td->template get<0>();
detail::synch_ptr synch = td->template get<1>();
boost::optional<boost::any*> result = synch->get_result();
if (result)
{
*(*result) = task();
}
else
{
task();
}
synch->notify();
}
}
detail::task_queue tasks;
boost::thread thread;
protected:
active(void) :
thread(boost::bind(&active<base>::thread_function, this))
{
}
~active()
{
tasks.terminate();
thread.join();
}
template <typename signature, bool force_wait = false>
struct proxy : public detail::proxy_impl<base, signature, force_wait>
{
typedef detail::proxy_impl<base, signature, force_wait> impl_type;
typedef typename impl_type::method_type method_type;
proxy(base* object, method_type method, active<base>* active) :
impl_type(object, method, active) {}
};
public:
detail::task_queue* queue(void) { return &tasks; }
};
};
