core_execution_pipeline.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_EXECUTION_PIPELINE_H_
#define HEDGEHOG_CORE_EXECUTION_PIPELINE_H_
 
#include <utility>
 
#include "../../tools/traits.h"
#include "../../api/graph/graph.h"
 
#include "../abstractions/base/node/execution_pipeline_node_abstraction.h"
#include "../abstractions/node/execution_pipeline_inputs_management_abstraction.h"
#include "../abstractions/node/execution_pipeline_outputs_management_abstraction.h"
 
namespace hh {
 
#ifndef DOXYGEN_SHOULD_SKIP_THIS
template<size_t Separator, class ...AllTypes>
class AbstractExecutionPipeline;
#endif //DOXYGEN_SHOULD_SKIP_THIS
 
namespace core {
 
template<size_t Separator, class ...AllTypes>
using EPIM =
    tool::ExecutionPipelineInputsManagementAbstractionTypeDeducer_t<tool::Inputs<Separator, AllTypes...>>;
 
template<size_t Separator, class ...AllTypes>
using EPOM =
    tool::ExecutionPipelineOutputsManagementAbstractionTypeDeducer_t<tool::Outputs<Separator, AllTypes...>>;
 
template<size_t Separator, class ...AllTypes>
class CoreExecutionPipeline
    : public abstraction::ExecutionPipelineNodeAbstraction,
      public abstraction::ClonableAbstraction,
      public EPIM<Separator, AllTypes...>,
      public EPOM<Separator, AllTypes...> {
 private:
  AbstractExecutionPipeline<Separator, AllTypes...> *const
      executionPipeline_ = nullptr; 
  std::vector<std::shared_ptr<CoreGraph<Separator, AllTypes...>>>
      coreGraphs_{}; 
  std::vector<int>
      deviceIds_{}; 
 
 public:
  CoreExecutionPipeline(
      AbstractExecutionPipeline<Separator, AllTypes...> *const &executionPipeline,
      std::shared_ptr<CoreGraph<Separator, AllTypes...>> const coreGraph,
      std::vector<int> deviceIds,
      std::string const name = "Execution pipeline")
      : abstraction::ExecutionPipelineNodeAbstraction(name, executionPipeline),
        EPIM<Separator, AllTypes...>(executionPipeline),
        EPOM<Separator, AllTypes...>(),
        executionPipeline_(executionPipeline),
        deviceIds_(std::move(deviceIds)) {
    initializeCoreExecutionPipeline(coreGraph);
  }
 
  CoreExecutionPipeline(
      AbstractExecutionPipeline<Separator, AllTypes...> *const &executionPipeline,
      std::shared_ptr<CoreGraph<Separator, AllTypes...>> const coreGraph,
      size_t numberGraphs,
      std::string const name = "Execution pipeline")
      : abstraction::ExecutionPipelineNodeAbstraction(name, executionPipeline),
        EPIM<Separator, AllTypes...>(executionPipeline),
        EPOM<Separator, AllTypes...>(),
        executionPipeline_(executionPipeline),
        deviceIds_(std::vector<int>(numberGraphs)) {
    std::iota(deviceIds_.begin(), deviceIds_.end(), 0);
    initializeCoreExecutionPipeline(coreGraph);
  }
 
  ~CoreExecutionPipeline() override = default;
 
  void preRun() override {}
 
  void run() override {
    std::chrono::time_point<std::chrono::system_clock>
        start,
        finish;
 
    using Inputs_t = typename EPIM<Separator, AllTypes...>::inputs_t;
    using Indices = std::make_index_sequence<std::tuple_size_v<Inputs_t>>;
 
    this->isActive(true);
 
    // Actual computation loop
    while (!this->canTerminate(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 te 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<Inputs_t>(Indices{});
      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->disconnectSwitch();
    for (auto &g : this->coreGraphs_) { g->waitForTermination(); }
    this->isActive(false);
  }
 
  [[nodiscard]] std::string extraPrintingInformation() const override { return {}; }
 
  void visit(Printer *printer) override {
    // Register the node
    if (printer->registerNode(this)) {
      printer->printExecutionPipelineHeader(this, this->coreSwitch());
      EPIM<Separator, AllTypes...>::printEdgesInformation(printer);
      for (auto coreGraph : coreGraphs_) { coreGraph->visit(printer); }
      printer->printExecutionPipelineFooter();
    }
  }
 
  void registerNode(abstraction::GraphNodeAbstraction *belongingGraph) override {
    NodeAbstraction::registerNode(belongingGraph);
    for (auto coreGraph : coreGraphs_) { coreGraph->registerNode(belongingGraph); }
  }
 
  [[nodiscard]] std::pair<std::chrono::nanoseconds, std::chrono::nanoseconds> minMaxExecutionDuration() const override {
    std::pair<std::chrono::nanoseconds, std::chrono::nanoseconds> minMaxExecDuration = {
        std::chrono::nanoseconds::max(), std::chrono::nanoseconds::min()
    };
 
    for(auto const & graph : coreGraphs_){
      auto const minMaxGraph = graph->minMaxExecutionDuration();
      minMaxExecDuration.first = std::min(minMaxExecDuration.first, minMaxGraph.first);
      minMaxExecDuration.second = std::max(minMaxExecDuration.second, minMaxGraph.second);
    }
    return minMaxExecDuration;
  }
 
  [[nodiscard]] std::pair<std::chrono::nanoseconds, std::chrono::nanoseconds> minMaxWaitDuration() const override {
    std::pair<std::chrono::nanoseconds, std::chrono::nanoseconds> minMaxWaitDuration = {
        std::chrono::nanoseconds::max(), std::chrono::nanoseconds::min()
    };
 
    for(auto const & graph : coreGraphs_){
      auto const minMaxGraph = graph->minMaxWaitDuration();
      minMaxWaitDuration.first = std::min(minMaxWaitDuration.first, minMaxGraph.first);
      minMaxWaitDuration.second = std::max(minMaxWaitDuration.second, minMaxGraph.second);
    }
    return minMaxWaitDuration;
  }
 
 private:
  void initializeCoreExecutionPipeline(std::shared_ptr<CoreGraph<Separator, AllTypes...>> const coreGraph) {
    if (coreGraph == nullptr) {
      throw std::runtime_error("An execution pipeline should be created with a valid graph (!= nullptr).");
    }
    if (coreGraph->graphStatus() != abstraction::GraphNodeAbstraction::INIT) {
      throw std::runtime_error("An execution pipeline should be created with a graph that is not already part of "
                               "another graph or being executed.");
    }
    if (deviceIds_.empty()) {
      throw std::runtime_error(
          "An execution pipeline should be created with a valid number of graph clones (!= 0) and with valid graph ids.");
    }
    coreGraphs_.reserve(deviceIds_.size());
    registerAndDuplicateGraph(coreGraph);
    connectCoreGraphs();
    setInitialized();
  }
 
  void registerAndDuplicateGraph(std::shared_ptr<CoreGraph<Separator, AllTypes...>> const coreGraph) {
    coreGraph->deviceId(deviceIds_.at(0));
    coreGraph->graphId(0);
    coreGraph->setInside();
    coreGraphs_.push_back(coreGraph);
    for (size_t graphId = 1; graphId < this->deviceIds_.size(); ++graphId) {
      std::map<NodeAbstraction *, std::shared_ptr<NodeAbstraction>> correspondenceMap;
      auto clone = std::dynamic_pointer_cast<CoreGraph<Separator, AllTypes...>>(
          coreGraph->clone(correspondenceMap)
      );
      clone->deviceId(deviceIds_.at(graphId));
      clone->graphId(graphId);
      clone->setInside();
      coreGraphs_.push_back(clone);
    }
  }
 
  void connectCoreGraphs() {
    for (auto &coreGraph : coreGraphs_) {
      EPIM<Separator, AllTypes...>::connectGraphToSwitch(coreGraph);
      EPOM<Separator, AllTypes...>::registerGraphOutputNodes(coreGraph);
    }
  }
 
  template<class InputTypes, size_t ...Indices>
  void operateReceivers(std::index_sequence<Indices...>) {
    (this->template operateReceiver<std::tuple_element_t<Indices, InputTypes>>(), ...);
  }
 
  template<class Input>
  void operateReceiver() {
    this->lockSlotMutex();
    auto typedReceiver = static_cast<abstraction::ReceiverAbstraction<Input> *>(this);
    if (!typedReceiver->empty()) {
      std::shared_ptr<Input> data = typedReceiver->getInputData();
      this->unlockSlotMutex();
      for (auto coreGraph : this->coreGraphs_) {
        if (EPIM<Separator, AllTypes...>::callSendToGraph(data, coreGraph->graphId())) {
          std::static_pointer_cast<abstraction::ReceiverAbstraction<Input>>(coreGraph)->receive(data);
          std::static_pointer_cast<abstraction::SlotAbstraction>(coreGraph)->wakeUp();
        }
      }
    } else {
      this->unlockSlotMutex();
    }
  }
 
  void launchGraphThreads(bool waitForInitialization) override {
    for (auto coreGraph : this->coreGraphs_) {
      coreGraph->createInnerGroupsAndLaunchThreads(waitForInitialization);
    }
  }
 
  [[nodiscard]] std::string id() const override { return this->coreSwitch()->id(); }
 
  std::shared_ptr<abstraction::NodeAbstraction> clone(
      std::map<NodeAbstraction *, std::shared_ptr<NodeAbstraction>> &correspondenceMap) override {
    return std::make_shared<CoreExecutionPipeline>(
        this->executionPipeline_,
        std::dynamic_pointer_cast<CoreGraph<Separator, AllTypes...>>(this->coreGraphs_[0]->clone(correspondenceMap)),
        this->deviceIds_);
  }
 
  void duplicateEdge(std::map<NodeAbstraction *, std::shared_ptr<NodeAbstraction>> &mapping) override {
    this->duplicateOutputEdges(mapping);
  }
 
};
 
}
}
#endif //HEDGEHOG_CORE_EXECUTION_PIPELINE_H_