range.hpp

Go to the documentation of this file.

#ifndef SUBPAR_RANGE_HPP
#define SUBPAR_RANGE_HPP
 
#include <limits>
#include <type_traits>
 
#ifndef SUBPAR_CUSTOM_PARALLELIZE_RANGE
#include <vector>
#include <stdexcept>
#include <thread>
#endif
 
#include "sanisizer/sanisizer.hpp"
 
namespace subpar {
 
template<typename Task_>
int sanitize_num_workers(const int num_workers, const Task_ num_tasks) {
    if (num_workers <= 0) {
        return num_tasks > 0;
    } else {
        return sanisizer::min(num_workers, num_tasks);
    }
}
 
template<bool nothrow_ = false, typename Task_, class Run_>
void parallelize_range(int num_workers, const Task_ num_tasks, const Run_ run_task_range) {
#ifdef SUBPAR_CUSTOM_PARALLELIZE_RANGE
    if constexpr(nothrow_) {
#ifdef SUBPAR_CUSTOM_PARALLELIZE_RANGE_NOTHROW
        SUBPAR_CUSTOM_PARALLELIZE_RANGE_NOTHROW(num_workers, num_tasks, run_task_range);
#else
        SUBPAR_CUSTOM_PARALLELIZE_RANGE(num_workers, num_tasks, run_task_range);
#endif
    } else {
        SUBPAR_CUSTOM_PARALLELIZE_RANGE(num_workers, num_tasks, run_task_range);
    }
 
#else
    if (num_tasks == 0) {
        return;
    }
 
    if (num_workers <= 1 || num_tasks == 1) {
        run_task_range(0, 0, num_tasks);
        return;
    }
 
    // All workers with indices below 'remainder' get an extra task to fill up the remainder.
    Task_ tasks_per_worker = 1;
    int remainder = 0;
    if (sanisizer::is_greater_than_or_equal(num_workers, num_tasks)) {
        num_workers = num_tasks;
    } else {
        tasks_per_worker = num_tasks / num_workers;
        remainder = num_tasks % num_workers;
    }
 
    // Avoid instantiating a vector if it is known that the function can't throw.
    auto errors = [&]{
        if constexpr(nothrow_) {
            return true;
        } else {
            return sanisizer::create<std::vector<std::exception_ptr> >(num_workers);
        }
    }();
 
#if defined(_OPENMP) && !defined(SUBPAR_NO_OPENMP_RANGE) && !defined(SUBPAR_NO_OPENMP)
#define SUBPAR_USES_OPENMP 1
#define SUBPAR_USES_OPENMP_RANGE 1
 
    // OpenMP doesn't guarantee that we'll actually start the specified number of workers,
    // so we need to do a loop here to ensure that each task range is executed.
    #pragma omp parallel for num_threads(num_workers)
    for (int w = 0; w < num_workers; ++w) {
        const Task_ start = w * tasks_per_worker + (w < remainder ? w : remainder); // need to shift the start by the number of previous 'w' that added a remainder.
        const Task_ length = tasks_per_worker + (w < remainder);
 
        if constexpr(nothrow_) {
            run_task_range(w, start, length);
        } else {
            try { 
                run_task_range(w, start, length);
            } catch (...) {
                errors[w] = std::current_exception();
            }
        }
    }
 
#else
// Wiping it out, just in case.
#undef SUBPAR_USES_OPENMP
#undef SUBPAR_USES_OPENMP_RANGE
 
    Task_ start = 0;
    std::vector<std::thread> workers;
    workers.reserve(sanisizer::cast<decltype(workers.size())>(num_workers)); // preallocate to ensure we don't get alloc errors during emplace_back().
 
    for (int w = 0; w < num_workers; ++w) {
        const Task_ length = tasks_per_worker + (w < remainder); 
 
        if constexpr(nothrow_) {
            workers.emplace_back(run_task_range, w, start, length);
        } else {
            workers.emplace_back([&run_task_range,&errors](int w, Task_ start, Task_ length) -> void {
                try {
                    run_task_range(w, start, length);
                } catch (...) {
                    errors[w] = std::current_exception();
                }
            }, w, start, length);
        }
 
        start += length;
    }
 
    for (auto& wrk : workers) {
        wrk.join();
    }
#endif
 
    if constexpr(!nothrow_) {
        for (const auto& e : errors) {
            if (e) {
                std::rethrow_exception(e);
            }
        }
    }
#endif
}
 
// Back-compatibility only.
template<typename Task_, class Run_>
void parallelize(int num_workers, Task_ num_tasks, Run_ run_task_range) {
    parallelize_range<false, Task_, Run_>(num_workers, num_tasks, std::move(run_task_range));
}
}
 
#endif