TaskGraphConf.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_TASKGRAPHCONF_HPP
#define HTGS_TASKGRAPHCONF_HPP

#include <htgs/core/graph/edge/ProducerConsumerEdge.hpp>
#include <htgs/core/graph/edge/GraphTaskProducerEdge.hpp>
#include <htgs/core/graph/edge/RuleEdge.hpp>
#include <htgs/core/graph/edge/GraphEdge.hpp>
#include <htgs/core/graph/edge/MemoryEdge.hpp>
#include <htgs/core/comm/TaskGraphCommunicator.hpp>
#include <htgs/core/graph/profile/TaskGraphProfiler.hpp>

#include <htgs/api/ExecutionPipeline.hpp>
#include <htgs/api/TGTask.hpp>
#include <htgs/core/graph/edge/GraphRuleProducerEdge.hpp>
#include <htgs/core/graph/edge/GraphTaskConsumerEdge.hpp>

#ifdef USE_CUDA
#include <htgs/core/memory/CudaMemoryManager.hpp>
#endif

#ifdef WS_PROFILE
#include <htgs/core/graph/profile/CustomProfile.hpp>
#include <htgs/core/graph/profile/WebSocketProfiler.hpp>
#endif

namespace htgs {

  template <class T, class U>
  class ExecutionPipeline;

  template <class T, class U>
  class TGTask;

template<class T, class U>
class TaskGraphConf : public AnyTaskGraphConf {
  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:

  TaskGraphConf() : AnyTaskGraphConf(0, 1, "") {
    this->input = std::shared_ptr<Connector<T>>(new Connector<T>());
    this->output = std::shared_ptr<Connector<U>>(new Connector<U>());

    this->input->incrementInputTaskCount();
    this->graphConsumerEdge = nullptr;
    this->graphProducerEdges = new std::list<GraphEdge<U> *>();

    this->edges = new std::list<EdgeDescriptor *>();

    // TODO: Delete or Add #ifdef
//    this->taskConnectorCommunicator = new TaskGraphCommunicator(nullptr, this->getAddress());

#ifdef WS_PROFILE
    // Create web socket profiler task
    WebSocketProfiler *profileTask = new WebSocketProfiler();

    // Create input connector and task manager
    std::shared_ptr<Connector<ProfileData>> wsConnector(new Connector<ProfileData>());
    this->wsProfileTaskManager= new TaskManager<ProfileData, VoidData>(profileTask,
                                                                       profileTask->getNumThreads(),
                                                                       profileTask->isStartTask(),
                                                                       profileTask->isPoll(),
                                                                       profileTask->getMicroTimeoutTime(),
                                                                       0,
                                                                       1,
                                                                       "0");

    this->wsProfileTaskManager->setInputConnector(wsConnector);
    wsConnector->incrementInputTaskCount();

    // Add task to communicator
    TaskNameConnectorPair pair("0:" + this->wsProfileTaskManager->getName(), wsConnector);

    this->taskConnectorCommunicator->addTaskNameConnectorPair(pair);

    // Send root graph creation
    std::shared_ptr<ProfileData> createGraphData(new CreateNodeProfile(this, nullptr, ""));
    this->sendProfileData(createGraphData);

    std::cout << "Creating " << this->wsProfileTaskManager->getName() << " with connector addr: " << wsConnector << std::endl;

#endif

  }

  TaskGraphConf(size_t pipelineId,
                size_t numPipelines,
                std::string baseAddress
  // TODO: Delete or Add #ifdef
//                TaskGraphCommunicator *parentCommunicator
#ifdef WS_PROFILE
                , TaskManager<ProfileData, VoidData> *wsProfileTaskManager
#endif
  ) : AnyTaskGraphConf(pipelineId, numPipelines, baseAddress) {
    this->input = std::shared_ptr<Connector<T>>(new Connector<T>());
    this->output = std::shared_ptr<Connector<U>>(new Connector<U>());

    this->input->incrementInputTaskCount();

    graphConsumerEdge = nullptr;
    graphProducerEdges = new std::list<GraphEdge<U> *>();

    edges = new std::list<EdgeDescriptor *>();

    // TODO: Delete or Add #ifdef
//    taskConnectorCommunicator = new TaskGraphCommunicator(parentCommunicator, this->getAddress());

#ifdef WS_PROFILE
    this->wsProfileTaskManager = wsProfileTaskManager;
    this->wsProfileThread = nullptr;
#endif

  }

  ~TaskGraphConf() override {
    for (auto edge : *edges) {
      if (edge != nullptr) {
        delete edge;
        edge = nullptr;
      }
    }
    delete edges;
    edges = nullptr;

#ifdef WS_PROFILE
    if (wsProfileThread != nullptr)
    {
      this->wsProfileTaskManager->getInputConnector()->producerFinished();
      this->wsProfileTaskManager->getInputConnector()->wakeupConsumer();
      this->wsProfileThread->join();

      delete wsProfileTaskManager;
      wsProfileTaskManager = nullptr;

      delete wsProfileThread;
      wsProfileThread = nullptr;

      delete runtimeThread;
      runtimeThread = nullptr;
    }
#endif

#ifdef USE_NVTX
    if (this->getAddress() == "0")
    {
      delete profiler;
      profiler = nullptr;
    }
#endif

    // TODO: Delete or Add #ifdef
//    if (taskConnectorCommunicator != nullptr)
//      taskConnectorCommunicator->terminateGracefully();
//
//    delete taskConnectorCommunicator;
//    taskConnectorCommunicator = nullptr;

    if (graphConsumerEdge)
    {
      delete graphConsumerEdge;
      graphConsumerEdge = nullptr;
    }

    for (auto edge : *graphProducerEdges) {
      if (edge != nullptr) {
        delete edge;
        edge = nullptr;
      }
    }

    delete graphProducerEdges;
    graphProducerEdges = nullptr;
  }

  AnyTaskGraphConf *copy() override {
    return copy(this->getPipelineId(), this->getNumPipelines());
  }

  TaskGraphConf<T, U> *copy(size_t pipelineId, size_t numPipelines) {
    return copy(pipelineId, numPipelines, nullptr, nullptr, this->getAddress());
  }

  TaskGraphConf<T, U> *copy(size_t pipelineId,
                            size_t numPipelines,
                            std::shared_ptr<Connector<T>> input,
                            std::shared_ptr<Connector<U>> output,
                            std::string baseAddress) {
                            // TODO: Delete or Add #ifdef
                            //TaskGraphCommunicator *parentCommunicator) {
#ifdef WS_PROFILE
    TaskGraphConf<T, U> *graphCopy = new TaskGraphConf<T, U>(pipelineId, numPipelines, baseAddress, parentCommunicator, this->wsProfileTaskManager);

    // Send child graph creation
    std::shared_ptr<ProfileData> createGraphData(new CreateNodeProfile(graphCopy, this, ""));
    this->sendProfileData(createGraphData);
#else
    TaskGraphConf<T, U> *graphCopy = new TaskGraphConf<T, U>(pipelineId, numPipelines, baseAddress/*TODO: Delete or Add #ifdef , parentCommunicator*/);
#endif

    // Copy the tasks to form lookup between old ITasks and new copies
    graphCopy->copyTasks(this->getTaskManagers());

    if (input != nullptr) {
      graphCopy->setInputConnector(input);
    }

    if (output != nullptr) {
      graphCopy->setOutputConnector(output);
    }

    // Copy the graph producer and consumer tasks
    GraphEdge<T> *consumerEdge = graphConsumerEdge->copy(graphCopy);
    consumerEdge->applyEdge(graphCopy);
    graphCopy->setGraphConsumerEdge(consumerEdge);

    for (auto producerEdge : *this->graphProducerEdges)
    {
      GraphEdge<U> *producerEdgeCopy = producerEdge->copy(graphCopy);
      producerEdgeCopy->applyEdge(graphCopy);
      graphCopy->addGraphProducerEdge(producerEdgeCopy);
    }



//    graphCopy->copyAndUpdateGraphConsumerTask(this->graphConsumerEdge);
//    graphCopy->copyAndUpdateGraphProducerTasks(this->graphProducerEdges);

    for (EdgeDescriptor *edgeDescriptor : *edges) {
      // Copy the edge, using the graph copy as a reference for where to get task copies
      EdgeDescriptor *edgeCopy = edgeDescriptor->copy(graphCopy);

      // Apply the edge on the graph copy
      edgeCopy->applyEdge(graphCopy);

      graphCopy->addEdgeDescriptor(edgeCopy);
    }

    return graphCopy;
  }

  template<class V, class W, class X>
  void addEdge(ITask<V, W> *producer, ITask<W, X> *consumer) {
    auto pce = new ProducerConsumerEdge<V, W, X>(producer, consumer);
    pce->applyEdge(this);
    this->addEdgeDescriptor(pce);
  }

  template<class V, class IRuleType, class W, class X>
  void addRuleEdge(Bookkeeper<V> *bookkeeper, std::shared_ptr<IRuleType> rule, ITask<W, X> *consumer) {
    static_assert(std::is_base_of<IRule<V, W>, IRuleType>::value,
                  "Type mismatch for IRule<V, W>, V must match the input type of the bookkeeper and W must match the input type of the consumer!");
    std::shared_ptr<IRule<V, W>> ruleCast = std::static_pointer_cast<IRule<V, W>>(rule);
    auto re = new RuleEdge<V, W, X>(bookkeeper, ruleCast, consumer);
    re->applyEdge(this);
    this->addEdgeDescriptor(re);
  }

  template<class V, class W, class X>
  void addRuleEdge(Bookkeeper<V> *bookkeeper, IRule<V, W> *iRule, ITask<W, X> *consumer) {
    std::shared_ptr<IRule<V, W>> rule = super::getIRule(iRule);

    auto re = new RuleEdge<V, W, X>(bookkeeper, rule, consumer);
    re->applyEdge(this);
    this->addEdgeDescriptor(re);
  }

#ifdef USE_CUDA

  template<class V, class IMemoryAllocatorType>
  void addCudaMemoryManagerEdge(std::string name,
                                AnyITask *getMemoryTask,
                                std::shared_ptr<IMemoryAllocatorType> allocator,
                                size_t memoryPoolSize,
                                MMType type,
                                int *gpuIds) {
    static_assert(std::is_base_of<IMemoryAllocator<V>, IMemoryAllocatorType>::value,
                  "Type mismatch for allocator, allocator must be a MemoryAllocator!");

    std::shared_ptr<IMemoryAllocator<V>> memAllocator = std::static_pointer_cast<IMemoryAllocator<V>>(allocator);

    MemoryManager<V> *memoryManager = new CudaMemoryManager<V>(name, gpuIds, memoryPoolSize, memAllocator, type);

    MemoryEdge<V> *memEdge = new MemoryEdge<V>(name, getMemoryTask, memoryManager);
    memEdge->applyEdge(this);
    this->addEdgeDescriptor(memEdge);
  }




  template<class V>
  void addCudaMemoryManagerEdge(std::string name,
                                AnyITask *getMemoryTask,
                                IMemoryAllocator<V> *allocator,
                                size_t memoryPoolSize,
                                MMType type,
                                int *gpuIds) {

    std::shared_ptr<IMemoryAllocator<V>> memAllocator = this->getMemoryAllocator(allocator);

    MemoryManager<V> *memoryManager = new CudaMemoryManager<V>(name, gpuIds, memoryPoolSize, memAllocator, type);

    MemoryEdge<V> *memEdge = new MemoryEdge<V>(name, getMemoryTask, memoryManager);
    memEdge->applyEdge(this);
    this->addEdgeDescriptor(memEdge);
  }
#endif


  template <class V>
  void addRuleEdgeAsGraphProducer(Bookkeeper<V> *bookkeeper, IRule<V, U> *iRule) {
    std::shared_ptr<IRule<V, U>> rule = super::getIRule(iRule);

    auto gore = new GraphRuleProducerEdge<V, U>(bookkeeper, rule, this->output);
    gore->applyEdge(this);
    this->addGraphProducerEdge(gore);
  }

    template <class V>
    void addCustomMemoryManagerEdge(AnyITask *getMemoryTask, MemoryManager<V> *memoryManager)
    {
      MemoryEdge<V> *memEdge = new MemoryEdge<V>(memoryManager->getMemoryManagerName(), getMemoryTask, memoryManager);
      memEdge->applyEdge(this);
      this->addEdgeDescriptor(memEdge);
    }

  template<class V, class IMemoryAllocatorType>
  void addMemoryManagerEdge(std::string name, AnyITask *getMemoryTask,
                            std::shared_ptr<IMemoryAllocatorType> allocator, size_t memoryPoolSize, MMType type) {
    static_assert(std::is_base_of<IMemoryAllocator<V>, IMemoryAllocatorType>::value,
                  "Type mismatch for allocator, allocator must be a MemoryAllocator!");

    std::shared_ptr<IMemoryAllocator<V>> memAllocator = std::static_pointer_cast<IMemoryAllocator<V>>(allocator);

    MemoryManager<V> *memoryManager = new MemoryManager<V>(name, memoryPoolSize, memAllocator, type);

    MemoryEdge<V> *memEdge = new MemoryEdge<V>(name, getMemoryTask, memoryManager);
    memEdge->applyEdge(this);
    this->addEdgeDescriptor(memEdge);
  }

  template<class V>
  void addMemoryManagerEdge(std::string name,
                            AnyITask *getMemoryTask,
                            IMemoryAllocator<V> *allocator,
                            size_t memoryPoolSize,
                            MMType type) {

    std::shared_ptr<IMemoryAllocator<V>> memAllocator = super::getMemoryAllocator(allocator);

    MemoryManager<V> *memoryManager = new MemoryManager<V>(name, memoryPoolSize, memAllocator, type);

    MemoryEdge<V> *memEdge = new MemoryEdge<V>(name, getMemoryTask, memoryManager);
    memEdge->applyEdge(this);
    this->addEdgeDescriptor(memEdge);
  }

//  AnyTaskManager *getGraphConsumerTaskManager() override {
//
//    return this->graphConsumerEdge->getTaskManager(this);
//  }

//  std::list<AnyTaskManager *> *getGraphProducerTaskManagers() override {
//    return this->graphProducerEdges;
//  }

  std::shared_ptr<AnyConnector> getInputConnector() override {
    return this->input;
  }

  std::shared_ptr<AnyConnector> getOutputConnector() override {
    return this->output;
  }

  // TODO: rename virtual function to capture functionality with initializing the profiler
  void updateTaskManagersAddressingAndPipelines() {
    for (auto t : *this->getTaskManagers()) {
      t->updateAddressAndPipelines(this->getAddress(), this->getPipelineId(), this->getNumPipelines());


#ifdef USE_NVTX
      if (this->getAddress() == "0") {
        nvtxDomainHandle_t graphDomain = nvtxDomainCreateA(TASK_GRAPH_PREFIX_NAME);
#ifdef USE_MINIMAL_NVTX
        profiler = new NVTXProfiler(TASK_GRAPH_PREFIX_NAME, nullptr, graphDomain, graphDomain, graphDomain, graphDomain, graphDomain, graphDomain);
#else
        profiler = new NVTXProfiler(TASK_GRAPH_PREFIX_NAME, graphDomain, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr);
#endif
      }
#endif

//      std::string taskAddressName = this->address + ":" + t->getName();
//      this->taskConnectorNameMap->insert(TaskNameConnectorPair(taskAddressName, t->getInputConnector()));
    }
  }

  // TODO: Delete or Add #ifdef
//  TaskGraphCommunicator *getTaskGraphCommunicator() const { return this->taskConnectorCommunicator; }

//  void updateCommunicator() override {
//
//    auto taskNameConnectorMap = this->getTaskConnectorNameMap();
//
//    // Send the map to the taskGraphCommunicator
//    this->taskConnectorCommunicator->addTaskNameConnectorMap(taskNameConnectorMap);
//
//    for (auto t : *this->getTaskManagers()) {
//      t->setTaskGraphCommunicator(this->taskConnectorCommunicator);
//    }
//
//#ifdef WS_PROFILE
//    if (this->getAddress() == "0") {
//      std::cout << "Launching threaed" << std::endl;
//      // Create thread
//      std::shared_ptr<std::atomic_size_t>
//          atomicNumThreads = std::shared_ptr<std::atomic_size_t>(new std::atomic_size_t(1));
//      runtimeThread = new TaskManagerThread(0, this->wsProfileTaskManager, atomicNumThreads);
//      this->wsProfileThread = new std::thread(&TaskManagerThread::run, runtimeThread);
//
//      WebSocketProfiler *profileTask = (WebSocketProfiler *)this->wsProfileTaskManager->getTaskFunction();
//      profileTask->waitForConnection();
//
//    }
//#endif
//
//
//  }

//  void setInputConnector(std::shared_ptr<Connector<T>> input) {
//    this->input = input;
//  }

//  void setOutputConnector(std::shared_ptr<Connector<U>> output) {
//    this->output = output;
//  }

  void incrementGraphProducer() {
    this->input->incrementInputTaskCount();
  }

  void finishedProducingData() {
    this->input->producerFinished();
    if (this->input->getProducerCount() == 0) {
      this->input->wakeupConsumer();
    }
#ifdef WS_PROFILE
    std::shared_ptr<ProfileData> updateStatus(new ChangeStatusProfile(this->input.get(), StatusCode::DECREMENT));
    this->sendProfileData(updateStatus);
#endif
  }

  template<class W>
  void setGraphConsumerTask(ITask<T, W> *task) {
    GraphTaskConsumerEdge<T, W> *consumerEdge = new GraphTaskConsumerEdge<T, W>(task, this->input);

    consumerEdge->applyEdge(this);
    if (this->graphConsumerEdge)
    {
      delete graphConsumerEdge;
    }

    this->graphConsumerEdge = consumerEdge;

#ifdef WS_PROFILE
    // Add nodes
    std::shared_ptr<ProfileData> connectorData(new CreateConnectorProfile(input.get(), this, input->getProducerCount(), "Graph Input"));
    std::shared_ptr<ProfileData> consumerData(new CreateNodeProfile(task, this, task->getName()));

    this->sendProfileData(consumerData);
    this->sendProfileData(connectorData);

    std::shared_ptr<ProfileData> connectorConsumerData(new CreateEdgeProfile(input.get(), task, "", nullptr));

    this->sendProfileData(connectorConsumerData);
#endif

  }

  template<class W>
  void addGraphProducerTask(ITask<W, U> *task) {
    GraphTaskProducerEdge<W, U> *producerEdge = new GraphTaskProducerEdge<W, U>(task, this->output);
    producerEdge->applyEdge(this);

    this->graphProducerEdges->push_back(producerEdge);

#ifdef WS_PROFILE
    // Add nodes
    std::shared_ptr<ProfileData> connectorData(new CreateConnectorProfile(output.get(), this, output->getProducerCount(), "Graph Output"));
    std::shared_ptr<ProfileData> producerData(new CreateNodeProfile(task, this, task->getName()));

    this->sendProfileData(producerData);
    this->sendProfileData(connectorData);

    std::shared_ptr<ProfileData> connectorProducerData(new CreateEdgeProfile(task, output.get(), "", nullptr));

    this->sendProfileData(connectorProducerData);
#endif

  }

  void produceData(T *data) {
    std::shared_ptr<T> dataPtr = std::shared_ptr<T>(data);
    this->input->produceData(dataPtr);
  }

  void produceData(std::shared_ptr<T> data) {
    this->input->produceData(data);
  }

  void produceData(std::list<std::shared_ptr<T>> *dataList) {
    this->input->produceData(dataList);
    if (this->input->isInputTerminated())
      this->input->wakeupConsumer();
  }

  std::shared_ptr<U> consumeData() {
#ifdef USE_NVTX
    nvtxRangeId_t id = profiler->startRangeWaiting(this->output->getQueueSize());
#endif
    std::shared_ptr<U> data = this->output->consumeData();
#ifdef USE_NVTX
    profiler->endRangeWaiting(id);
#endif
    return data;
  }

  std::shared_ptr<U> pollData(size_t microTimeout) {
#ifdef USE_NVTX
    nvtxRangeId_t id = profiler->startRangeWaiting(this->output->getQueueSize());
#endif
    std::shared_ptr<U> data = this->output->pollConsumeData(microTimeout);
#ifdef USE_NVTX
    profiler->endRangeWaiting(id);
#endif
    return data;
  }

  bool isOutputTerminated() {
    return this->output->isInputTerminated();
  }

  void setOutputConnector(std::shared_ptr<AnyConnector> connector) {
    if (graphProducerEdges != nullptr) {
      for (auto producerEdge : *graphProducerEdges)
        producerEdge->updateEdge(std::dynamic_pointer_cast<Connector<U>>(connector), this);
    }

    this->output = std::dynamic_pointer_cast<Connector<U>>(connector);

  }

    void setInputConnector(std::shared_ptr<AnyConnector> connector) {
      if (graphConsumerEdge != nullptr) {
        graphConsumerEdge->updateEdge(std::dynamic_pointer_cast<Connector<T>>(connector), this);
      }

      this->input = std::dynamic_pointer_cast<Connector<T>>(connector);

    }


  template<class V>
  [[gnu::deprecated("Replaced by calling 'releaseMemory' directory with htgs::MemoryData (or m_data_t)")]]
  void releaseMemory(m_data_t<V> memory) {
    memory->releaseMemory();
    // TODO: Delete or Add #ifdef
//    std::shared_ptr<DataPacket> dataPacket = std::shared_ptr<DataPacket>(new DataPacket("TaskGraph",
//                                                                                        this->getAddress(),
//                                                                                        memory->getMemoryManagerName(),
//                                                                                        memory->getAddress(),
//                                                                                        memory));
#ifdef USE_NVTX
    profiler->addReleaseMarker();
#endif

//    this->taskConnectorCommunicator->produceDataPacket(dataPacket);


  }

  std::string genCustomDotForTasks(ProfileUtils *profileUtils, int colorFlag)
  {
    std::ostringstream oss;

    for (AnyTaskManager *bTask : *this->getTaskManagers()) {
      oss << bTask->getTaskFunction()->genCustomDot(profileUtils, colorFlag);
    }
    return oss.str();
  }

  std::string genDotGraphEdgesWithoutConnectors(std::map<AnyTaskManager *, TaskManagerProfile *> *allTaskManagerProfiles, int flags)
  {


    std::map<std::shared_ptr<AnyConnector>, AnyTaskManager *> inputConnectorMap;
    std::map<std::shared_ptr<AnyConnector>, AnyITask *> inputConnectorDotMap;

    for (auto tmanProfiles : *allTaskManagerProfiles) {
      auto bTask = tmanProfiles.first;
      if (bTask->getThreadId() == 0) {
        inputConnectorMap.insert(
            std::pair<std::shared_ptr<AnyConnector>, AnyTaskManager *>(bTask->getInputConnector(), bTask));
        inputConnectorDotMap.insert(
            std::pair<std::shared_ptr<AnyConnector>, AnyITask *>(bTask->getInputConnector(), bTask->getTaskFunction()));
      }
    }

    // Add in the graph's input/output connectors as well
    inputConnectorMap.insert(std::pair<std::shared_ptr<AnyConnector>, AnyTaskManager *>(this->getInputConnector(), nullptr));
    inputConnectorMap.insert(std::pair<std::shared_ptr<AnyConnector>, AnyTaskManager *>(this->getOutputConnector(), nullptr));

//    inputConnectorDotMap.insert(std::pair<std::shared_ptr<AnyConnector>, std::string>(this->getInputConnector(), this->getInputConnector()->getDotId()));
//    inputConnectorDotMap.insert(std::pair<std::shared_ptr<AnyConnector>, std::string>(this->getOutputConnector(), this->getOutputConnector()->getDotId()));

    std::ostringstream oss;

    for (auto tmanProfiles : *allTaskManagerProfiles)
    {
      auto bTask = tmanProfiles.first;

      if (bTask->getThreadId() == 0) {
        oss << bTask->getTaskFunction()->genDotProducerEdgeToTask(inputConnectorDotMap, flags);
        oss << bTask->getTaskFunction()->genDotConsumerEdgeFromConnector(this->getInputConnector(), flags);
        oss << bTask->getTaskFunction()->genDotProducerEdgeFromConnector(this->getOutputConnector(), flags);
      }

//      auto findConsumer = inputConnectorMap.find(bTask->getOutputConnector());
//      if (findConsumer != inputConnectorMap.end())
//      {
//        // bTask is producing for findConsumer
//        oss << bTask->getTaskFunction()->genDotProducerEdgeWithoutConnector(findConsumer->second->getTaskFunction()->getDotId(), flags);
//      }
//
//      if (this->getInputConnector() == bTask->getInputConnector())
//      {
        // input connector is connected to the task
//        oss << this->getInputConnector()->getDotId() << " -> " << bTask->getTaskFunction()->getDotId() << ";" << std::endl;
//        oss << bTask->getTaskFunction()->genDotConsumerEdgeWithoutConnector(this->getInputConnector()->getDotId(), flags);
//      }
//
//      if (this->getOutputConnector() == bTask->getOutputConnector())
//      {
//        // the output connector for the graph is connected to the task
//        oss <<bTask->getTaskFunction()->genDotProducerEdgeWithoutConnector(this->getOutputConnector()->getDotId(), flags);
//      }


    }

    return oss.str();

  }

  std::string genDotGraph(int flags, int colorFlag, std::string graphTitle = "", std::string customTitleText = "") override {

    std::ostringstream oss;

    // Create header info for graphViz dot file
    oss << "digraph { rankdir=\"TB\"" << std::endl;
    oss << "forcelabels=true;" << std::endl;
    oss << "node[shape=record, fontsize=10, fontname=\"Verdana\"];" << std::endl;
    oss << "edge[fontsize=10, fontname=\"Verdana\"];" << std::endl;

    std::string graphTitleStr = graphTitle == "" ? "" : (graphTitle + "\\n");
    std::string computeTimeStr = this->getGraphComputeTime() == 0 ? "" : "Compute time: " + std::to_string((double)this->getGraphComputeTime() / 1000000.0) + " s\\n";
    std::string createTimeStr = this->getGraphCreationTime() == 0 ? "" : "Creation time: " + std::to_string((double)this->getGraphCreationTime() / 1000000.0) + " s\\n";



    oss << "graph [compound=true, labelloc=top, labeljust=left, "
        << "label=\"" << graphTitleStr << computeTimeStr << createTimeStr <<customTitleText << "\",pad=\"0.5\", nodesep=\"0.5\", ranksep=\"0\"];" << std::endl;


    // Gather profile data
    TaskGraphProfiler profiler(flags);
    profiler.buildProfile(this);

    oss << genDotGraphContent(flags);
    if ((flags & DOTGEN_FLAG_SHOW_CONNECTORS) == 0 && (flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) == 0) {
      oss << genDotGraphEdgesWithoutConnectors(profiler.getTaskManagerProfiles(), flags);
    }

    oss << profiler.genDotProfile(oss.str(), colorFlag);
    oss << genCustomDotForTasks(profiler.getProfileUtils(), colorFlag);

    if (this->graphConsumerEdge != nullptr)
      oss << this->getInputConnector()->getDotId() << "[label=\"Graph Input\\n"
          << ((flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) == 0 ? std::to_string(this->getInputConnector()->getProducerCount()) : "Active Producers: " + std::to_string(this->getInputConnector()->getProducerCount()))
          << ((flags & DOTGEN_FLAG_SHOW_CURRENT_Q_SZ) == 0 && (flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) == 0 ? "" : "\\nQueue Size: " + std::to_string(this->getInputConnector()->getQueueSize()))
          << (((DOTGEN_FLAG_SHOW_IN_OUT_TYPES & flags) != 0) ? ("\\n" + this->getInputConnector()->typeName()) : "")
          << "\"];" << std::endl;


//    if (getGraphProducerTaskManagers()->size() > 0)
    if (this->graphProducerEdges->size() > 0)
      oss << "{ rank = sink; " << this->getOutputConnector()->getDotId() << "[label=\"Graph Output\\n"
        << ((flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) == 0 ? std::to_string(this->getInputConnector()->getProducerCount()) : "Active Producers: " + std::to_string(this->getOutputConnector()->getProducerCount()))
        << ((flags & DOTGEN_FLAG_SHOW_CURRENT_Q_SZ) == 0 && (flags & DOTGEN_FLAG_SHOW_CONNECTOR_VERBOSE) == 0 ? "" : "\\nQueue Size: " + std::to_string(this->getOutputConnector()->getQueueSize()))
          << (((DOTGEN_FLAG_SHOW_IN_OUT_TYPES & flags) != 0) ? ("\\n" + this->getOutputConnector()->typeName()) : "")
          << "\"]; }" << std::endl;

    if (oss.str().find("mainThread") != std::string::npos) {
      oss << "{ rank = sink; mainThread[label=\"Main Thread\", fillcolor = aquamarine4]; }" << std::endl;
    }

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

    return oss.str();
  }

  void debug() {
    HTGS_DEBUG("-----------------------------------------------");
    HTGS_DEBUG("TaskGraphConf -- num vertices: " << this->getTaskManagers()->size() << " -- DETAILS:");

    for (AnyTaskManager *t : *this->getTaskManagers()) {
      t->debug();
    }
    HTGS_DEBUG("-----------------------------------------------");
  }

#ifdef WS_PROFILE
  void sendProfileData(std::shared_ptr<ProfileData> profileData) override
  {
    this->wsProfileTaskManager->getInputConnector()->produceAnyData(profileData);
  }
#endif

  ExecutionPipeline<T, U> *createExecutionPipeline(size_t numPipelines, std::string name = "Execution Pipeline", bool waitForInit = true)
  {
    return new ExecutionPipeline<T, U>(numPipelines, this, name, waitForInit);
  }

  TGTask<T, U> *createTaskGraphTask(std::string name = "TGTask", bool waitForInit = true)
  {
    return new TGTask<T, U>(this, name, waitForInit);
  }

  GraphEdge<T> *getGraphConsumerEdge() const {
    return graphConsumerEdge;
  }

  std::list<GraphEdge<U> *> *getGraphProducerEdges() const {
    return graphProducerEdges;
  }

   private:




//  void copyAndUpdateGraphConsumerTask(AnyTaskManager *taskManager) {
//    if (taskManager != nullptr) {
//      AnyTaskManager *copy = this->getTaskManagerCopy(taskManager->getTaskFunction());
//      this->graphConsumerEdge = copy;
//      this->graphConsumerEdge->setInputConnector(this->input);
//      this->addTaskManager(this->graphConsumerEdge);
//
//#ifdef WS_PROFILE
//      // TODO: Ensure this is doable with visualizer, this is an update edge and remove node type task
//      // Add nodes
//#endif
//
//    }
//  }
//
//  void copyAndUpdateGraphProducerTasks(std::list<AnyTaskManager *> *taskManagers) {
//    for (auto taskManager : *taskManagers) {
//      if (taskManager != nullptr) {
//        AnyTaskManager *copy = this->getTaskManagerCopy(taskManager->getTaskFunction());
//
//        copy->setOutputConnector(this->output);
//        this->graphProducerEdges->push_back(copy);
//
//        this->output->incrementInputTaskCount();
//        this->addTaskManager(copy);
//
//#ifdef WS_PROFILE
//        // TODO: Ensure this is doable with visualizer, this is an update edge and remove node type task
//        // Add nodes
//#endif
//      }
//    }
//  }

  void setGraphConsumerEdge(GraphEdge<T> *consumerEdge)
  {
    this->graphConsumerEdge = consumerEdge;
  }

  void addGraphProducerEdge(GraphEdge<U> *producerEdge)
  {
    graphProducerEdges->push_back(producerEdge);
  }

  void addEdgeDescriptor(EdgeDescriptor *edge) {
    this->edges->push_back(edge);
  }

  typedef AnyTaskGraphConf super;

  std::list<EdgeDescriptor *> *
      edges; 

  GraphEdge<T> *graphConsumerEdge; 
  std::list<GraphEdge<U> *> *
      graphProducerEdges; 

  std::shared_ptr<Connector<T>> input; 
  std::shared_ptr<Connector<U>> output; 

  // TODO: Delete or Add #ifdef
//  TaskGraphCommunicator *taskConnectorCommunicator; //!< The task graph communicator for the task graph.


#ifdef WS_PROFILE
  TaskManager<ProfileData, VoidData> *wsProfileTaskManager; // !< The task manager for the web socket profiler
  std::thread *wsProfileThread; // !< The thread for the web socket profiler task manager
  TaskManagerThread *runtimeThread; 
#endif

#ifdef USE_NVTX
  NVTXProfiler *profiler;
#endif

};
}

#endif //HTGS_TASKGRAPHCONF_HPP