4
\$\begingroup\$

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;
    }
}
\$\endgroup\$
3
\$\begingroup\$

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

\$\endgroup\$
3
\$\begingroup\$
  • 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

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.