ExecutionPipeline.hpp

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 HTGS_EXECUTIONPIPELINE_H
#define HTGS_EXECUTIONPIPELINE_H

#include <cstring>

#include <htgs/core/rules/ExecutionPipelineBroadcastRule.hpp>
#include <htgs/api/ITask.hpp>
#include <htgs/api/Bookkeeper.hpp>
#include <htgs/api/TaskGraphConf.hpp>
#include <htgs/api/TaskGraphRuntime.hpp>
#include <htgs/api/IData.hpp>

namespace htgs {

  class TaskGraphRuntime;

  template<class V, class W>
  class IRule;

  template<class V, class W>
  class TaskGraphConf;



template<class T, class U>
class ExecutionPipeline : public ITask<T, U> {
  static_assert(std::is_base_of<IData, T>::value, "T must derive from IData");
  static_assert(std::is_base_of<IData, U>::value, "U must derive from IData");

 public:
  ExecutionPipeline(size_t numPipelines, TaskGraphConf<T, U> *graph, std::string name = "Execution Pipeline", bool waitForInit = true) {
    this->numPipelinesExec = numPipelines;
    this->graph = graph;
    this->inputBk = new Bookkeeper<T>();
    this->runtimes = new std::vector<TaskGraphRuntime *>();
    this->inputRules = std::shared_ptr<IRuleList<T, T>>(new IRuleList<T, T>());
    this->graphs = new std::vector<TaskGraphConf<T, U> *>();
    this->waitForInit = waitForInit;
    this->name = name;
  }

  ExecutionPipeline(size_t numPipelines, TaskGraphConf<T, U> *graph, std::shared_ptr<IRuleList<T, T>> rules, std::string name = "Execution Pipeline", bool waitForInit = true) {
    this->numPipelinesExec = numPipelines;
    this->graph = graph;
    this->inputBk = new Bookkeeper<T>();
    this->runtimes = new std::vector<TaskGraphRuntime *>();
    this->inputRules = rules;
    this->graphs = new std::vector<TaskGraphConf<T, U> *>();
    this->waitForInit = waitForInit;
    this->name = name;
  }

  ~ExecutionPipeline() {
    for (TaskGraphRuntime *runtime : *runtimes) {
      HTGS_DEBUG_VERBOSE("Execution pipeline: " << this << " Freeing memory for runtime: " << runtime);
      delete runtime;
      runtime = nullptr;
    }


    HTGS_DEBUG_VERBOSE("Execution pipeline: " << this << " Freeing memory for graph: " << graph);
    delete graph;
    graph = nullptr;

    delete runtimes;
    runtimes = nullptr;

    delete inputBk;
    inputBk = nullptr;

    delete graphs;
    graphs = nullptr;

//    delete inputRules;
//    inputRules = nullptr;
  }

  size_t getNumGraphsSpawned() override {
    return this->numPipelinesExec * graph->getNumberOfSubGraphs() + this->numPipelinesExec;
//        (graph->getNumberOfSubGraphs() == 0 ? 1 : graph->getNumberOfSubGraphs());
  }

  void addInputRule(IRule<T, T> *rule) {
    this->inputRules->push_back(std::shared_ptr<IRule<T, T>>(rule));
  }

  void addInputRule(std::shared_ptr<IRule<T, T>> rule) {
    this->inputRules->push_back(rule);
  }

  void initialize() {
    HTGS_DEBUG("Initializing Execution pipeline with " << this->numPipelinesExec << " pipelines");

    // Add a default broadcast rule if the pipeline has no rules
    if (this->inputRules->size() == 0) {
      std::cerr
          << "ERROR: Your execution pipeline does not have any decomposition rules. You must add at least one: addInputRule(IRule)"
          << std::endl;
      exit(-1);
//      this->addInputRule(new ExecutionPipelineBroadcastRule<T>());
    }

    // Create output connector using the execution pipelines output
    std::shared_ptr<Connector<U>>
        outputConnector = std::static_pointer_cast<Connector<U>>(this->getOwnerTaskManager()->getOutputConnector());

    for (size_t i = 0; i < numPipelinesExec; i++) {
      HTGS_DEBUG("Adding pipeline " << i);
      TaskGraphConf<T, U>
          *graphCopy = this->graph->copy(i, this->numPipelinesExec, nullptr, outputConnector, this->getAddress());
      // TODO: Remove or Add #ifdef this->getTaskGraphCommunicator());


#ifdef WS_PROFILE
      // TODO: Update parent for graphCopy to point to the execution pipeline
#endif
      HTGS_DEBUG("Setting up input and output of pipeline " << i);

      for (std::shared_ptr<IRule<T, T>> rule : *this->inputRules) {

        RuleManager<T, T> *ruleManager = new RuleManager<T, T>(rule /* TODO: Remove or Add #ifdef , this->getTaskGraphCommunicator()*/);
        ruleManager->setOutputConnector(graphCopy->getInputConnector());
        ruleManager->initialize(i, this->numPipelinesExec, this->getAddress());

        this->inputBk->addRuleManager(ruleManager);
      }

      graphs->push_back(graphCopy);

    }

    for (TaskGraphConf<T, U> *g : *graphs) {
      TaskGraphRuntime *runtime = new TaskGraphRuntime(g);
      runtime->executeRuntime();
      this->runtimes->push_back(runtime);
    }

    if (waitForInit) {
      for (TaskGraphConf <T, U> *g : *graphs) {
        g->waitForInitialization();
      }
    }
  }

  void shutdown() {
    HTGS_DEBUG("Shutting down " << this->getName());
    this->inputBk->shutdown();

    // Spawn thread for each runtime to properly wait without blocking.
    std::vector<std::thread *> shutdownThreads;

    for (size_t i = 0; i < runtimes->size(); i++)
    {
      std::thread *t = new std::thread(&ExecutionPipeline<T, U>::shutdownParallel, this, i);
      shutdownThreads.push_back(t);
    }

    for (std::thread *t : shutdownThreads)
    {
      if (t->joinable())
        t->join();

      delete t;
    }

//
//    for (std::vector<TaskGraphRuntime *>::reverse_iterator r = runtimes->rbegin();
//         r != runtimes->rend(); ++r ) {
//      (*r)->waitForRuntime();
//    }

//    for (TaskGraphRuntime *r : *this->runtimes) {
//      r->waitForRuntime();
//    }
  }

  void shutdownParallel(int id) {
    (*runtimes)[id]->waitForRuntime();
  }

  void executeTask(std::shared_ptr<T> data) {
    if (data != nullptr) {
      this->inputBk->executeTask(data);
    }
  }

  std::string getName() {
    return name;
  }

  ITask<T, U> *copy() {
    return new ExecutionPipeline<T, U>(this->numPipelinesExec,
                                       this->graph->copy(this->getPipelineId(), this->getNumPipelines()),
                                       this->inputRules, this->name, this->waitForInit);
  }

  void debug() {
    HTGS_DEBUG(this->getName() << " " << numPipelinesExec << " pipelines; details:");
    inputBk->debug();
  }

  virtual void gatherProfileData(std::map<AnyTaskManager *, TaskManagerProfile *> *taskManagerProfiles) override {
    // Gather profile data for each graph
    if (graphs->size() > 0) {
      for (auto g : *graphs) {
        g->gatherProfilingData(taskManagerProfiles);
      }
    } else {
      graph->gatherProfilingData(taskManagerProfiles);
    }

  }

  void printProfile() override {
    for (auto g : *graphs) {
      g->printProfile();
    }
  }

//#ifdef PROFILE
//  std::string getDotProfile(int flags,
//                            std::unordered_map<std::string, double> *mmap, double val,
//                            std::string desc, std::unordered_map<std::string, std::string> *colorMap)
//  {
//    std::ostringstream oss;
//
//    if (graphs->size() > 0)
//    {
//      for (auto g : *graphs)
//      {
//        oss << g->genProfileGraph(flags, mmap, desc, colorMap);
//      }
//    }
//    else
//    {
//      oss << graph->genProfileGraph(flags, mmap, desc, colorMap);
//    }
//
//    return oss.str();
//  }
//
//#endif

  std::string genCustomDot(ProfileUtils *profileUtils, int colorFlag) override {
    if (profileUtils == nullptr)
      return "";

    std::ostringstream oss;
    int pipeline = 0;
    for (auto g : *graphs) {
      double time = 0.0;
      if (colorFlag == DOTGEN_COLOR_COMP_TIME)
        time = (double) g->getGraphComputeTime();
      else if (colorFlag == DOTGEN_COLOR_WAIT_TIME)
        time = (double) this->getOwnerTaskManager()->getWaitTime();
      else if (colorFlag == DOTGEN_COLOR_MAX_Q_SZ)
        time = (double) this->getOwnerTaskManager()->getMaxQueueSize();

      oss << "subgraph cluster_" << this->getDotId() << std::to_string(pipeline) << " {" << std::endl;

      std::string color = profileUtils->getColorForTime(time);

      oss << (colorFlag != 0 ? "penwidth=5\ncolor=\"" + color + "\"" : "color=orange");
      oss << std::endl;
      pipeline++;

      oss << g->genCustomDotForTasks(profileUtils, colorFlag);

      oss << "}" << std::endl;
    }

    return oss.str();
  }

    virtual std::string genDotProducerEdgeToTask(std::map<std::shared_ptr<AnyConnector>, AnyITask *> &inputConnectorDotMap, int dotFlags) override
    {
      std::ostringstream oss;

//      oss << this->getOwnerTaskManager()->getInputConnector()->genDot(dotFlags);

      std::string inputRuleNames;
      int count = 0;
      for (std::shared_ptr<IRule<T, T>> rule : *inputRules) {
        if (count == 0)
          inputRuleNames = inputRuleNames + rule->getName();
        else
          inputRuleNames = inputRuleNames + ", " + rule->getName();
        count++;
      }

      if (graphs->size() > 0)
      {
        for (auto g : *graphs)
        {
          auto graphConsumer = g->getGraphConsumerEdge();

          oss << inputBk->getDotId()
              << " -> " << graphConsumer->getTaskManager(g)->getTaskFunction()->getConsumerDotIds() << "[label=\"" << inputRuleNames
              << "\"];" << std::endl;
        }
      } else {
          auto graphConsumer = graph->getGraphConsumerEdge();


          oss << inputBk->getDotId() << " -> "
              << graphConsumer->getTaskManager(graph)->getTaskFunction()->getConsumerDotIds() << "[label=\"" << inputRuleNames
              << "\"];"<< std::endl;


      }

      return oss.str();

    }

    virtual std::string genDotProducerEdgeFromConnector(std::shared_ptr<AnyConnector> connector, int flags)
    {
      return "";
    }

    virtual std::string genDotConsumerEdgeFromConnector(std::shared_ptr<AnyConnector> connector, int flags) override
    {

      if (this->getOwnerTaskManager()->getInputConnector() != nullptr && connector != nullptr &&
      this->getOwnerTaskManager()->getInputConnector() == connector) {
        return connector->getDotId() + " -> " + inputBk->getDotId() + ";\n";
      }
      return "";
    }

    std::string getConsumerDotIds() override {

    std::ostringstream oss;

      // Get inputRule edge name
      std::string inputRuleNames;
      int count = 0;
      for (std::shared_ptr<IRule<T, T>> rule : *inputRules) {
        if (count == 0)
          inputRuleNames = inputRuleNames + rule->getName();
        else
          inputRuleNames = inputRuleNames + ", " + rule->getName();
        count++;
      }

      oss << inputBk->getDotId();

//      if (this->graphs->size() > 0)
//      {
//        oss << "{";
//        for (auto g : *this->graphs) {
//          oss << g->getGraphConsumerTaskManager()->getTaskFunction()->getConsumerDotIds() << ";";
//        }
//        oss << "} ";
//      } else {
//        oss << graph->getGraphConsumerTaskManager()->getTaskFunction()->getConsumerDotIds();
//      }

      oss << " [label=\"" << inputRuleNames
              << "\"]" << std::endl;

      return oss.str();
    }

//    std::string genDotConsumerEdgeFromConnector(std::shared_ptr<AnyConnector> connector, int flags) override
//    {
//      return "";
//    }
//
//
//    std::string genDotProducerEdgeToTask(std::map<std::shared_ptr<AnyConnector>, AnyITask *> &inputConnectorDotMap, int dotFlags) override
//    {
//      std::ostringstream oss;
//
//      oss << this->getOwnerTaskManager()->getInputConnector()->genDot(dotFlags);
//
//      std::string inputRuleNames;
//      int count = 0;
//      for (std::shared_ptr<IRule<T, T>> rule : *inputRules) {
//        if (count == 0)
//          inputRuleNames = inputRuleNames + rule->getName();
//        else
//          inputRuleNames = inputRuleNames + ", " + rule->getName();
//        count++;
//      }
//
//      if (graphs->size() > 0)
//      {
//        for (auto g : *graphs)
//        {
//          oss << this->getOwnerTaskManager()->getInputConnector()->getDotId()
//              << " -> " << g->getGraphConsumerTaskManager()->getTaskFunction()->getDotId() << "[label=\"" << inputRuleNames
//              << "\"];" << std::endl;
//        }
//      } else {
//        oss << this->getOwnerTaskManager()->getInputConnector()->getDotId() << " -> "
//            << graph->getGraphConsumerTaskManager()->getTaskFunction()->getDotId() << "[label=\"" << inputRuleNames
//            << "\"];"<< std::endl;
//      }
//
//      return oss.str();
//    }

  std::string genDot(int flags,
                     std::string dotId,
                     std::shared_ptr<AnyConnector> input,
                     std::shared_ptr<AnyConnector> output) override {
    std::ostringstream oss;

    std::string aliveLabel =
        ((flags & DOTGEN_FLAG_SHOW_TASK_LIVING_STATUS) != 0) ? ("\\nLiving threads: " + std::to_string(this->getOwnerTaskManager()->getThreadsRemaining())) : "";
    std::string inOutLabel =
        (((flags & DOTGEN_FLAG_SHOW_IN_OUT_TYPES) != 0) ? ("\\nin: " + this->inTypeName() + "\\nout: "
                                                              + this->outTypeName()) : "");
    oss << inputBk->getDotId() + "[label=\"" + "Execution Pipeline Bookkeeper";
    oss << inOutLabel + "\\n";
    oss << aliveLabel;
    oss << "\",shape=" + this->getDotShape();
    oss << ",style=filled";
    oss << ",fillcolor=lightgrey";
    oss << ",color=orange";
    oss << ",width=.2,height=.2];\n";


    if ((flags & DOTGEN_FLAG_SHOW_CONNECTORS) != 0 || (flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) != 0) {
      oss << input->genDot(flags);
    }
    // Get inputRule edge name
    std::string inputRuleNames;
    int count = 0;
    for (std::shared_ptr<IRule<T, T>> rule : *inputRules) {
      if (count == 0)
        inputRuleNames = inputRuleNames + rule->getName();
      else
        inputRuleNames = inputRuleNames + ", " + rule->getName();
      count++;
    }

    if ((flags & DOTGEN_FLAG_SHOW_CONNECTORS) != 0 || (flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) != 0) {
      // Draw decomposition rule edge to each graph
      oss << input->getDotId() << " -> " << inputBk->getDotId() << ";" << std::endl;
      if (graphs->size() > 0) {
        for (auto g : *graphs) {

          oss << inputBk->getDotId() << " -> " << g->getInputConnector()->getDotId() << "[label=\"" << inputRuleNames
              << "\"];" << std::endl;
        }
      } else {
        oss << inputBk->getDotId() << " -> " << graph->getInputConnector()->getDotId() << "[label=\"" << inputRuleNames
            << "\"];" << std::endl;
      }
    }

    // Setup subgraph to draw execPipeline graphs
    if (this->graphs->size() > 0) {
      int pipeline = 0;
      for (auto g : *graphs) {
        std::string computeTimeStr = g->getGraphComputeTime() == 0 ? "" : "Compute time: " + std::to_string((double)g->getGraphComputeTime() / 1000000.0) + " s\\n";
        std::string createTimeStr = g->getGraphCreationTime() == 0 ? "" : "Creation time: " + std::to_string((double)g->getGraphCreationTime() / 1000000.0) + " s\\n";

        oss << "subgraph cluster_" << dotId << std::to_string(pipeline) << " {" << std::endl;
        oss << "label=\"" << getName() << std::to_string(pipeline) << "\\n" << computeTimeStr << createTimeStr << "\";" << std::endl;
        oss << "style=\"dashed\";" << std::endl;
        oss << "style =\"filled\";" << std::endl;
        oss << "fillcolor=lightgrey;" << std::endl;
        oss << "color=orange;" << std::endl;
        oss << g->genDotGraphContent(flags);
        oss << "}" << std::endl;
        pipeline++;

        std::string clean = cleanupVisualization(g, oss.str());

        oss.str("");
        oss.clear();
        oss << clean;
      }
    } else {
      oss << "subgraph cluster_" << dotId << " {" << std::endl;
      oss << "label=\"" << getName() << " x" << this->numPipelinesExec << "\";" << std::endl;
      oss << "style=\"dashed\";" << std::endl;
      oss << "style =\"filled\";" << std::endl;
      oss << "fillcolor=lightgrey;" << std::endl;
      oss << "color=orange;" << std::endl;

      graph->setOutputConnector(output);
      oss << graph->genDotGraphContent(flags);
      oss << "}" << std::endl;

      std::string clean = cleanupVisualization(graph, oss.str());

      oss.str("");
      oss.clear();
      oss << clean;
    }

    return oss.str();

  }

 private:

  std::string cleanupVisualization(TaskGraphConf<T, U> *graph, std::string str)
  {
    std::istringstream iss(str);


    std::ostringstream ossFinal;

    auto outputConnectorName = graph->getOutputConnector()->getDotId();

    std::string line;
    std::vector<std::string> savedLines;
    while(getline(iss, line))
    {
      if (line.find(outputConnectorName) == std::string::npos)
      {
        ossFinal << line << std::endl;
      }
      else
      {
        savedLines.push_back(line);
      }
    }

    for(std::string line2 : savedLines)
    {
      ossFinal << line2 << std::endl;
    }

    return ossFinal.str();
  }

  size_t numPipelinesExec; 
  Bookkeeper<T> *inputBk; 
  TaskGraphConf<T, U> *graph; 
  std::shared_ptr<IRuleList<T, T>> inputRules; 
  std::vector<TaskGraphRuntime *>
      *runtimes; 
  std::vector<TaskGraphConf<T, U> *> *graphs; 
  bool waitForInit; 
  std::string name; 
};
}

#endif //HTGS_EXECUTIONPIPELINE_H