std::packaged_task with std::placeholders

Question

MAJOR EDIT TO SIMPLIFY CODE (and solved)

I would like to be able to make a packaged task that has a free unbound argument, which I will then add at call time of the packaged task.

In this case, I want the first argument to the function (of type size_t) to be unbound.

Here is a working minimal example (this was the solution):

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>
#include <cstdlib>
#include <cstdio>

//REV: I'm trying to "trick" this into for double testfunct( size_t arg1, double arg2), take enqueue( testfunct, 1.0 ), and then internally, execute
// return testfunct( internal_size_t, 1.0 )

template<typename F, typename... Args>
auto enqueue(F&& f, Args&&... args) 
  -> std::future<typename std::result_of<F(size_t, Args...)>::type>
{
  using return_type = typename std::result_of<F(size_t, Args...)>::type;

  //REV: this is where the error was, I was being stupid and thinking this task_contents which would be pushed to the queue should be same (return?) type as original function? Changed to auto and everything worked... (taking into account Jans's result_type(size_t) advice into account.
  //std::function<void(size_t)> task_contents = std::bind( std::forward<F>(f), std::placeholders::_1, std::forward<Args>(args)... );
  auto task_contents = std::bind( std::forward<F>(f), std::placeholders::_1, std::forward<Args>(args)... );

  std::packaged_task<return_type(size_t)> rawtask(
                          task_contents );

  std::future<return_type> res = rawtask.get_future();

  size_t arbitrary_i = 10;
  rawtask(arbitrary_i);
  return res;
}


double testfunct( size_t threadidx, double& arg1 )
{
  fprintf(stdout, "Double %lf Executing on thread %ld\n", arg1, threadidx );
  std::this_thread::sleep_for( std::chrono::milliseconds(1000) );
  return 10; //true;
}

int main()
{
  std::vector<std::future<double>> myfutures;

  for(size_t x=0; x<100; ++x)
    {
      double a=x*10;
      myfutures.push_back(
              enqueue( testfunct, std::ref(a) )
                  );
    }

  for(size_t x=0; x<100; ++x)
    {
      double r = myfutures[x].get();
      fprintf(stdout, "Got %ld: %f\n", x, r );
    }
}

Answer 1

The main issues are on ThreadPool::enqueue:

std::function<void(size_t)> task1 = std::bind( std::forward<F>(f), std::placeholders::_1, std::forward<Args>(args)... );

Here, the type of task1 is std::function<void(std::size_t)> but the result of the std::bind when evaluated with funct is convertible to std::function<bool(std::size_t)> and even though as @T.C has pointed out, you can assign the result of the bind to task1, in order to pass task1 to std::make_shared you need to honor the return_type you've got.

Change the above to:

std::function<return_type(size_t)> task1 = std::bind( std::forward<F>(f), std::placeholders::_1, std::forward<Args>(args)... );

Now the same for:

auto task = std::make_shared< std::packaged_task<return_type()> >( task1 );

but in this case is the parameter type that is missing. Change it to:

auto task = std::make_shared< std::packaged_task<return_type(std::size_t)> >( task1 );

ThreadPool::tasks store function objects of type std::function<void(std::size_t)> but you're storing lambda that receive no arguments. Change the tasks.emplace(...) to:

tasks.emplace([task](std::size_t a){ (*task)(a); });

Answer 2

The code is not very well formatted, but a solution.

First, you should wrap the results in the lambda creation, not pass functions that can return anything. But if you want to use a shared pointer on a task, this works.

In the prototype:

std::future<void> enqueue(std::function<void(size_t)> f);

using Task = std::function<void(size_t)>;
// the task queue
std::queue<Task> tasks;

std::optional<Task> pop_one();

Implementation becomes:

ThreadPool::ThreadPool(size_t threads)
    :   stop(false)
{
    for(size_t i = 0;i<threads;++i)
        workers.emplace_back(
        [this,i]
            {
                for(;;)
                {
                auto task = pop_one();
                if(task)
                {
                    (*task)(i);
                }
                else break;
                }
            }
        );
}

std::optional<ThreadPool::Task> ThreadPool::pop_one()
{
    std::unique_lock<std::mutex> lock(this->queue_mutex);
    this->condition.wait(lock,
        [this]{ return this->stop || !this->tasks.empty(); });
    if(this->stop && this->tasks.empty())
    {
        return std::optional<Task>();
    }
    auto task = std::move(this->tasks.front()); //REV: this moves into my thread the front of the tasks queue.
    this->tasks.pop();

    return task;
}

template<typename T>
std::future<T> ThreadPool::enqueue(std::function<T(size_t)> fun)
{
    auto task = std::make_shared< std::packaged_task<T(size_t)> >([=](size_t size){return fun(size);});

    auto res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queue_mutex);

        // don't allow enqueueing after stopping the pool
        if(stop)
        {
            throw std::runtime_error("enqueue on stopped ThreadPool");
        }

        tasks.emplace([=](size_t size){(*task)(size);});
    }
    condition.notify_one();
    return res;
}

And now you can have your main:

int main()
{
  size_t nthreads=3;
  ThreadPool tp(nthreads);
  std::vector<std::future<bool>> myfutures;

  for(size_t x=0; x<100; ++x)
    {
      myfutures.push_back(
          tp.enqueue<bool>([=](size_t threadidx) {return funct(threadidx, (double)x * 10.0);}));
    }

  for(size_t x=0; x<100; ++x)
    {
      bool r = myfutures[x].get();
      std::cout << "Got " << r << "\n";
    }
}

There is now an explicit return type when wrapping the lambda, as the return type is templated.