gmTaskManager.h

Go to the documentation of this file.

/************************************************************************
**
** Copyright (C) 2014 by Carlos Augusto Teixera Mendes
** All rights reserved.
**
** This file is part of the "GeMA" software. It's use should respect
** the terms in the license agreement that can be found together
** with this source code.
** It is provided AS IS, with NO WARRANTY OF ANY KIND,
** INCLUDING THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR
** A PARTICULAR PURPOSE.
**
************************************************************************/
 
#ifndef _GEMA_TASK_MANAGER_H_
#define _GEMA_TASK_MANAGER_H_
 
#include "gmCoreConfig.h"
#include "gmValueInfo.h"
#include "gmThreadManager.h"
 
#include "gmTrace.h"
 
#include <luaEnv.h>
 
//C++ 20 concepts are used to check template parameters and provide better error messages
//For now we are on C++17, but looking forward it is a good practice to use concepts when possible.
#if (defined(_MSVC_LANG) && _MSVC_LANG >= 202002L) || (__cplusplus >= 202002L)
#include <concepts>
#endif
 
class GmSimulationData;
class GmMesh;
class GmCellMesh;
class GmCell;
class GmCellGroupSet;
class GmNodeSet;
 
 
class GMC_API_EXPORT GmNodeTaskDataSet
{
public:
  virtual ~GmNodeTaskDataSet() {}
 
  virtual int nextNode() = 0;
 
  virtual void reset() = 0;
};
 
class GMC_API_EXPORT GmCellTaskDataSet
{
public:
  virtual ~GmCellTaskDataSet() {}
 
  virtual GmCell* nextCell() = 0;
 
  virtual void reset() = 0;
};
 
class GMC_API_EXPORT GmTaskManager
{
public:
 
  enum NodePartitionStrategy
  {
    NPS_DEFAULT          = -1,  
    NPS_SEQUENTIAL_BLOCK =  0,  
    NPS_STRIDE           =  1,  
 
    //-----------------------------
    // No entries should be added after this line
    // IMPORTANT: When adding entries or CHANGING enum order, please remember 
    //            to also change str2NodePartitionStrategy() AND 
    //            GmLuaUtils::checkAndLoadTaskOptions()   
    //-----------------------------
    NPS_NUM_STRATEGIES      
  };
 
  enum CellPartitionStrategy
  {
    CPS_DEFAULT          = -1,  
    CPS_SEQUENTIAL_BLOCK =  0,  
    CPS_STRIDE           =  1,  
    //-----------------------------
    // No entries should be added after this line
    // IMPORTANT: When adding entries or CHANGING enum order, please remember 
    //            to also change str2CellPartitionStrategy() AND 
    //            GmLuaUtils::checkAndLoadTaskOptions()   
    //-----------------------------
    CPS_NUM_STRATEGIES      
  };
 
  GmTaskManager(GmSimulationData* simData, GmThreadManager* tm);
 
  static NodePartitionStrategy str2NodePartitionStrategy(QString str);
  static CellPartitionStrategy str2CellPartitionStrategy(QString str);
 
  GmThreadManager* threadManager() const { return _threadManager; }
 
  int numTasks() const { return _threadManager->numTasks(); }
 
  template <class T> 
  GmThreadTaskResult runParallelNodeLoop(const GmMesh* m, GmAffectedNodes affNodes, T& taskObj,
                                         int ntasks = 0, NodePartitionStrategy strategy = NPS_DEFAULT,
                                         int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  GmThreadTaskResult runParallelNodeLoop(const GmNodeSet* ns, T& taskObj,
                                         int ntasks = 0, NodePartitionStrategy strategy = NPS_DEFAULT,
                                         int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  GmThreadTaskResult runParallelCountLoop(int count, T& taskObj,
                                         int ntasks = 0, NodePartitionStrategy strategy = NPS_DEFAULT,
                                         int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  void               execParallelNodeLoop(const GmMesh* m, GmAffectedNodes affNodes, T& taskObj,
                                          int ntasks = 0, NodePartitionStrategy strategy = NPS_DEFAULT,
                                          int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  void               execParallelNodeLoop(const GmNodeSet* m, T& taskObj,
                                          int ntasks = 0, NodePartitionStrategy strategy = NPS_DEFAULT,
                                          int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  GmThreadTaskResult runParallelCellLoop(GmCellMesh* m, bool activeOnly, T& taskObj,
                                         int ntasks = 0, CellPartitionStrategy strategy = CPS_DEFAULT,
                                         int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  GmThreadTaskResult runParallelCellLoop(GmCellGroupSet* gs, bool activeOnly, T& taskObj,
                                         int ntasks = 0, CellPartitionStrategy strategy = CPS_DEFAULT,
                                         int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class T>
  void               execParallelCellLoop(GmCellMesh* m, bool activeOnly, T& taskObj,
                                          int ntasks = 0, CellPartitionStrategy strategy = CPS_DEFAULT,
                                          int nworkers = -1, const QVector<int>& taskAffinity = QVector<int>());
 
  template <class M, class T> bool addNodeTasks(const M* m, GmAffectedNodes affNodes, T& taskObj,
                                       int ntasks, NodePartitionStrategy strategy, int nworkers,
                                       const QVector<int>& taskAffinity, bool* ok);
 
  template <class M, class T> bool addCellTasks(M* m, bool activeOnly, T& taskObj,
                                                int ntasks, CellPartitionStrategy strategy, int nworkers,
                                                const QVector<int>& taskAffinity, bool* ok);
//       Parallel Accessor  reducttion      //
 
//C++ 20 concepts are used to check template parameters and provide better error messages
//For now we are on C++17, but looking forward it is a good practice to use concepts when possible.
#if (defined(_MSVC_LANG) && _MSVC_LANG >= 202002L) || (__cplusplus >= 202002L)
#include <concepts>
#endif
 
/*
Traverse single accessor, (aggregation)
*/
template<typename Accessor, typename ReduceF, typename JoinF, typename InitV>
  void traverseAccessor(int nworkers, const Accessor* acc, const GmCellMesh* cmesh, const ReduceF& reduce, const JoinF& join, InitV& val);
 
template<typename Accessor, typename ReduceF, typename JoinF, typename InitV>
  void traverseAccessor(int nworkers, const Accessor* acc1, const Accessor* acc2, const GmCellMesh* cmesh, const ReduceF& reduce, const JoinF& join, InitV& val);
 
private:
  void adjustExecParameters(int* ntasks, int* nworkers);
 
  template <class M>
  QVector<GmNodeTaskDataSet*> buildNodePartition(const M* m, int n, GmAffectedNodes affNodes,
                                                 NodePartitionStrategy strategy) const;
 
  template <class M>
  QVector<GmCellTaskDataSet*> buildCellPartition(M* m, int n, bool activeOnly,
                                                 CellPartitionStrategy strategy) const;
 
  GmThreadManager* _threadManager; 
 
  int                    _defNumTasks;      
  NodePartitionStrategy  _defNodeStrategy;  
  CellPartitionStrategy  _defCellStrategy;  
};
 
template <class T>
GmThreadTaskResult GmTaskManager::runParallelNodeLoop(const GmMesh* m, GmAffectedNodes affNodes, T& taskObj,
                                                      int ntasks, NodePartitionStrategy strategy,
                                                      int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addNodeTasks<GmMesh, T>(m, affNodes, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
    return ok ? GM_THREAD_OK : GM_THREAD_ABORTED; // false + ok = no tasks  
 
  // Run tasks!
  return _threadManager->runTasks(nworkers);
}
 
template <class T>
GmThreadTaskResult GmTaskManager::runParallelNodeLoop(const GmNodeSet* ns, T& taskObj,
    int ntasks, NodePartitionStrategy strategy,
    int nworkers, const QVector<int>& taskAffinity)
{
    S_TRACE();
 
    // Add tasks to the manager.
    bool ok;
    if (!addNodeTasks<GmNodeSet, T>(ns, GM_BOTH, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
        return ok ? GM_THREAD_OK : GM_THREAD_ABORTED; // false + ok = no tasks  
 
    // Run tasks!
    return _threadManager->runTasks(nworkers);
}
 
template <class T>
GmThreadTaskResult GmTaskManager::runParallelCountLoop(int count, T& taskObj,
                                                      int ntasks, NodePartitionStrategy strategy,
                                                      int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addNodeTasks<int, T>(&count, GM_BOTH, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
    return ok ? GM_THREAD_OK : GM_THREAD_ABORTED; // false + ok = no tasks  
 
  // Run tasks!
  return _threadManager->runTasks(nworkers);
}
 
template <class T>
void GmTaskManager::execParallelNodeLoop(const GmMesh* m, GmAffectedNodes affNodes, T& taskObj,
                                         int ntasks, NodePartitionStrategy strategy,
                                         int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addNodeTasks<GmMesh, T>(m, affNodes, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
  {
    if(!ok)
      luaL_error(_threadManager->threadLuaEnv(0)->state(), "%s", luaPrintable(QObject::tr("Task execution aborted.")));
    return;
  }
 
  // Run tasks!
  _threadManager->execTasks(nworkers);
}
 
template <class T>
void GmTaskManager::execParallelNodeLoop(const GmNodeSet* m, T& taskObj,
    int ntasks, NodePartitionStrategy strategy,
    int nworkers, const QVector<int>& taskAffinity)
{
    S_TRACE();
 
    // Add tasks to the manager.
    bool ok;
    if (!addNodeTasks<GmNodeSet, T>(m, GM_BOTH, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
    {
        if (!ok)
            luaL_error(_threadManager->threadLuaEnv(0)->state(), "%s", luaPrintable(QObject::tr("Task execution aborted.")));
        return;
    }
 
    // Run tasks!
    _threadManager->execTasks(nworkers);
}
 
template <class M, class T>
bool GmTaskManager::addNodeTasks(const M* m, GmAffectedNodes affNodes, T& taskObj,
                                 int ntasks, NodePartitionStrategy strategy, int nworkers, 
                                 const QVector<int>& taskAffinity, bool* ok)
{
  S_TRACE();
 
  *ok = true;
 
  // Adjust execution parameters, converting possible default options to real values
  adjustExecParameters(&ntasks, &nworkers);
  if(strategy == NPS_DEFAULT)
    strategy = _defNodeStrategy;
  assert(ntasks > 0 && nworkers >= 0 && strategy >= 0 && strategy < NPS_NUM_STRATEGIES);
 
  // Partition the mesh nodes
  QVector<GmNodeTaskDataSet*> taskData = buildNodePartition<M>(m, ntasks, affNodes, strategy);
  if(taskData.isEmpty())  // No tasks 
    return false;
 
  // Check affinity vector
  bool useAffinity = !taskAffinity.isEmpty();
  if(useAffinity && taskData.size() != taskAffinity.size())
  {
    qDeleteAll(taskData);
    gmErrorMsg(_threadManager->logger(), QObject::tr("Error creating node tasks. Affinity vector size (%1) is different from the number of tasks (%2).")
                                                     .arg(taskAffinity.size()).arg(taskData.size()));
    *ok = false;
    return false;  // No way to support the affinity request
  }
 
  // Add tasks to the manager. The thread manager takes ownership and destroys the tasks
  class NodeTask: public GmThreadTask
  {
  public:
    NodeTask(T& task, GmNodeTaskDataSet* dataSet) : _task(task), _dataSet(dataSet) {}
    virtual ~NodeTask() { delete _dataSet; }
 
    virtual GmThreadTaskResult run(const GmThread* thread) { return _task(_dataSet, thread); }
 
  private:
    T&                 _task;    
    GmNodeTaskDataSet* _dataSet; 
  };
 
  int i = 0;
  foreach(GmNodeTaskDataSet* ds, taskData)
    _threadManager->addTask(new NodeTask(taskObj, ds), useAffinity ? taskAffinity[i++] : 0);
 
  return true;
}
 
template <class T>
GmThreadTaskResult GmTaskManager::runParallelCellLoop(GmCellMesh* m, bool activeOnly, T& taskObj,
                                                      int ntasks, CellPartitionStrategy strategy,
                                                      int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addCellTasks<GmCellMesh, T>(m, activeOnly, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
    return ok ? GM_THREAD_OK : GM_THREAD_ABORTED; // false + ok = no tasks  
 
  // Run tasks!
  return _threadManager->runTasks(nworkers);
}
 
template <class T>
GmThreadTaskResult GmTaskManager::runParallelCellLoop(GmCellGroupSet* gs, bool activeOnly, T& taskObj,
                                                      int ntasks, CellPartitionStrategy strategy,
                                                      int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addCellTasks<GmCellGroupSet, T>(gs, activeOnly, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
    return ok ? GM_THREAD_OK : GM_THREAD_ABORTED; // false + ok = no tasks  
 
  // Run tasks!
  return _threadManager->runTasks(nworkers);
}
 
template <class T>
void GmTaskManager::execParallelCellLoop(GmCellMesh* m, bool activeOnly, T& taskObj,
                                         int ntasks, CellPartitionStrategy strategy,
                                         int nworkers, const QVector<int>& taskAffinity)
{
  S_TRACE();
 
  // Add tasks to the manager.
  bool ok;
  if(!addCellTasks<GmCellMesh, T>(m, activeOnly, taskObj, ntasks, strategy, nworkers, taskAffinity, &ok))
  {
    if(!ok)
      luaL_error(_threadManager->threadLuaEnv(0)->state(), "%s", luaPrintable(QObject::tr("Task execution aborted.")));
    return;
  }
 
  // Run tasks!
  _threadManager->execTasks(nworkers);
}
 
template <class M, class T>
bool GmTaskManager::addCellTasks(M* m, bool activeOnly, T& taskObj,
                                 int ntasks, CellPartitionStrategy strategy, int nworkers,
                                 const QVector<int>& taskAffinity, bool* ok)
{
  S_TRACE();
 
  *ok = true;
 
  // Adjust execution parameters, converting possible default options to real values
  adjustExecParameters(&ntasks, &nworkers);
  if(strategy == CPS_DEFAULT)
    strategy = _defCellStrategy;
  assert(ntasks > 0 && nworkers >= 0 && strategy >= 0 && strategy < CPS_NUM_STRATEGIES);
 
  // Partition the mesh nodes
  QVector<GmCellTaskDataSet*> taskData = buildCellPartition(m, ntasks, activeOnly, strategy);
  if(taskData.isEmpty())  // No tasks 
    return false;
 
  // Check affinity vector
  bool useAffinity = !taskAffinity.isEmpty();
  if(useAffinity && taskData.size() != taskAffinity.size())
  {
    qDeleteAll(taskData);
    gmErrorMsg(_threadManager->logger(), QObject::tr("Error creating cell tasks. Affinity vector size (%1) is different from the number of tasks (%2).")
                                                     .arg(taskAffinity.size()).arg(taskData.size()));
    *ok = false;
    return false;  // No way to support the affinity request
  }
 
  // Add tasks to the manager. The thread manager takes ownership and destroys the tasks
  class CellTask: public GmThreadTask
  {
  public:
    CellTask(T& task, GmCellTaskDataSet* dataSet) : _task(task), _dataSet(dataSet) {}
    virtual ~CellTask() { delete _dataSet; }
 
    virtual GmThreadTaskResult run(const GmThread* thread) { return _task(_dataSet, thread); }
 
  private:
    T&                 _task;    
    GmCellTaskDataSet* _dataSet; 
  };
 
  int i = 0;
  foreach(GmCellTaskDataSet* ds, taskData)
    _threadManager->addTask(new CellTask(taskObj, ds), useAffinity ? taskAffinity[i++] : 0);
 
  return true;
}
 
//Parallel accessor traversal
 
/*
Traverse single accessor, (aggregation)
*/
template<typename Accessor, typename ReduceF, typename JoinF, typename InitV>
#if (defined(_MSVC_LANG) && _MSVC_LANG >= 202002L) || (__cplusplus >= 202002L)
    requires std::invocable<ReduceF, const double*, InitV&>
&& std::invocable<JoinF, const InitV&, InitV&>
&& (std::same_as<Accessor, GmValueAccessor> || std::same_as<Accessor, GmCellAccessor> || std::same_as<Accessor, GmGaussAccessor>)
#endif
void GmTaskManager::traverseAccessor(int nworkers, const Accessor* acc, const GmCellMesh* cmesh, const ReduceF& reduce, const JoinF& join, InitV& val)
{
    S_TRACE();
    assert(acc);
    GmTLS<InitV, true> tval;
    tval.init(val);
 
    const GmMeshGaussIndex* gi;
    const GmElementMesh* emesh;
    if constexpr (std::is_same_v<Accessor, GmGaussAccessor>)
    {
        gi = acc->gaussIndex();
        emesh = gi->mesh();
        assert(emesh == dynamic_cast<const GmElementMesh*>(cmesh));
    }
    else if constexpr (std::is_same_v<Accessor, GmCellAccessor>)
    {
    }
    else if constexpr (std::is_same_v<Accessor, GmValueAccessor>)
    {
    }
    else {
        assert(0);
        return;
    }
 
    auto task = [&](GmNodeTaskDataSet* dataSet, const GmThread* thread) {
        int i;
        InitV& lval = tval.localData();
 
        while ((i = dataSet->nextNode()) != -1)
        {
            if (thread->cancelPending())
                return GM_THREAD_CANCELED;
 
            const double* v;
            if constexpr (std::is_same_v<Accessor, GmGaussAccessor>)
            {
                //For non-function accessors it would be better to just traverse
                //the memory, but to be general we need to follow the (cell,ip)
                //interface.
                int ip;
                int cId = gi->cellIp(i, &ip);
                v = acc->valueAt(emesh->cell(cId), ip, NULL);
            }
            else if constexpr (std::is_same_v<Accessor, GmCellAccessor>)
            {
                //To avoid warnings with user functions with no coordinates
                //we use ip=0. The default would be to evaluate at the cell 
                //center but that requires the cell nodes or knowing the 
                //natural coordinates dimension.
                //If we traverse a cell accessor that is a function varying
                //within the cell, any kind of reduction operation is already
                //defeated, so we go for the cheapest "ip=0"
                int ip = 0;
                v = acc->valueAt(cmesh->cell(i), NULL, ip);
            }
            else if constexpr (std::is_same_v<Accessor, GmValueAccessor>)
            {
                v = acc->value(i);
            }
            reduce(v, lval);
        }
        return GM_THREAD_OK;
        };
 
    runParallelCountLoop(acc->size(), task, nworkers);
 
    //Join thread results
    for (int i = 0; i < tval.size(); ++i) {
        join(tval.localData(i), val);
    }
}
 
/*
Traverse two accessors,(error calculation, etc).
*/
template<typename Accessor, typename ReduceF, typename JoinF, typename InitV>
#if (defined(_MSVC_LANG) && _MSVC_LANG >= 202002L) || (__cplusplus >= 202002L)
    requires std::invocable<ReduceF, const double*, const double*, InitV&>
&& std::invocable<JoinF, const InitV&, InitV&>
&& (std::same_as<Accessor, GmValueAccessor> || std::same_as<Accessor, GmCellAccessor> || std::same_as<Accessor, GmGaussAccessor>)
#endif
void GmTaskManager::traverseAccessor(int nworkers, const Accessor* acc1, const Accessor* acc2, const GmCellMesh* cmesh, const ReduceF& reduce, const JoinF& join, InitV& val)
{
    S_TRACE();
    assert(acc1 && acc2);
    GmTLS<InitV, true> tval;
    tval.init(val);
 
    const GmMeshGaussIndex* gi;
    const GmElementMesh* emesh;
    if constexpr (std::is_same_v<Accessor, GmGaussAccessor>)
    {
        gi = acc1->gaussIndex();
        emesh = gi->mesh();
        assert(emesh == cmesh);
    }
    else if constexpr (std::is_same_v<Accessor, GmCellAccessor>) {}
    else if constexpr (std::is_same_v<Accessor, GmValueAccessor>) {}
    else {
        assert(0);
        return;
    }
    assert(acc1->size() == acc2->size());
 
    auto task = [&](GmNodeTaskDataSet* dataSet, const GmThread* thread) {
        int i;
        InitV& lval = tval.localData();
 
        while ((i = dataSet->nextNode()) != -1)
        {
            if (thread->cancelPending())
                return GM_THREAD_CANCELED;
 
            const double* v1, * v2;
            if constexpr (std::is_same_v<Accessor, GmGaussAccessor>)
            {
                //For non-function accessors it would be better to just traverse
                //the memory, but to be general we need to follow the (cell,ip)
                //interface.
                int ip;
                int cId = gi->cellIp(i, &ip);
                const GmCell* cell = emesh->cell(cId);
                v1 = acc1->valueAt(cell, ip, NULL);
                v2 = acc2->valueAt(cell, ip, NULL);
            }
            else if constexpr (std::is_same_v<Accessor, GmCellAccessor>)
            {
                const GmCell* cell = cmesh->cell(i);
                v1 = acc1->valueAt(cell, NULL);
                v2 = acc2->valueAt(cell, NULL);
            }
            else if constexpr (std::is_same_v<Accessor, GmValueAccessor>)
            {
                v1 = acc1->value(i);
                v2 = acc2->value(i);
            }
            reduce(v1, v2, lval);
        }
        return GM_THREAD_OK;
        };
 
    runParallelCountLoop(acc1->size(), task, nworkers);
 
    //Join thread results
    for (int i = 0; i < tval.size(); ++i) {
        join(tval.localData(i), val);
    }
}
 
#endif