parallel.hh

Go to the documentation of this file.

#ifndef UTOPIA_CORE_PARALLEL_HH
#define UTOPIA_CORE_PARALLEL_HH
 
// Include one of the available headers, preferring the external one
#if defined(HAVE_ONEDPL)
#include <oneapi/dpl/execution>
#include <oneapi/dpl/algorithm>
#elif defined(USE_INTERNAL_PSTL)
#include <execution>
#endif
 
// PSTL is available
#if defined(HAVE_ONEDPL) || defined(USE_INTERNAL_PSTL)
#define HAVE_PSTL
#endif
 
// PSTL is available and parallelism is enabled
#if defined(HAVE_PSTL) && defined(ENABLE_PARALLEL_STL)
#define UTOPIA_PARALLEL // Now we're cookin'!
#define MAYBE_UNUSED    // Make sure to get proper warnings for unused parameter
 
// Inject oneDPL execution policies into std namespace to have agnostic client
// code
#if defined(HAVE_ONEDPL)
namespace std::execution {
    using namespace oneapi::dpl::execution;
}
#endif
 
#else
#define MAYBE_UNUSED [[maybe_unused]]  // Avoid warning for unused parameter
#endif
 
#include <algorithm>
#include <exception>
#include <tuple>
 
#include <utopia/core/logging.hh>
#include <utopia/data_io/cfg_utils.hh>
 
namespace Utopia
{
 
 
enum ExecPolicy
{
 
    seq,
    unseq,    
    par,      
    par_unseq 
};
 
 
class ParallelExecution
{
private:
    inline static bool _enabled = false;
 
 
    static std::shared_ptr<spdlog::logger> get_logger()
    {
        const auto log = spdlog::get(Utopia::log_core);
        if (not log)
        {
            throw std::runtime_error("Cannot fetch core logger!");
        }
 
        return log;
    }
 
public:
    enum Setting
    {
        enabled, 
        disabled 
    };
 
 
    static void init(const DataIO::Config& cfg)
    {
        bool setting = false;
 
        // Try fetching settings on parallel execution
        if (const YAML::Node& cfg_par = cfg["parallel_execution"])
        {
            setting = get_as<bool>("enabled", cfg_par);
        }
 
        if (setting)
            set(Setting::enabled);
        else
            set(Setting::disabled);
    }
 
 
    static void set(const Setting value)
    {
        _enabled = (value == Setting::enabled);
 
        const auto log = get_logger();
        std::string msg = fmt::format("Parallel execution {}",
                                      _enabled ? "enabled" : "disabled");
 
#ifdef UTOPIA_PARALLEL
        log->info(msg);
#else
        msg += ", but settings do NOT apply";
        if (_enabled)
            log->warn(msg);
        else
            log->debug(msg);
#endif
    }
 
 
    static bool is_enabled() { return _enabled; }
 
 
    static bool is_applied()
    {
#ifdef UTOPIA_PARALLEL
        return _enabled;
#else
        return false;
#endif
    }
};
 
 
template<class Func, class... Args>
auto
exec_parallel(MAYBE_UNUSED const Utopia::ExecPolicy policy,
              Func&& f,
              Args&&... args)
{
#ifdef UTOPIA_PARALLEL
    if (Utopia::ParallelExecution::is_enabled())
    {
        if (policy == Utopia::ExecPolicy::unseq)
            return f(std::forward_as_tuple(std::execution::unseq, args...));
        else if (policy == Utopia::ExecPolicy::par)
            return f(std::forward_as_tuple(std::execution::par, args...));
        else if (policy == Utopia::ExecPolicy::par_unseq)
            return f(std::forward_as_tuple(std::execution::par_unseq, args...));
    }
#endif
 
#ifdef HAVE_PSTL
    return f(std::forward_as_tuple(std::execution::seq, args...));
#else
    return f(std::forward_as_tuple(args...));
#endif
}
 
} // namespace Utopia
 
namespace std
{
 
 
template<class InputIt, class OutputIt>
OutputIt
copy(const Utopia::ExecPolicy policy,
     InputIt first,
     InputIt last,
     OutputIt d_first)
{
    return Utopia::exec_parallel(
        policy,
        [](auto&& args_tpl) {
            auto copy = [](auto&&... args){ return std::copy(args...); };
            return std::apply(copy, args_tpl);
        },
        first,
        last,
        d_first);
}
 
 
template<class InputIt, class UnaryFunction>
void
for_each(const Utopia::ExecPolicy policy,
         InputIt first,
         InputIt last,
         UnaryFunction f)
{
    Utopia::exec_parallel(
        policy,
        [](auto&& args_tpl) {
            auto for_each = [](auto&&... args){ std::for_each(args...); };
            std::apply(for_each, args_tpl);
        },
        first,
        last,
        f);
}
 
 
template<class InputIt, class OutputIt, class UnaryOperation>
OutputIt
transform(const Utopia::ExecPolicy policy,
          InputIt first1,
          InputIt last1,
          OutputIt d_first,
          UnaryOperation unary_op)
{
    return Utopia::exec_parallel(
        policy,
        [](auto&& args_tpl) {
            auto transform = [](auto&&... args) {
                return std::transform(args...);
            };
            return std::apply(transform, args_tpl);
        },
        first1,
        last1,
        d_first,
        unary_op);
}
 
 
template<class InputIt1, class InputIt2, class OutputIt, class BinaryOperation>
OutputIt
transform(const Utopia::ExecPolicy policy,
          InputIt1 first1,
          InputIt1 last1,
          InputIt2 first2,
          OutputIt d_first,
          BinaryOperation binary_op)
{
    return Utopia::exec_parallel(
        policy,
        [](auto&& args_tpl) {
            auto transform = [](auto&&... args) {
                return std::transform(args...);
            };
            return std::apply(transform, args_tpl);
        },
        first1,
        last1,
        first2,
        d_first,
        binary_op);
}
 
} // namespace std
 
#endif // UTOPIA_CORE_PARALLEL_HH