core_execution_pipeline.h

// 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.
#include <utility>



#ifndef HEDGEHOG_CORE_EXECUTION_PIPELINE_H
#define HEDGEHOG_CORE_EXECUTION_PIPELINE_H

#include "core_switch.h"
#include "../core_task.h"
#include "../../../api/graph.h"

namespace hh {

#ifndef DOXYGEN_SHOULD_SKIP_THIS
template<class GraphOutput, class ...GraphInputs>
class AbstractExecutionPipeline;
#endif //DOXYGEN_SHOULD_SKIP_THIS

namespace core {

template<class GraphOutput, class ...GraphInputs>
class CoreExecutionPipeline : public CoreTask<GraphOutput, GraphInputs...> {
 private:
  AbstractExecutionPipeline<GraphOutput, GraphInputs...> *executionPipeline_ = nullptr; 
  size_t numberGraphs_ = 0; 
  std::vector<int> deviceIds_ = {}; 

 protected:
  std::shared_ptr<CoreSwitch<GraphInputs...>> coreSwitch_; 
  std::vector<std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>>> epGraphs_ = {}; 

 public:
  CoreExecutionPipeline() = delete;

  CoreExecutionPipeline(std::string_view const &name,
                        AbstractExecutionPipeline<GraphOutput, GraphInputs...> *executionPipeline,
                        std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>> coreBaseGraph,
                        size_t numberGraphs,
                        std::vector<int> const &deviceIds,
                        bool automaticStart)
      : CoreTask<GraphOutput, GraphInputs...>(name, 1, NodeType::ExecutionPipeline, nullptr, automaticStart),
        executionPipeline_(executionPipeline),
        numberGraphs_(numberGraphs),
        deviceIds_(deviceIds),
        coreSwitch_(std::make_shared<CoreSwitch<GraphInputs...>>()) {
    if (this->numberGraphs_ == 0) { this->numberGraphs_ = 1; }

    if (coreBaseGraph->isInside() || this->isInside()) {
      std::ostringstream oss;
      oss << "You can not modify a graph that is connected inside another graph: " << __FUNCTION__;
      HLOG_SELF(0, oss.str())
      throw (std::runtime_error(oss.str()));
    }
    if (numberGraphs_ != deviceIds.size()) {
      std::ostringstream oss;
      oss
          << "The number of device Ids do not correspond to the number of coreGraph duplications you sent to the "
             "execution pipeline \""
          << name
          << "\". Even if you do not associate the graphs duplicates to a specific device, please set the deviceIds "
             "to the map to the number of duplicates specified."
          << std::endl;

      HLOG_SELF(0, oss.str())
      throw (std::runtime_error(oss.str()));
    }
    epGraphs_.reserve(this->numberGraphs_);

    coreBaseGraph->graphId(0);
    coreBaseGraph->deviceId(this->deviceIds_[0]);
    connectGraphToEP(coreBaseGraph);

    this->duplicateGraphs();
  }

  virtual ~CoreExecutionPipeline() = default;

  [[nodiscard]] behavior::Node *node() override { return this->executionPipeline_; }

  [[nodiscard]] std::vector<int> const &deviceIds() const { return deviceIds_; }

  [[nodiscard]] size_t numberGraphs() const { return numberGraphs_; }

  [[nodiscard]] std::string id() const override { return this->coreSwitch_->id(); }

  AbstractExecutionPipeline<GraphOutput, GraphInputs...> *executionPipeline() const {
    return executionPipeline_;
  }

  int deviceId() override {
    std::ostringstream oss;
    oss << "Internal error, an execution pipeline has not device id: " << __FUNCTION__;
    HLOG_SELF(0, oss.str())
    throw (std::runtime_error(oss.str()));
  }

  std::set<CoreSender<GraphOutput> *>
  getSenders() override {
    std::set<CoreSender<GraphOutput> *>
        res = {},
        senders = {};
    for (auto epGraph : this->epGraphs_) {
      senders.clear();
      senders = epGraph->getSenders();
      mergeSenders(res, senders);
    }
    return res;
  }

  std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>> baseCoreGraph() {
    return this->epGraphs_.empty() ? nullptr : this->epGraphs_[0];
  }

  void addReceiver(CoreReceiver<GraphOutput> *receiver) override {
    for (CoreReceiver<GraphOutput> *r : receiver->receivers()) {
      if (auto coreQueueReceiver = dynamic_cast<CoreQueueReceiver<GraphOutput> *>(r)) {
        this->destinations()->insert(coreQueueReceiver);
      } else {
        std::ostringstream oss;
        oss << "Internal error, a receiver added to an execution pipeline is not a coreQueueReceiver : "
            << __FUNCTION__;
        HLOG_SELF(0, oss.str())
        throw (std::runtime_error(oss.str()));
      }
    }
    for (auto epGraph : this->epGraphs_) {
      connectGraphsOutputToReceiver(epGraph.get(), receiver);
    }
  }

  void addSlot(CoreSlot *slot) override {
    for (auto graph : this->epGraphs_) { graph->addSlot(slot); }
  }

  void visit(AbstractPrinter *printer) override {
    if (printer->hasNotBeenVisited(this)) {
      printer->printExecutionPipelineHeader(this, coreSwitch_.get());
      for (auto graph: epGraphs_) {
        (this->printEdgeSwitchGraphs<GraphInputs>(printer, graph.get()), ...);
        graph->visit(printer);
      }
      printer->printExecutionPipelineFooter();
    }
  }

  void createCluster([[maybe_unused]]std::shared_ptr<std::multimap<CoreNode *,
                                                                   std::shared_ptr<CoreNode>>> &) override {
    for (std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>> epGraph : this->epGraphs_) {
      epGraph->createInnerClustersAndLaunchThreads();
    }
  }

  [[nodiscard]] std::chrono::duration<uint64_t, std::micro> maxExecutionTime() const override {
    std::chrono::duration<uint64_t, std::micro> ret = std::chrono::duration<uint64_t, std::micro>::min();
    for (auto graph: epGraphs_) {
      std::chrono::duration<uint64_t, std::micro> temp = graph->maxExecutionTime();
      if (temp > ret) ret = temp;
    }
    return ret;
  }

  [[nodiscard]] std::chrono::duration<uint64_t, std::micro> minExecutionTime() const override {
    std::chrono::duration<uint64_t, std::micro> ret = std::chrono::duration<uint64_t, std::micro>::max();
    for (auto graph: epGraphs_) {
      std::chrono::duration<uint64_t, std::micro> temp = graph->minExecutionTime();
      if (temp < ret) ret = temp;
    }
    return ret;
  }

  [[nodiscard]] std::chrono::duration<uint64_t, std::micro> maxWaitTime() const override {
    std::chrono::duration<uint64_t, std::micro> ret = std::chrono::duration<uint64_t, std::micro>::min();
    for (auto graph: epGraphs_) {
      std::chrono::duration<uint64_t, std::micro> temp = graph->maxWaitTime();
      if (temp > ret) ret = temp;
    }
    return ret;
  }

  [[nodiscard]] std::chrono::duration<uint64_t, std::micro> minWaitTime() const override {
    std::chrono::duration<uint64_t, std::micro> ret = std::chrono::duration<uint64_t, std::micro>::max();
    for (auto graph: epGraphs_) {
      std::chrono::duration<uint64_t, std::micro> temp = graph->minWaitTime();
      if (temp < ret) ret = temp;
    }
    return ret;
  }

 protected:
  bool callCanTerminate(bool lock) override {
    bool result;
    if (lock) { this->lockUniqueMutex(); }
    result = !this->hasNotifierConnected() && this->receiversEmpty();
    HLOG_SELF(2, "callCanTerminate: " << std::boolalpha << result)
    if (lock) { this->unlockUniqueMutex(); }
    return result;
  };

 private:
  void duplicateGraphs() {
    for (size_t numberGraph = 1; numberGraph < this->numberGraphs(); ++numberGraph) {
      auto graphDuplicate =
          std::dynamic_pointer_cast<CoreGraph<GraphOutput, GraphInputs...>>(this->baseCoreGraph()->clone());
      graphDuplicate->graphId(numberGraph);
      graphDuplicate->deviceId(this->deviceIds_[numberGraph]);
      connectGraphToEP(graphDuplicate);
    }
  }

  template<class GraphInput>
  void addEdgeSwitchGraph(std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>> &graph) {
    auto coreSender = std::static_pointer_cast<CoreSender<GraphInput>>(this->coreSwitch_);
    auto coreNotifier = std::static_pointer_cast<CoreNotifier>(coreSender);
    auto coreSlot = std::static_pointer_cast<CoreSlot>(graph);
    auto coreReceiver = std::static_pointer_cast<CoreReceiver<GraphInput>>(graph);

    for (auto r : coreReceiver->receivers()) { coreSender->addReceiver(r); }
    for (auto s : coreSender->getSenders()) {
      coreReceiver->addSender(s);
      coreSlot->addNotifier(s);
    }
    for (CoreSlot *slot : coreSlot->getSlots()) { coreNotifier->addSlot(slot); }
  }

  void connectGraphToEP(std::shared_ptr<CoreGraph<GraphOutput, GraphInputs...>> &coreGraph) {
    coreGraph->setInside();
    coreGraph->belongingNode(this);
    coreGraph->hasBeenRegistered(true);
    (addEdgeSwitchGraph<GraphInputs>(coreGraph), ...);
    this->epGraphs_.push_back(coreGraph);
  }

  void connectGraphsOutputToReceiver(CoreGraph<GraphOutput, GraphInputs...> *graph,
                                     CoreReceiver<GraphOutput> *coreReceiver) {
    for (CoreReceiver<GraphOutput> *receiver : coreReceiver->receivers()) { graph->addReceiver(receiver); }
  }

  void mergeSenders(std::set<CoreSender<GraphOutput> *
  > &superSet, std::set<CoreSender<GraphOutput> *> &graphSenders) {
    for (CoreSender<GraphOutput> *sender : graphSenders) { superSet.insert(sender); }
  }

  template<class GraphInput>
  void printEdgeSwitchGraphs(AbstractPrinter *printer, CoreGraph<GraphOutput, GraphInputs...> *graph) {
    for (CoreNode *graphInputNode : *(graph->inputsCoreNodes())) {
      if (auto *coreQueueReceiver = dynamic_cast<CoreQueueReceiver<GraphInput> *>(graphInputNode)) {
        printer->printEdgeSwitchGraphs(coreQueueReceiver,
                                       this->id(),
                                       traits::type_name<GraphInput>(),
                                       coreQueueReceiver->queueSize(),
                                       coreQueueReceiver->maxQueueSize(),
                                       traits::is_managed_memory_v<GraphInput>);
      }
    }
  }
};

}
}
#endif //HEDGEHOG_CORE_EXECUTION_PIPELINE_H