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 "sanisizer/sanisizer.hpp"
 
#include <thread>
#include <cmath>
#include <vector>
#include <string>
#include <stdexcept>
#include <algorithm>
 
namespace tatami_r {
 
inline manticore::Executor* executor_ptr = NULL;
inline manticore::Executor& executor() {
    if (executor_ptr) {
        return *executor_ptr;
    } else {
        // In theory, 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
        // In practice, this doesn't seem to be the case on a Mac, requiring us to use `set_executor()`.
        static manticore::Executor mexec;
        return mexec;
    }
}
 
inline void set_executor(manticore::Executor* ptr) {
    executor_ptr = ptr;
}
 
template<class Function_, class Index_>
void parallelize(const Function_ fun, const 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);
    auto errors = sanisizer::create<std::vector<std::exception_ptr> >(nthreads);
 
    Index_ start = 0;
    for (int w = 0; w < nthreads; ++w) {
        Index_ length = tasks_per_worker + (w < remainder);
 
        runners.emplace_back(
            [&](const int id, const Index_ s, const Index_ l) -> void {
                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