ConcurrencyUtils.h

Aller à la documentation de ce fichier.

// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
//-----------------------------------------------------------------------------
// Copyright 2000-2024 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
// See the top-level COPYRIGHT file for details.
// SPDX-License-Identifier: Apache-2.0
//-----------------------------------------------------------------------------
/*---------------------------------------------------------------------------*/
/* ConcurrencyUtils.h                                          (C) 2000-2024 */
/*                                                                           */
/* Classes gérant la concurrence (tâches, boucles parallèles, ...)           */
/*---------------------------------------------------------------------------*/
#ifndef ARCANE_UTILS_CONCURRENCYUTILS_H
#define ARCANE_UTILS_CONCURRENCYUTILS_H
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#include "arcane/utils/UtilsTypes.h"
#include "arcane/utils/RangeFunctor.h"
#include "arcane/utils/FatalErrorException.h"
#include "arcane/utils/ForLoopTraceInfo.h"
#include "arcane/utils/ParallelLoopOptions.h"
 
#include <optional>
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
namespace Arcane
{
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
/*
 * TODO:
 * - Vérifier les fuites memoires
 * - BIEN INDIQUER QU'IL NE FAUT PLUS UTILISER UNE TACHE APRES LE WAIT!!!
 * - Regarder mecanisme pour les exceptions.
 * - Surcharger les For et Foreach sans specifier le block_size
 */
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT TaskContext
{
 public:
  explicit TaskContext(ITask* atask) : m_task(atask) {}
 public:
  ITask* task() const { return m_task; }
 private:
  ITask* m_task;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT ITaskFunctor
{
 public:
  virtual ~ITaskFunctor() = default;
 protected:
  ITaskFunctor(const ITaskFunctor&) = default;
  ITaskFunctor() = default;
 public:
  virtual void executeFunctor(const TaskContext& tc) =0;
  virtual ITaskFunctor* clone(void* buffer,Integer size) =0;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<typename InstanceType>
class TaskFunctor
: public ITaskFunctor
{
 public:
  typedef void (InstanceType::*FunctorType)();
 public:
  TaskFunctor(InstanceType* instance,FunctorType func)
  : m_instance(instance), m_function(func)
  {
  }
  TaskFunctor(const TaskFunctor& rhs) = default;
  TaskFunctor& operator=(const TaskFunctor& rhs) = delete;
 public:
  void executeFunctor(const TaskContext& /*tc*/) override
  {
    (m_instance->*m_function)();
  }
  ITaskFunctor* clone(void* buffer,Integer size) override
  {
    if (sizeof(*this)>(size_t)size)
      ARCANE_FATAL("INTERNAL: task functor buffer is too small");
    return new (buffer) TaskFunctor<InstanceType>(*this);
  }
 private:
  InstanceType* m_instance;
  FunctorType m_function;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<typename InstanceType>
class TaskFunctorWithContext
: public ITaskFunctor
{
 public:
  typedef void (InstanceType::*FunctorType)(const TaskContext& tc);
 public:
  TaskFunctorWithContext(InstanceType* instance,FunctorType func)
  : ITaskFunctor(), m_instance(instance), m_function(func)
  {
  }
 public:
  void executeFunctor(const TaskContext& tc) override
  {
    (m_instance->*m_function)(tc);
  }
  ITaskFunctor* clone(void* buffer,Integer size) override
  {
    if (sizeof(*this)>(size_t)size)
      ARCANE_FATAL("INTERNAL: task functor buffer is too small");
    return new (buffer) TaskFunctorWithContext<InstanceType>(*this);
  }
 private:
  InstanceType* m_instance;
  FunctorType m_function;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT ITask
{
  friend class TaskFactory;
 
 public:
 
  virtual ~ITask(){}
 
 public:
  virtual void launchAndWait() =0;
  virtual void launchAndWait(ConstArrayView<ITask*> tasks) =0;
 
 protected:
 
  virtual ITask* _createChildTask(ITaskFunctor* functor) =0;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT ITaskImplementation
{
 public:
  virtual ~ITaskImplementation(){}
 public:
  virtual void initialize(Int32 nb_thread) =0;
  virtual void terminate() =0;
  virtual ITask* createRootTask(ITaskFunctor* f) =0;
 
  virtual void executeParallelFor(Integer begin,Integer size,const ParallelLoopOptions& options,IRangeFunctor* f) =0;
 
  virtual void executeParallelFor(Integer begin,Integer size,Integer block_size,IRangeFunctor* f) =0;
 
  virtual void executeParallelFor(Integer begin,Integer size,IRangeFunctor* f) =0;
 
  virtual void executeParallelFor(const ParallelFor1DLoopInfo& loop_info) =0;
 
  virtual void executeParallelFor(const ComplexForLoopRanges<1>& loop_ranges,
                                  const ParallelLoopOptions& options,
                                  IMDRangeFunctor<1>* functor) =0;
  virtual void executeParallelFor(const ComplexForLoopRanges<2>& loop_ranges,
                                  const ParallelLoopOptions& options,
                                  IMDRangeFunctor<2>* functor) =0;
  virtual void executeParallelFor(const ComplexForLoopRanges<3>& loop_ranges,
                                  const ParallelLoopOptions& options,
                                  IMDRangeFunctor<3>* functor) =0;
  virtual void executeParallelFor(const ComplexForLoopRanges<4>& loop_ranges,
                                  const ParallelLoopOptions& options,
                                  IMDRangeFunctor<4>* functor) =0;
 
  virtual bool isActive() const =0;
 
  virtual Int32 nbAllowedThread() const =0;
 
  virtual Int32 currentTaskThreadIndex() const =0;
 
  virtual Int32 currentTaskIndex() const =0;
 
  virtual void printInfos(std::ostream& o) const =0;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT TaskFactory
{
 private:
  TaskFactory();
 public:
 public:
 
  template<typename InstanceType> static ITask*
  createTask(InstanceType* instance,void (InstanceType::*function)(const TaskContext& tc))
  {
    TaskFunctorWithContext<InstanceType> functor(instance,function);
    return m_impl->createRootTask(&functor);
  }
 
  template<typename InstanceType> static ITask*
  createTask(InstanceType* instance,void (InstanceType::*function)())
  {
    TaskFunctor<InstanceType> functor(instance,function);
    return m_impl->createRootTask(&functor);
  }
 
  template<typename InstanceType> static ITask*
  createChildTask(ITask* parent_task,InstanceType* instance,void (InstanceType::*function)(const TaskContext& tc))
  {
    ARCANE_CHECK_POINTER(parent_task);
    TaskFunctorWithContext<InstanceType> functor(instance,function);
    return parent_task->_createChildTask(&functor);
  }
 
  template<typename InstanceType> static ITask*
  createChildTask(ITask* parent_task,InstanceType* instance,void (InstanceType::*function)())
  {
    ARCANE_CHECK_POINTER(parent_task);
    TaskFunctor<InstanceType> functor(instance,function);
    return parent_task->_createChildTask(&functor);
  }
 
  static void executeParallelFor(Integer begin,Integer size,const ParallelLoopOptions& options,IRangeFunctor* f)
  {
    m_impl->executeParallelFor(begin,size,options,f);
  }
 
  static void executeParallelFor(Integer begin,Integer size,Integer block_size,IRangeFunctor* f)
  {
    m_impl->executeParallelFor(begin,size,block_size,f);
  }
 
  static void executeParallelFor(Integer begin,Integer size,IRangeFunctor* f)
  {
    m_impl->executeParallelFor(begin,size,f);
  }
 
  static void executeParallelFor(const ParallelFor1DLoopInfo& loop_info)
  {
    m_impl->executeParallelFor(loop_info);
  }
 
  static void executeParallelFor(const ComplexForLoopRanges<1>& loop_ranges,
                                 const ParallelLoopOptions& options,
                                 IMDRangeFunctor<1>* functor)
  {
    m_impl->executeParallelFor(loop_ranges,options,functor);
  }
 
  static void executeParallelFor(const ComplexForLoopRanges<2>& loop_ranges,
                                 const ParallelLoopOptions& options,
                                 IMDRangeFunctor<2>* functor)
  {
    m_impl->executeParallelFor(loop_ranges,options,functor);
  }
 
  static void executeParallelFor(const ComplexForLoopRanges<3>& loop_ranges,
                                 const ParallelLoopOptions& options,
                                 IMDRangeFunctor<3>* functor)
  {
    m_impl->executeParallelFor(loop_ranges,options,functor);
  }
 
  static void executeParallelFor(const ComplexForLoopRanges<4>& loop_ranges,
                                 const ParallelLoopOptions& options,
                                 IMDRangeFunctor<4>* functor)
  {
    m_impl->executeParallelFor(loop_ranges,options,functor);
  }
 
  static Int32 nbAllowedThread()
  {
    return m_impl->nbAllowedThread();
  }
 
  static Int32 currentTaskThreadIndex()
  {
    return m_impl->currentTaskThreadIndex();
  }
 
  static Int32 currentTaskIndex()
  {
    return m_impl->currentTaskIndex();
  }
 
 public:
 
  static void setDefaultParallelLoopOptions(const ParallelLoopOptions& v)
  {
    m_default_loop_options = v;
  }
 
  static const ParallelLoopOptions& defaultParallelLoopOptions()
  {
    return m_default_loop_options;
  }
 
 public:
 
  static bool isActive()
  {
    return m_impl->isActive();
  }
 
  static void printInfos(std::ostream& o)
  {
    return m_impl->printInfos(o);
  }
 
  static IObservable* createThreadObservable();
 
  static IObservable* destroyThreadObservable();
 
  static void terminate();
 
 public:
 
  static void setVerboseLevel(Integer v) { m_verbose_level = v; }
 
  static Integer verboseLevel() { return m_verbose_level; }
 
 public:
 
  static void _internalSetImplementation(ITaskImplementation* task_impl);
 
 private:
 
  static ITaskImplementation* m_impl;
  static IObservable* m_created_thread_observable;
  static IObservable* m_destroyed_thread_observable;
  static Int32 m_verbose_level;
  static ParallelLoopOptions m_default_loop_options;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT ForLoopRunInfo
{
 public:
 
  using ThatClass = ForLoopRunInfo;
 
 public:
 
  ForLoopRunInfo() = default;
  explicit ForLoopRunInfo(const ParallelLoopOptions& options)
  : m_options(options) {}
  ForLoopRunInfo(const ParallelLoopOptions& options,const ForLoopTraceInfo& trace_info)
  : m_options(options), m_trace_info(trace_info) {}
  explicit ForLoopRunInfo(const ForLoopTraceInfo& trace_info)
  : m_trace_info(trace_info) {}
 
 public:
 
  std::optional<ParallelLoopOptions> options() const { return m_options; }
  ThatClass& addOptions(const ParallelLoopOptions& v) { m_options = v; return (*this); }
  const ForLoopTraceInfo& traceInfo() const { return m_trace_info; }
  ThatClass& addTraceInfo(const ForLoopTraceInfo& v) { m_trace_info = v; return (*this); }
 
  void setExecStat(ForLoopOneExecStat* v) { m_exec_stat = v; }
 
  ForLoopOneExecStat* execStat() const { return m_exec_stat; }
 
 protected:
 
  std::optional<ParallelLoopOptions> m_options;
  ForLoopTraceInfo m_trace_info;
  ForLoopOneExecStat* m_exec_stat = nullptr;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
class ARCANE_UTILS_EXPORT ParallelFor1DLoopInfo
{
 public:
 
 using ThatClass = ParallelFor1DLoopInfo;
 
 public:
 
  ParallelFor1DLoopInfo(Int32 begin,Int32 size,IRangeFunctor* functor)
  : m_begin(begin), m_size(size), m_functor(functor) {}
  ParallelFor1DLoopInfo(Int32 begin,Int32 size,IRangeFunctor* functor,const ForLoopRunInfo& run_info)
  : m_run_info(run_info), m_begin(begin), m_size(size), m_functor(functor) {}
  ParallelFor1DLoopInfo(Int32 begin,Int32 size, Int32 block_size,IRangeFunctor* functor)
  : m_begin(begin), m_size(size), m_functor(functor)
  {
    ParallelLoopOptions opts(TaskFactory::defaultParallelLoopOptions());
    opts.setGrainSize(block_size);
    m_run_info.addOptions(opts);
  }
 
 public:
 
  Int32 beginIndex() const { return m_begin; }
  Int32 size() const { return m_size; }
  IRangeFunctor* functor() const { return m_functor; }
  ForLoopRunInfo& runInfo() { return m_run_info; }
  const ForLoopRunInfo& runInfo() const { return m_run_info; }
 
 private:
 
  ForLoopRunInfo m_run_info;
  Int32 m_begin = 0;
  Int32 m_size = 0;
  IRangeFunctor* m_functor = nullptr;
};
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<int RankValue,typename LambdaType,typename... ReducerArgs> inline void
arcaneParallelFor(const ComplexForLoopRanges<RankValue>& loop_ranges,
                  const ParallelLoopOptions& options,
                  const LambdaType& lambda_function,
                  const ReducerArgs&... reducer_args)
{
  // Modif Arcane 3.7.9 (septembre 2022)
  // Effectue une copie pour privatiser au thread courant les valeurs de la lambda.
  // Cela est nécessaire pour que objets comme les reducers soient bien pris
  // en compte.
  // TODO: regarder si on pourrait faire la copie uniquement une fois par thread
  // si cette copie devient couteuse.
  // NOTE: A partir de la version 3.12.15 (avril 2024), avec la nouvelle version
  // des réducteurs (Reduce2), cette privatisation n'est plus utile. Une fois
  // qu'on aura supprimer les anciennes classes gérant les réductions (Reduce),
  // on pourra supprimer cette privatisation
  auto xfunc = [&lambda_function,reducer_args...] (const ComplexForLoopRanges<RankValue>& sub_bounds)
  {
    using Type = typename std::remove_reference<LambdaType>::type;
    Type private_lambda(lambda_function);
    arcaneSequentialFor(sub_bounds,private_lambda,reducer_args...);
  };
  LambdaMDRangeFunctor<RankValue,decltype(xfunc)> ipf(xfunc);
  TaskFactory::executeParallelFor(loop_ranges,options,&ipf);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template <int RankValue, typename LambdaType, typename... ReducerArgs> inline void
arcaneParallelFor(const SimpleForLoopRanges<RankValue>& loop_ranges,
                  const ParallelLoopOptions& options,
                  const LambdaType& lambda_function,
                  const ReducerArgs&... reducer_args)
{
  ComplexForLoopRanges<RankValue> complex_loop_ranges{ loop_ranges };
  arcaneParallelFor(complex_loop_ranges, options, lambda_function, reducer_args...);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<int RankValue,typename LambdaType> inline void
arcaneParallelFor(const ComplexForLoopRanges<RankValue>& loop_ranges,
                  const LambdaType& lambda_function)
{
  ParallelLoopOptions options;
  arcaneParallelFor(loop_ranges,options,lambda_function);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
template<int RankValue,typename LambdaType> inline void
arcaneParallelFor(const SimpleForLoopRanges<RankValue>& loop_ranges,
                  const LambdaType& lambda_function)
{
  ParallelLoopOptions options;
  ComplexForLoopRanges<RankValue> complex_loop_ranges{loop_ranges};
  arcaneParallelFor(complex_loop_ranges,options,lambda_function);
}
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
} // End namespace Arcane
 
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
 
#endif