parallelize.hpp

Go to the documentation of this file.

#ifndef TATAMI_R_PARALLELIZE_HPP
#define TATAMI_R_PARALLELIZE_HPP
 
#ifdef TATAMI_R_PARALLELIZE_UNKNOWN
#include "manticore/manticore.hpp"
#include <thread>
#include <cmath>
#include <vector>
#include <string>
#include <stdexcept>
#include <algorithm>
 
namespace tatami_r {
 
inline manticore::Executor& executor() {
    // This should end up resolving to a single instance, even across dynamically linked libraries:
    // https://stackoverflow.com/questions/52851239/local-static-variable-linkage-in-a-template-class-static-member-function
    static manticore::Executor mexec;
    return mexec;
}
 
template<class Function_, class Index_>
void parallelize(Function_ fun, Index_ ntasks, int nthreads) {
    if (ntasks == 0) {
        return;
    }
 
    if (nthreads <= 1 || ntasks == 1) {
        fun(0, 0, ntasks);
        return;
    }
 
    Index_ tasks_per_worker = ntasks / nthreads;
    int remainder = ntasks % nthreads;
    if (tasks_per_worker == 0) {
        tasks_per_worker = 1; 
        remainder = 0;
        nthreads = ntasks;
    }
 
    auto& mexec = executor();
    mexec.initialize(nthreads, "failed to execute R command");
 
    std::vector<std::thread> runners;
    runners.reserve(nthreads);
    std::vector<std::exception_ptr> errors(nthreads);
 
    Index_ start = 0;
    for (int w = 0; w < nthreads; ++w) {
        Index_ length = tasks_per_worker + (w < remainder);
 
        runners.emplace_back([&](int id, Index_ s, Index_ l) {
            try {
                fun(id, s, l);
            } catch (...) {
                errors[id] = std::current_exception();
            }
            mexec.finish_thread();
        }, w, start, length);
 
        start += length;
    }
 
    mexec.listen();
    for (auto& x : runners) {
        x.join();
    }
 
    for (const auto& err : errors) {
        if (err) {
            std::rethrow_exception(err);
        }
    }
}
 
}
 
#endif
#endif