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>

class GmSimulationData;
class GmMesh;
class GmCellMesh;
class GmCell;
class GmCellGroupSet;


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,  

    //-----------------------------
    // 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,  

    //-----------------------------
    // 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>
  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>
  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 T> bool addNodeTasks(const GmMesh* 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);

private:
  void adjustExecParameters(int* ntasks, int* nworkers);

  QVector<GmNodeTaskDataSet*> buildNodePartition(const GmMesh* 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<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>
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<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>
bool GmTaskManager::addNodeTasks(const GmMesh* 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, 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;
}

#endif