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) {
    // This code mirrors the return logic in the default parallelize_range(), but would be an upper bound even with a custom SUBPAR_CUSTOM_PARALLELIZE_RANGE.
    // Remember that run_task_range must be called with a non-empty range so if num_tasks = 0, there is no choice but to not perform any calls.
    // Similarly, we can't perform more than one call if num_tasks = 1, and we can't perform more calls than there are workers.
 
    if (num_tasks <= 0) {
        return 0;
    }
 
    if (num_workers <= 1 || num_tasks == 1) {
        return 1;
    }
 
    return sanisizer::min(num_workers, num_tasks);
}
 
template<bool nothrow_ = false, typename Task_, class Run_>
int 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
        return SUBPAR_CUSTOM_PARALLELIZE_RANGE_NOTHROW(num_workers, num_tasks, run_task_range);
#else
        return SUBPAR_CUSTOM_PARALLELIZE_RANGE(num_workers, num_tasks, run_task_range);
#endif
    } else {
        return SUBPAR_CUSTOM_PARALLELIZE_RANGE(num_workers, num_tasks, run_task_range);
    }
 
#else
    if (num_tasks <= 0) {
        return 0;
    }
 
    if (num_workers <= 1 || num_tasks == 1) {
        run_task_range(0, 0, num_tasks);
        return 1;
    }
 
    // 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;
    }
 
    const auto get_start = [&tasks_per_worker,&remainder](const int w) -> Task_ {
        // Need to shift the start by the number of previous 'w' that added a remainder.
        return w * tasks_per_worker + (w < remainder ? w : remainder);
    };
 
    const auto get_length = [&tasks_per_worker,&remainder](const int w) -> Task_ {
        return tasks_per_worker + (w < remainder); 
    };
 
    // 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 = get_start(w);
        const Task_ length = get_length(w);
 
        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
 
    // We run the first job on the current thread, to avoid having to spin up an unnecessary worker.
    std::vector<std::thread> workers;
    sanisizer::reserve(workers, num_workers - 1); // preallocate to ensure we don't get alloc errors during emplace_back().
 
    for (int w = 1; w < num_workers; ++w) {
        const Task_ start = get_start(w);
        const Task_ length = get_length(w);
 
        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);
        }
    }
 
    {
        const Task_ start = get_start(0);
        const Task_ length = get_length(0);
 
        if constexpr(nothrow_) {
            run_task_range(0, start, length);
        } else {
            try {
                run_task_range(0, start, length);
            } catch (...) {
                errors[0] = std::current_exception();
            }
        }
    }
 
    for (auto& wrk : workers) {
        wrk.join();
    }
#endif
 
    if constexpr(!nothrow_) {
        for (const auto& e : errors) {
            if (e) {
                std::rethrow_exception(e);
            }
        }
    }
#endif
 
    return num_workers;
}
 
// 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