core_task.h

Go to the documentation of this file.

// NIST-developed software is provided by NIST as a public service. You may use, copy and distribute copies of the
// software in any medium, provided that you keep intact this entire notice. You may improve, modify and create
// derivative works of the software or any portion of the software, and you may copy and distribute such modifications
// or works. Modified works should carry a notice stating that you changed the software and should note the date and
// nature of any such change. Please explicitly acknowledge the National Institute of Standards and Technology as the
// source of the software. NIST-developed software is expressly provided "AS IS." NIST MAKES NO WARRANTY OF ANY KIND,
// EXPRESS, IMPLIED, IN FACT OR ARISING BY OPERATION OF LAW, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTY OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT AND DATA ACCURACY. NIST NEITHER REPRESENTS NOR
// WARRANTS THAT THE OPERATION OF THE SOFTWARE WILL BE UNINTERRUPTED OR ERROR-FREE, OR THAT ANY DEFECTS WILL BE
// CORRECTED. NIST DOES NOT WARRANT OR MAKE ANY REPRESENTATIONS REGARDING THE USE OF THE SOFTWARE OR THE RESULTS
// THEREOF, INCLUDING BUT NOT LIMITED TO THE CORRECTNESS, ACCURACY, RELIABILITY, OR USEFULNESS OF THE SOFTWARE. You
// are solely responsible for determining the appropriateness of using and distributing the software and you assume
// all risks associated with its use, including but not limited to the risks and costs of program errors, compliance
// with applicable laws, damage to or loss of data, programs or equipment, and the unavailability or interruption of
// operation. This software is not intended to be used in any situation where a failure could cause risk of injury or
// damage to property. The software developed by NIST employees is not subject to copyright protection within the
// United States.
 
#ifndef HEDGEHOG_CORE_TASK_H
#define HEDGEHOG_CORE_TASK_H
 
#include <ostream>
#include <sstream>
 
#include "../../tools/concepts.h"
#include "../../tools/traits.h"
 
#include "../abstractions/base/clonable_abstraction.h"
#include "../abstractions/base/groupable_abstraction.h"
#include "../abstractions/base/cleanable_abstraction.h"
#include "../abstractions/base/can_terminate_abstraction.h"
 
#include "../abstractions/base/node/task_node_abstraction.h"
 
#include "../abstractions/node/task_inputs_management_abstraction.h"
#include "../abstractions/node/task_outputs_management_abstraction.h"
 
namespace hh {
 
#ifndef DOXYGEN_SHOULD_SKIP_THIS
template<size_t Separator, class ...AllTypes>
class AbstractTask;
#endif //DOXYGEN_SHOULD_SKIP_THIS
 
namespace core {
 
template<size_t Separator, class ...AllTypes>
using TIM = tool::TaskInputsManagementAbstractionTypeDeducer_t<tool::Inputs<Separator, AllTypes...>>;
 
template<size_t Separator, class ...AllTypes>
using TOM = tool::TaskOutputsManagementAbstractionTypeDeducer_t<tool::Outputs<Separator, AllTypes...>>;
 
template<size_t Separator, class ...AllTypes>
class CoreTask
    : public abstraction::TaskNodeAbstraction,
      public abstraction::ClonableAbstraction,
      public abstraction::CleanableAbstraction,
      public abstraction::CanTerminateAbstraction,
      public abstraction::GroupableAbstraction<AbstractTask<Separator, AllTypes...>, CoreTask<Separator, AllTypes...>>,
      public TIM<Separator, AllTypes...>,
      public TOM<Separator, AllTypes...> {
 
 private:
  AbstractTask<Separator, AllTypes...> *const
      task_ = nullptr; 
 
  bool const
      automaticStart_ = false; 
 
 public:
  explicit CoreTask(AbstractTask<Separator, AllTypes...> *const task) :
      TaskNodeAbstraction("Task", task),
      CleanableAbstraction(static_cast<behavior::Cleanable *>(task)),
      CanTerminateAbstraction(static_cast<behavior::CanTerminate *>(task)),
      abstraction::GroupableAbstraction<AbstractTask<Separator, AllTypes...>, CoreTask<Separator, AllTypes...>>(task,
                                                                                                                1),
      TIM<Separator, AllTypes...>(task, this),
      TOM<Separator, AllTypes...>(),
      task_(task),
      automaticStart_(false) {}
 
  CoreTask(AbstractTask<Separator, AllTypes...> *const task,
           std::string const &name, size_t const numberThreads, bool const automaticStart) :
      TaskNodeAbstraction(name, task),
      CleanableAbstraction(static_cast<behavior::Cleanable *>(task)),
      CanTerminateAbstraction(static_cast<behavior::CanTerminate *>(task)),
      abstraction::GroupableAbstraction<AbstractTask<Separator, AllTypes...>, CoreTask<Separator, AllTypes...>>(task,
                                                                                                                numberThreads),
      TIM<Separator, AllTypes...>(task, this),
      TOM<Separator, AllTypes...>(),
      task_(task),
      automaticStart_(automaticStart) {
    if (this->numberThreads() == 0) { throw std::runtime_error("A task needs at least one thread."); }
  }
 
  template<class ConcreteMultiReceivers, class ConcreteMultiExecutes, class ConcreteMultiSenders>
  CoreTask(AbstractTask<Separator, AllTypes...> *const task,
           std::string const &name, size_t const numberThreads, bool const automaticStart,
           std::shared_ptr<implementor::ImplementorSlot> concreteSlot,
           std::shared_ptr<ConcreteMultiReceivers> concreteMultiReceivers,
           std::shared_ptr<ConcreteMultiExecutes> concreteMultiExecutes,
           std::shared_ptr<implementor::ImplementorNotifier> concreteNotifier,
           std::shared_ptr<ConcreteMultiSenders> concreteMultiSenders) :
      TaskNodeAbstraction(name, task),
      CleanableAbstraction(static_cast<behavior::Cleanable *>(task)),
      CanTerminateAbstraction(static_cast<behavior::CanTerminate *>(task)),
      abstraction::GroupableAbstraction<AbstractTask<Separator, AllTypes...>, CoreTask<Separator, AllTypes...>>
          (task, numberThreads),
      TIM<Separator, AllTypes...>(this, concreteSlot, concreteMultiReceivers, concreteMultiExecutes),
      TOM<Separator, AllTypes...>(concreteNotifier, concreteMultiSenders),
      task_(task),
      automaticStart_(automaticStart) {
    if (this->numberThreads() == 0) { throw std::runtime_error("A task needs at least one thread."); }
  }
 
  ~CoreTask() override = default;
 
  [[nodiscard]] std::shared_ptr<AbstractMemoryManager> memoryManager() const override {
    return this->task_->memoryManager();
  }
 
  bool callCanTerminate(bool lock) override {
    bool result;
    if (lock) { this->lockSlotMutex(); }
    result = this->callNodeCanTerminate();
    if (lock) { this->unlockSlotMutex(); }
    return result;
  };
 
  void preRun() override {
    this->nvtxProfiler()->startRangeInitializing();
    this->task_->initialize();
    if (this->task_->memoryManager() != nullptr) {
      this->task_->memoryManager()->profiler(this->nvtxProfiler());
      this->task_->memoryManager()->deviceId(this->deviceId());
      this->task_->memoryManager()->initialize();
    }
    this->nvtxProfiler()->endRangeInitializing();
    this->setInitialized();
  }
 
  void run() override {
    std::chrono::time_point<std::chrono::system_clock>
        start,
        finish;
 
    this->isActive(true);
    this->nvtxProfiler()->initialize(this->threadId());
    this->preRun();
 
    if (this->automaticStart_) {
      start = std::chrono::system_clock::now();
      this->callAllExecuteWithNullptr();
      finish = std::chrono::system_clock::now();
      this->incrementExecutionDuration(std::chrono::duration_cast<std::chrono::nanoseconds>(finish - start));
    }
 
    // Actual computation loop
    while (!this->callCanTerminate(true)) {
      // Wait for a data to arrive or termination
      start = std::chrono::system_clock::now();
      volatile bool canTerminate = this->wait();
      finish = std::chrono::system_clock::now();
      this->incrementWaitDuration(std::chrono::duration_cast<std::chrono::nanoseconds>(finish - start));
 
      // If loop can terminate break the loop early
      if (canTerminate) { break; }
 
      // Operate the connectedReceivers to get a data and send it to execute
      start = std::chrono::system_clock::now();
      this->operateReceivers();
      finish = std::chrono::system_clock::now();
      this->incrementExecutionDuration(std::chrono::duration_cast<std::chrono::nanoseconds>(finish - start));
    }
 
    // Do the shutdown phase
    this->postRun();
    // Wake up node that this is linked to
    this->wakeUp();
  }
 
  void postRun() override {
    this->nvtxProfiler()->startRangeShuttingDown();
    this->isActive(false);
    this->task_->shutdown();
    this->notifyAllTerminated();
    this->nvtxProfiler()->endRangeShuttingDown();
  }
 
  bool wait() {
    this->nvtxProfiler()->startRangeWaiting(this->totalNumberElementsReceived());
    std::unique_lock<std::mutex> lock(*(this->mutex()));
    this->slotConditionVariable()->wait(
        lock,
        [this]() { return !this->receiversEmpty() || this->callCanTerminate(false); }
    );
    this->nvtxProfiler()->endRangeWaiting();
    return callCanTerminate(false);
  }
 
  void createGroup(std::map<NodeAbstraction *, std::vector<NodeAbstraction *>> &map) override {
    abstraction::SlotAbstraction *coreCopyAsSlot;
    abstraction::NotifierAbstraction *coreCopyAsNotifier;
 
    for (size_t threadId = 1; threadId < this->numberThreads(); ++threadId) {
      auto taskCopy = this->callCopyAndRegisterInGroup();
 
      if (taskCopy == nullptr) {
        std::ostringstream oss;
        oss << "A copy for the task \"" << this->name()
            << "\" has been invoked but return nullptr. To fix this error, overload the AbstractTask::copy function and "
               "return a valid object.";
        throw (std::runtime_error(oss.str()));
      }
 
      // Copy the memory manager
      taskCopy->connectMemoryManager(this->task_->memoryManager());
 
      auto taskCoreCopy = dynamic_cast<CoreTask<Separator, AllTypes...> *>(taskCopy->core().get());
 
      if (taskCoreCopy == nullptr) {
        std::ostringstream oss;
        oss << "A copy for the task \"" << this->name()
            << "\" does not have the same type of cores than the original task.";
        throw (std::runtime_error(oss.str()));
      }
 
      // Deal with the group registration in the graph
      map.at(static_cast<NodeAbstraction *>(this)).push_back(taskCoreCopy);
 
      // Copy inner structures
      taskCoreCopy->copyInnerStructure(this);
 
      // Make necessary connections
      coreCopyAsSlot = static_cast<abstraction::SlotAbstraction *>(taskCoreCopy);
      coreCopyAsNotifier = static_cast<abstraction::NotifierAbstraction *>(taskCoreCopy);
 
      for (auto predecessorNotifier : static_cast<abstraction::SlotAbstraction *>(this)->connectedNotifiers()) {
        for (auto notifier : predecessorNotifier->notifiers()) {
          for (auto slot : coreCopyAsSlot->slots()) {
            slot->addNotifier(notifier);
            notifier->addSlot(slot);
          }
        }
      }
 
      for (auto successorSlot : static_cast<abstraction::NotifierAbstraction *>(this)->connectedSlots()) {
        for (auto slot : successorSlot->slots()) {
          for (auto notifier : coreCopyAsNotifier->notifiers()) {
            slot->addNotifier(notifier);
            notifier->addSlot(slot);
          }
        }
      }
    }
  }
 
  [[nodiscard]] bool hasMemoryManagerAttached() const override { return this->memoryManager() != nullptr; }
 
  [[nodiscard]] std::string extraPrintingInformation() const override {
    return this->task_->extraPrintingInformation();
  }
 
  void copyInnerStructure(CoreTask<Separator, AllTypes...> *copyableCore) override {
    TIM<Separator, AllTypes...>::copyInnerStructure(copyableCore);
    TOM<Separator, AllTypes...>::copyInnerStructure(copyableCore);
  }
 
  [[nodiscard]] std::vector<std::pair<std::string const, std::string const>> ids() const override {
    return {{this->id(), this->groupRepresentativeId()}};
  }
  void visit(Printer *printer) override {
    if (printer->registerNode(this)) {
      printer->printNodeInformation(this);
      TIM<Separator, AllTypes...>::printEdgesInformation(printer);
    }
  }
 
  std::shared_ptr<abstraction::NodeAbstraction> clone(
      [[maybe_unused]] std::map<NodeAbstraction *, std::shared_ptr<NodeAbstraction>> &correspondenceMap) override {
    auto clone = std::dynamic_pointer_cast<AbstractTask<Separator, AllTypes...>>(this->callCopy());
    if (this->hasMemoryManagerAttached()) { clone->connectMemoryManager(this->memoryManager()->copy()); }
    return clone->core();
  }
 
  void duplicateEdge(std::map<NodeAbstraction *, std::shared_ptr<NodeAbstraction>> &mapping) override {
    this->duplicateOutputEdges(mapping);
  }
 
};
}
}
#endif //HEDGEHOG_CORE_TASK_H