How to create a queue that contains boost :: packaged_task <> with functions that return arbitrary types?

Question

I am trying to build a work queue of functions that must be executed by a single thread and can be powered by many threads. For this, I planned to use boost :: packaged_task and boost :: unique_future. The idea would be:

Value Foo = queue.add (myFunc) .get ();

which blocks until the function is executed. So queue.add (...) takes a boost :: function and returns boost :: unique_future. Then inside he creates boost :: packaged_task, using boost :: function for his constructor.

The problem I am facing is that boost :: function <...> will not be the same every time. In particular, the return value for it will change (functions, however, will never take any parameters). So I should have an add function that looks something like this:

 template <typename ResultType> boost::unique_future<ResultType> add(boost::function<ResultType ()> f) { boost::packaged_task<boost::function<ResultType ()> > task(f); queue.push_back(task); return task.get_future(); } 

Well, that doesn't seem too bad, but then I ran into the problem of defining a “queue”. I think I have no choice but to use boost :: any, since the types will not be constant:

 std::list<boost::any> queue; // note: I'm not concerned with thread-safety yet 

But then I encounter a problem when I try to implement my executeSingle (takes only one element from the queue to execute):

 void executeSingle() { boost::any value = queue.back(); boost::packaged_task<?> task = boost::packaged_task<?>(boost::move(value)); // actually execute task task(); queue.pop_back(); } 

"?" that i'm not sure. I cannot call executeSingle with a template, as it called from a separate thread. I tried using boost :: any, but getting an error:

  conversion from 'boost::any' to non-scalar type boost::detail::thread_move_t<boost:thread>' requested. 

The funny thing is that at the moment I don’t care about the packaged_task return type, I just want to execute it, but I can figure out the details of the template.

Any understanding would be greatly appreciated!

Answer 1

You must save boost::function<void()> . Note that boost::packaged_task<R>::operator() does not return anything; it populates the associated boost::future . In fact, even if he returned something, you can still use boost::function<void()> , since you still won’t be interested in the return value: all you care about is call queue.back()() . If that were the case, boost::function<void()>::operator() took care of discarding the return value for you.

As a minor note, you may need to change the signature of your add method for the template on a typical Functor type, rather than boost::function , and use boost::result_of to get the result type for boost::packaged_task .

My whole suggestion:

 template<typename Functor> boost::future<typename boost::result_of<Functor()>::type> queue::add(Functor functor) // assuming your class is named queue { typedef typename boost::result_of<Functor()>::type result_type; boost::packaged_task<result_type> task(functor); boost::unique_future<result_type> future = task.get_future(); internal_queue.push_back(boost::move(task)); // assuming internal_queue member return boost::move(future); } void queue::executeSingle() { // Note: do you really want LIFO here? queue.back()(); queue.pop_back(); } 

EDIT

How to take care of the semantics of movement inside queue::add

 typedef typename boost::result_of<Functor()>::type result_type; typedef boost::packaged_task<result_type> task_type; boost::shared_ptr<task_type> task = boost::make_shared<task_type>(functor); boost::unique_future<result_type> future = task->get_future(); /* boost::shared_ptr is possibly move-enabled so you can try moving it */ internal_queue.push_back( boost::bind(dereference_functor(), task) ); return boost::move(future); 

where dereference_functor can be:

 struct dereference_functor { template<typename Pointer> void operator()(Pointer const& p) const { (*p)(); } }; 

You can also replace the bind expression for clearer

 boost::bind(&task_type::operator(), task) 

which also does not require a special functor. However, if there are multiple task_type::operator() overloads, this may require disambiguation; the code may also be broken if in the future a change in Boost.Thread leads to an overload.

Answer 2

You are using old-fashioned virtual functions. Define the base class task_base using the virtual execute method, then define the derived class of the template that contains the specific task instance. Something like lines:

 struct task_base { virtual void execute() = 0; }; template<typename ResultType> struct task_holder : task_base { task_holder(boost::packaged_task<boost::function<ResultType ()> >&& task) : m_task(task) { } void execute() { m_task(); } private: boost::packaged_task<boost::function<ResultType ()> > m_task; }; 

And define a queue to store unique_ptr<task_base> . This is essentially what boost::any does, only you use a specific function, namely execute .

NOTE. Unverified code! And I'm still not very familiar with rvalue links. This is just to give you an idea of ​​what the code looks like.

Answer 3

Somewhat belated, but you might want to use Boost.Asio instead of rolling your queue solution.

While it has grown as an I / O library, it supports asynchronous calls that way. Just define io_service somewhere, run it inside the thread, and then post functors to get a call in that thread.