Threaded generalised transform

Question

I'm writing an application which works with huge amounts of sequential data, and often found the need to use std::transform. I see two potential improvements to std::transform:

1. Allow for variable number parameters.
2. Take advantage of the linear separability of the data by multithreading.

Can anyone suggest any design/performance improvements on my implementation?

threaded_transform.h

#include <vector>
#include <thread>

template<typename InputIterator, typename OutputIterator,
         typename Function, typename... Params>
OutputIterator
trans(InputIterator first, InputIterator last, OutputIterator result,
          Function f, Params... params)
{
    for (; first != last; ++first, ++result)
        *result = f(*first, params...);
    return result;
}

template<typename InputIterator, typename OutputIterator,
         typename Function, typename... Params>
OutputIterator
threaded_transform(unsigned num_threads, InputIterator first,
                   InputIterator last, OutputIterator result,
                   Function f, Params... params)
{
    std::size_t num_values = last - first;
    std::size_t num_values_per_threads = num_values / num_threads;

    std::vector<std::thread> threads;
    threads.reserve(num_threads);

    for (unsigned i = 1; i <= num_threads; ++i) {
        if (i == num_threads) {
            // The last thread processes the remaining values.
            threads.push_back(std::thread(trans<InputIterator, OutputIterator, Function, Params...>,
                                          first, last, result, f, params...));
        } else {
            threads.push_back(std::thread(trans<InputIterator, OutputIterator, Function, Params...>,
                                          first, first + num_values_per_threads, result, f, params...));
        }
        first  += num_values_per_threads;
        result += num_values_per_threads;
    }

    for (auto& thread : threads)
        thread.join();

    return result;
}

main.cpp

#include <vector>
#include "threaded_transform.h"

int main()
{
    auto sum = [] (int a, int b) { return a + b; };

    std::vector<int> values = {1,2,3,4,5,6,7,8,9,10};
    std::vector<int> results;
    results.resize(10);

    threaded_transform(4, values.cbegin(), values.cend(), results.begin(), sum, 10);

    for (auto result : results) {
        std::cout << result << std::endl;
    }
}

Answer 1

You probably want to forward your parameters:

   *result = f(*first, params...);

Try:

   *result = f(*first, std::forward<Params>(params)...);

To go along with forwardign you probably want two versions of trans() on that takes values by reference/value one that takes r-value references:

// Normal parameters.
trans(InputIterator first, InputIterator last, OutputIterator result,
      Function f, Params const&... params)

// R-Value parameters.
trans(InputIterator first, InputIterator last, OutputIterator result,
      Function f, Params&&... params)

Going to main transform function.

I am not sure you in realty want to pass a function and arguments. That's the whole point of the lambda. So you can wrap the function call and its parameters into a function.

threaded_transform(4, values.cbegin(), values.cend(), results.begin(), sum, 10);

// Or would you prefer:

threaded_transform(4, values.cbegin(), values.cend(), results.begin(),
    [](int other){ return add(other, 10);}
);

//Or even
threaded_transform(4, values.cbegin(), values.cend(), results.begin(),
    [](int other){ return 10 + other;}
);

If you do this you should write details about your iterator requirements.

The requirements for std::transform()

template<class InputIterator1, class InputIterator2, class OutputIterator, class BinaryFunction>
OutputIterator transform(InputIterator1 first1, InputIterator1 last1,
                         InputIterator2 first2, OutputIterator result,
                         BinaryFunction binary_op);


Where:
   InputIterator must be a model of Input Iterator.
   OutputIterator must be a model of Output Iterator. 

In your threaded implementation you have a more stringent requirement for the output iterator. I believe it needs to be random access iterator.

    OutputIterator must be a model of Random Access Iterator. 

See: http://www.sgi.com/tech/stl/RandomAccessIterator.html

Answer 2

• std::size_t num_values_per_threads = num_values / num_threads;
  This may unbalance the workload. say we have 11 elements and 4 threads. 11 / 4 = 2. Workload for thread 0, 1, 2 is 2 elements, for thread 3 it is 5 elements. The last thread has the most work limiting the total throughput.
  Additionally you are wasting a thread by making it wait for the futures and doing no actual work. I recommend something like
  std::size_t num_values_per_threads = (num_values + num_threads - 1) / num_threads;
  This gives us num_values_per_threads = 3 with a rest of 2 for the calling thread before it joins with the thread handles. Since the last thread starts last it is probably a good idea to give it a little bit less work to have everyone finish roughly at the same time.

• Providing the number of threads to launch is old and boring. The new hotness is to let the runtime system figure that out. Pseudo-code:

  parallel_transform(begin_range, end_range)
  {
      auto future = async(parallel_transform, begin_range, mid_range);
      parallel_transform(mid_range, end_range);
  }
  

  This should launch as many threads as the hardware can handle without specifying the number explicitly. It is a quality of implementation thing though, there is a chance that you do not get any concurrency with this.

• Why do you use trans instead of std::transform? As far as I can tell they do the same thing and std::transform is more familiar and less difficult to understand that trans.

• std::size_t num_values = last - first;
  This requires random access iterators. It would be nice to make it work with forward iterators so you can use threaded_transform on std::lists. This is a bit more work though.

• Prefer free standing begin and end instead of member functions. values.cbegin() -> cbegin(values). The reason is that C-style arrays and some use defined containers do not have member functions begin and end and especially not cbegin and cend. However, it is usually easy to provide a free function overload for them, so the free function version is more consistent.

• You forgot #include <iostream> for std::cout.

• What you are trying to do is being proposed to the standard under the names Parallelism TS and Parallel STL and is in the process of getting into the C++ standard. You can find some experimental implementations online. There is a good chance every compiler's STL implementation has that by 2017. You are just a bit ahead of the standard. Good :D