Tasks.cpp

Go to the documentation of this file.

// This file is part of the Peano project. For conditions of distribution and
// use, please see the copyright notice at www.peano-framework.org
#include "tarch/multicore/Task.h"
#include "tarch/Assertions.h"
#include "tarch/logging/Statistics.h"
#include "tarch/mpi/Rank.h"
#include "tarch/multicore/Core.h"
#include "tarch/multicore/multicore.h"
#include "tarch/multicore/otter.h"
#include "tarch/services/ServiceRepository.h"
 
#include "tarch/multicore/BooleanSemaphore.h"
#include "tarch/multicore/Lock.h"
 
#if defined(SharedTBB)
#include <tbb/task_arena.h>
#include <tbb/task_group.h>
#include <oneapi/tbb/task_arena.h>
#include <oneapi/tbb/concurrent_hash_map.h>
#include <atomic>
 
 
#include "TaskHandleRepository.h"
 
 
 
namespace {
  class GlobalTaskGroup : public ::oneapi::tbb::task_group {
    public:
      GlobalTaskGroup() : task_group() {}
      ~GlobalTaskGroup() {
        wait();
      }
  };
 
  ::tarch::multicore::tbb::TaskHandleRepository taskHandleRepository;
  
  GlobalTaskGroup     globalTaskGroup;
 
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
  #else
  static ::oneapi::tbb::dynamic_task_graph                  taskGraph(globalTaskGroup);
  #endif
}
 
 
void tarch::multicore::native::spawnAndWaitAsTaskLoop(const std::vector<tarch::multicore::Task*>& tasks) {
  assertion(not tasks.empty());
 
  ::tbb::task_group taskGroup;
  for (auto& p : tasks) {
    taskGroup.run([&tasks, p]() -> void {
      while (p->run()) {
        tarch::multicore::Core::getInstance().yield();
      }
      delete p;
    });
  }
 
  taskGroup.wait();
}
 
 
 
void tarch::multicore::native::processFusedTask(
  Task* myTask, const std::list<tarch::multicore::Task*>& tasksOfSameType, int device
) {
  std::list<tarch::multicore::Task*> copyOfTasksOfSameType = tasksOfSameType;
  globalTaskGroup.run([&, myTask, copyOfTasksOfSameType] {
    bool stillExecuteLocally = myTask->fuse(copyOfTasksOfSameType, device);
    if (stillExecuteLocally) {
      tarch::multicore::native::spawnTask(myTask, std::set<TaskNumber>(), NoOutDependencies);
    } else {
      delete myTask;
    }
  });
}
 
 
void tarch::multicore::native::waitForTasks(const std::set<TaskNumber>& inDependencies) {
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
   ::oneapi::tbb::task_handle*  handle = new ::oneapi::tbb::task_handle(
     globalTaskGroup.defer(
       []() -> void {
       }
     )
   );
  #else
   ::tbb::dynamic_task_graph_node* handle = new ::tbb::dynamic_task_graph_node(
     []() -> void {
     }
   );
  #endif
 
  taskHandleRepository.addDependencies( handle, inDependencies );
 
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
  //globalTaskGroup.run( std::move(*handle) );
  globalTaskGroup.run_and_wait( handle );
  #else
  taskGraph.put( *handle );
  taskGraph.wait( *handle );
  #endif
 
  delete handle;
}
 
 
void tarch::multicore::native::waitForAllTasks() {
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
  #else
  taskGraph.wait_for_all();
  #endif
 
  globalTaskGroup.wait();
}
 
 
void tarch::multicore::native::spawnTask(
  Task*                        job,
  const std::set<TaskNumber>&  inDependencies,
  const TaskNumber&            taskNumber
) {
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
  ::oneapi::tbb::task_handle*  handle = new ::oneapi::tbb::task_handle(
      globalTaskGroup.defer(
        [job]() -> void {
          while (job->run()) {
          }
          delete job;
        }
      )
    );
  #elif PeanoDebug>0
    ::tbb::dynamic_task_graph_node* handle = new ::tbb::dynamic_task_graph_node(
      [job]() -> void {
        while (job->run()) {
        }
        delete job;
      },
      job->toString()
    );
  #else
    ::tbb::dynamic_task_graph_node* handle = new ::tbb::dynamic_task_graph_node(
      [job]() -> void {
        while (job->run()) {
        }
        delete job;
      }
    );
  #endif
 
  taskHandleRepository.addDependency( handle, taskNumber );
  taskHandleRepository.addDependencies( handle, inDependencies );
 
  #if !defined(SharedTBBExtension) and !defined(TBB_USE_TASK_GROUP_PREVIEW)
  //globalTaskGroup.run( std::move(*handle) );
  globalTaskGroup.run( handle );
  #else
  taskGraph.put( *handle );
  #endif
 
  if (taskNumber==NoOutDependencies) {
    delete handle;
  }
  else {
    taskHandleRepository.registerTask(handle, taskNumber);
  }
}
 
 
#endif // defined(SharedTBB)