Parallel Loops in C++

Parallel Loops in C++

I wonder if there is a light, straight forward way to have loops such as for and range based-for loops compute in parallel in C++. How would you implement such a thing? From Scala I know the map, filter and foreach functions and maybe it would also be possible to perform these in parallel? Is there an easy way to achieve this in C++?

My primary platform is Linux, but it would be nice if it worked cross-platform.

7

8 Answers

With the parallel algorithms in C++17 we can now use:

std::vector<std::string> foo;
std::for_each(
    std::execution::par,
    foo.begin(),
    foo.end(),
    [](auto&& item)
    {
        //do stuff with item
    });

to compute loops in parallel. The first parameter specifies the execution policy

3

What is your platform? You can look at OpenMP, though it's not a part of C++. But it is widely supported by compilers.

As for range-based for loops, see, e.g., Using OpenMP with C++11 range-based for loops?.

I've also seen few documents at that indicate some efforts to incorporate parallel constructs/algorithms into future C++, but don't know what's their current status.

UPDATE

Just adding some exemplary code:

template <typename RAIter>
void loop_in_parallel(RAIter first, RAIter last) {
   const size_t n = std::distance(first, last);

   #pragma omp parallel for
   for (size_t i = 0; i < n; i++) {
       auto& elem = *(first + i);
       // do whatever you want with elem
    }
}

The number of threads can be set at runtime via the OMP_NUM_THREADS environment variable.

5

With C++11 you can parallelize a for loop with only a few lines of codes.

My function parallel_for() (define later in the post) splits a for loop into smaller chunks (sub loops), and each chunk assigned to a thread. Here is the usage:

/// Say you want to parallelize this:
for(int i = 0; i < nb_elements; ++i)
    computation(i);    

/// Then you would do:
parallel_for(nb_elements, [&](int start, int end){ 
    for(int i = start; i < end; ++i)
        computation(i); 
});

My parallel_for() also works within a class:

struct My_obj {

    /// Replacing:
    void sequential_for(){
        for(int i = 0; i < nb_elements; ++i)
            computation(i);
    }

    /// By:
    void process_chunk(int start, int end)
    {
        for(int i = start; i < end; ++i)
            computation(i);
    }

    void threaded_for(){
        parallel_for(nb_elements, [this](int s, int e){ 
            this->process_chunk(s, e); 
        } );
    }

    
};

Finally here is the implementation of parallel_for(), just paste in a header file and use it at will:

#include <algorithm>
#include <thread>
#include <functional>
#include <vector>

/// @param[in] nb_elements : size of your for loop
/// @param[in] functor(start, end) :
/// your function processing a sub chunk of the for loop.
/// "start" is the first index to process (included) until the index "end"
/// (excluded)
/// @code
///     for(int i = start; i < end; ++i)
///         computation(i);
/// @endcode
/// @param use_threads : enable / disable threads.
///
///
static
void parallel_for(unsigned nb_elements,
                  std::function<void (int start, int end)> functor,
                  bool use_threads = true)
{
    // -------
    unsigned nb_threads_hint = std::thread::hardware_concurrency();
    unsigned nb_threads = nb_threads_hint == 0 ? 8 : (nb_threads_hint);

    unsigned batch_size = nb_elements / nb_threads;
    unsigned batch_remainder = nb_elements % nb_threads;

    std::vector< std::thread > my_threads(nb_threads);

    if( use_threads )
    {
        // Multithread execution
        for(unsigned i = 0; i < nb_threads; ++i)
        {
            int start = i * batch_size;
            my_threads[i] = std::thread(functor, start, start+batch_size);
        }
    }
    else
    {
        // Single thread execution (for easy debugging)
        for(unsigned i = 0; i < nb_threads; ++i){
            int start = i * batch_size;
            functor( start, start+batch_size );
        }
    }

    // Deform the elements left
    int start = nb_threads * batch_size;
    functor( start, start+batch_remainder);

    // Wait for the other thread to finish their task
    if( use_threads )
        std::for_each(my_threads.begin(), my_threads.end(), std::mem_fn(&std::thread::join));
}

Lastly you can define macros to get even more compact expression:

#define PARALLEL_FOR_BEGIN(nb_elements) parallel_for(nb_elements, [&](int start, int end){ for(int i = start; i < end; ++i)
#define PARALLEL_FOR_END()})

Now converting a sequential for:

for(int i = 0; i < nb_elements; ++i)
    computation(i);

Is only a matter of doing:

PARALLEL_FOR_BEGIN(nb_edges)
{
    computation(i);
}PARALLEL_FOR_END();
1

std::async may be a good fit here, if you are happy to let the C++ runtime control the parallelism.

Example from the cppreference.com:

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <future>

template <typename RAIter>
int parallel_sum(RAIter beg, RAIter end)
{
    auto len = end - beg;
    if(len < 1000)
        return std::accumulate(beg, end, 0);

    RAIter mid = beg + len/2;
    auto handle = std::async(std::launch::async,
                              parallel_sum<RAIter>, mid, end);
    int sum = parallel_sum(beg, mid);
    return sum + handle.get();
}

int main()
{
    std::vector<int> v(10000, 1);
    std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << '\n';
}
6

This can be done using threads specifically pthreads library function that can be used to perform operations concurrently.

You can read more about them here :

std::thread can also be used :

Below is a code in which i use the thread id of each thread to split the array into two halves :

#include <iostream>
#include <cstdlib>
#include <pthread.h>

using namespace std;

#define NUM_THREADS 2

int arr[10] = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};

void *splitLoop(void *threadid)
{
   long tid;
   tid = (long)threadid;
   //cout << "Hello World! Thread ID, " << tid << endl;
   int start = (tid * 5);
   int end = start + 5;
   for(int i = start;i < end;i++){
      cout << arr[i] << " ";
   }
   cout << endl;
   pthread_exit(NULL);
}

int main ()
{
   pthread_t threads[NUM_THREADS];
   int rc;
   int i;
   for( i=0; i < NUM_THREADS; i++ ){
      cout << "main() : creating thread, " << i << endl;
      rc = pthread_create(&threads[i], NULL, 
                          splitLoop, (void *)i);
      if (rc){
         cout << "Error:unable to create thread," << rc << endl;
         exit(-1);
      }
   }
   pthread_exit(NULL);
}

Also remember while compiling you have to use the -lpthread flag.

Link to solution on Ideone :

3

As this thread has been my answer almost every time I've looked for a method to parallelize something, I've decided to add a bit to it, based on the method by arkan (see his answer).

The two following methods are almost the same and allow a simple syntax. Simply include the header file in your project and call one of the parallel version:

Example:

#include "par_for.h"

int main() {
//replace - 
for(unsigned i = 0; i < 10; ++i){
    std::cout << i << std::endl;
}

//with -
//method 1:
pl::thread_par_for(0, 10, [&](unsigned i){
            std::cout << i << std::endl;   //do something here with the index i
        });   //changing the end to },false); will make the loop sequential

//or method 2:
pl::async_par_for(0, 10, [&](unsigned i){
            std::cout << i << std::endl;   //do something here with the index i
        });   //changing the end to },false); will make the loop sequential

return 0;
}

header file - par_for.h:

#include <thread>
#include <vector>
#include <functional>
#include <future>

namespace pl{

    void thread_par_for(unsigned start, unsigned end, std::function<void(unsigned i)> fn, bool par = true){

        //internal loop
        auto int_fn = [&fn](unsigned int_start, unsigned seg_size){
            for (unsigned j = int_start; j < int_start+seg_size; j++){
                fn(j);
            }
        };

        //sequenced for
        if(!par){
            return int_fn(start, end);
        }

        //get number of threads
        unsigned nb_threads_hint = std::thread::hardware_concurrency();
        unsigned nb_threads = nb_threads_hint == 0 ? 8 : (nb_threads_hint);

        //calculate segments
        unsigned total_length = end - start;
        unsigned seg = total_length/nb_threads;
        unsigned last_seg = seg + total_length%nb_threads;

        //launch threads - parallel for
        auto threads_vec = std::vector<std::thread>();
        threads_vec.reserve(nb_threads);
        for(int k = 0; k < nb_threads-1; ++k){
            unsigned current_start = seg*k;
            threads_vec.emplace_back(std::thread(int_fn, current_start, seg));
        }
        {
            unsigned current_start = seg*(nb_threads-1);
            threads_vec.emplace_back(std::thread(int_fn, current_start, last_seg));
        }
        for (auto& th : threads_vec){
            th.join();
        }
    }




    void async_par_for(unsigned start, unsigned end, std::function<void(unsigned i)> fn, bool par = true){

        //internal loop
        auto int_fn = [&fn](unsigned int_start, unsigned seg_size){
            for (unsigned j = int_start; j < int_start+seg_size; j++){
                fn(j);
            }
        };

        //sequenced for
        if(!par){
            return int_fn(start, end);
        }

        //get number of threads
        unsigned nb_threads_hint = std::thread::hardware_concurrency();
        unsigned nb_threads = nb_threads_hint == 0 ? 8 : (nb_threads_hint);

        //calculate segments
        unsigned total_length = end - start;
        unsigned seg = total_length/nb_threads;
        unsigned last_seg = seg + total_length%nb_threads;

        //launch threads - parallel for
        auto fut_vec = std::vector<std::future<void>>();
        fut_vec.reserve(nb_threads);
        for(int k = 0; k < nb_threads-1; ++k){
            unsigned current_start = seg*k;
            fut_vec.emplace_back(async(int_fn, current_start, seg));
        }
        {
            unsigned current_start = seg*(nb_threads-1);
            fut_vec.emplace_back(std::async(std::launch::async, int_fn, current_start, last_seg));
        }
        for (auto& th : fut_vec){
            th.get();
        }
    }
}

Some simple tests suggest the method with async is faster, probably because the standard library controls whether to actually launch a new thread or not.

The Concurrency::parallel_for (PPL) is also one of the nice opions to do task parallelism.

Taken from C++ Coding Exercise – Parallel For – Monte Carlo PI Calculation

int main() {
    srand(time(NULL)); // seed
    const int N1 = 1000;
    const int N2 = 100000;
    int n = 0;
    int c = 0;
    Concurrency::critical_section cs;
    // it is better that N2 >> N1 for better performance
    Concurrency::parallel_for(0, N1, [&](int i) {
        int t = monte_carlo_count_pi(N2);
        cs.lock(); // race condition
        n += N2;   // total sampling points
        c += t;    // points fall in the circle
        cs.unlock();
    });
    cout < < "pi ~= " << setprecision(9) << (double)c / n * 4.0 << endl;
    return 0;
}
1

Starting from C++17 std::for_each has overloads that allow parallel execution. In my case, however, my algorithm required specific number of threads for optimal execution, while std::for_each implementation from VS 2022 uses number of threads based on std::thread::hardware_concurrency.

For those who want to be able to control number of parallel workers this simple implementation should behave similarly to std::for_each without requiring to have C++17:

template <class Iter, class Func>
void parallel_for_each(unsigned threadCount, Iter first, Iter last, Func func)
{
    Iter it = first;
    if (it == last)
        return;
    if (++it == last)
    {
        func(*first);
        return;
    }

    if (threadCount == 0)
        threadCount = std::max(2u, std::thread::hardware_concurrency());

    std::mutex mx;
    std::vector<std::thread> threads;
    threads.reserve(threadCount - 1);

    auto func2 = [&]() {
        for (;;)
        {
            Iter it;
            {
                std::lock_guard<std::mutex> lock(mx);
                it = first;
                if (it == last)
                    break;
                ++first;
            }
            func(*it);
        }
    };
    for (unsigned i = 0; i < threadCount - 1; ++i, ++it)
    {
        if (it == last)
            break;
        threads.emplace_back(std::thread(func2));
    }
    func2();
    for (auto& th : threads)
        th.join();
}

template <class Iter, class Func>
void parallel_for_each(Iter first, Iter last, Func func)
{
    parallel_for_each(std::thread::hardware_concurrency(), first, last, func);
}

Your Answer

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

James H. Sterling
Author

James H. Sterling

James Sterling reports on renewable energy developments, climate policy, ecological conservation, and green tech innovations around the globe.