ExecutionFlowGraph.h

Go to the documentation of this file.

#ifndef GAUDIHIVE_EXECUTIONFLOWGRAPH_H
#define GAUDIHIVE_EXECUTIONFLOWGRAPH_H

// std includes
#include <vector>
#include <algorithm>
#include <unordered_map>
#include <chrono>
#include <fstream>
#include <sstream>

#include <boost/graph/adjacency_list.hpp>
#include <boost/graph/graphml.hpp>

// fwk includes
#include "AlgsExecutionStates.h"
#include "EventSlot.h"
#include "IGraphVisitor.h"
#include "GaudiKernel/Algorithm.h"
#include "GaudiKernel/CommonMessaging.h"

#include "CPUCruncher.h"

namespace boost {

  struct AlgoNodeStruct {
    AlgoNodeStruct () {}
    AlgoNodeStruct (const std::string& name, const int index = -1, const int& rank = -1, const double& runtime = -1, const double& eccentricity = -1.0) :
      m_name(name), m_index(index), m_rank(rank), m_runtime(runtime), m_eccentricity(eccentricity) {}
    std::string m_name;
    int m_index;
    int m_rank;
    double m_runtime;
    double m_eccentricity;
  };

  typedef boost::adjacency_list<boost::vecS, boost::vecS, boost::bidirectionalS, AlgoNodeStruct> ExecPlan;
  typedef graph_traits<ExecPlan>::vertex_descriptor AlgoVertex;
}

namespace concurrency {

  typedef AlgsExecutionStates::State State;
  class ExecutionFlowGraph;

  class ControlFlowNode {
  public:
    ControlFlowNode(ExecutionFlowGraph& graph, unsigned int nodeIndex, const std::string& name) :
      m_graph(&graph), m_nodeIndex(nodeIndex), m_nodeName(name) {};
    virtual ~ControlFlowNode() {};
    virtual void initialize(const std::unordered_map<std::string,unsigned int>& algname_index_map) = 0;
    virtual bool accept(IGraphVisitor& visitor) = 0;
    virtual bool promoteToControlReadyState(const int& slotNum,
                                            AlgsExecutionStates& states,
                                            std::vector<int>& node_decisions) const = 0;
    virtual int updateState(AlgsExecutionStates& states,
                            std::vector<int>& node_decisions) const = 0;
    virtual void printState(std::stringstream& output,
                            AlgsExecutionStates& states,
                            const std::vector<int>& node_decisions,
                            const unsigned int& recursionLevel) const = 0;
    const unsigned int& getNodeIndex() const { return m_nodeIndex; }
    const std::string& getNodeName() const { return m_nodeName; }
    virtual void updateDecision(const int& slotNum,
                                AlgsExecutionStates& states,
                                std::vector<int>& node_decisions,
                                const AlgorithmNode* requestor = nullptr) const = 0;
  public:
    ExecutionFlowGraph* m_graph;
  protected:
    std::string stateToString(const int& stateId) const;
    unsigned int m_nodeIndex;
    std::string m_nodeName;
  };


  class DecisionNode : public ControlFlowNode {
  public:
    DecisionNode(ExecutionFlowGraph& graph, unsigned int nodeIndex, const std::string& name, bool modeOR, bool allPass, bool isLazy) :
      ControlFlowNode(graph, nodeIndex, name),
      m_modeOR(modeOR), m_allPass(allPass), m_isLazy(isLazy), m_children()
      {};
    virtual ~DecisionNode();
    virtual void initialize(const std::unordered_map<std::string,unsigned int>& algname_index_map);
    virtual bool accept(IGraphVisitor& visitor);
    virtual bool promoteToControlReadyState(const int& slotNum,
                                            AlgsExecutionStates& states,
                                            std::vector<int>& node_decisions) const;
    virtual void updateDecision(const int& slotNum,
                                AlgsExecutionStates& states,
                                std::vector<int>& node_decisions,
                                const AlgorithmNode* requestor = nullptr) const;
    virtual int updateState(AlgsExecutionStates& states,
                            std::vector<int>& node_decisions) const;
    void addParentNode(DecisionNode* node);
    void addDaughterNode(ControlFlowNode* node);
    const std::vector<ControlFlowNode*>& getDaughters() const {return m_children;}
    virtual void printState(std::stringstream& output,
                            AlgsExecutionStates& states,
                            const std::vector<int>& node_decisions,
                            const unsigned int& recursionLevel) const;
  public:
    bool m_modeOR;
    bool m_allPass;
    bool  m_isLazy;
  private:
    std::vector<ControlFlowNode*> m_children;
    std::vector<DecisionNode*> m_parents;
  };

  class DataNode;

  class AlgorithmNode : public ControlFlowNode {
  public:
    AlgorithmNode(ExecutionFlowGraph& graph, unsigned int nodeIndex, const std::string& algoName, bool inverted, bool allPass) :
      ControlFlowNode(graph, nodeIndex, algoName),
      m_algoIndex(0),m_algoName(algoName),m_inverted(inverted),m_allPass(allPass),m_rank(-1)
      {};
    ~AlgorithmNode();
    virtual void initialize(const std::unordered_map<std::string,unsigned int>& algname_index_map);
    virtual bool accept(IGraphVisitor& visitor);
    void addParentNode(DecisionNode* node);

    void addSupplierNode(AlgorithmNode* node) { m_suppliers.push_back(node); }
    void addConsumerNode(AlgorithmNode* node) { m_consumers.push_back(node); }
    void attachAlgorithm(IAlgorithm* ialgo) { m_representatives.push_back(ialgo); }
    const std::vector<IAlgorithm*>& getAlgorithmRepresentatives () const { return m_representatives; }
    const std::vector<AlgorithmNode*>& getSupplierNodes() const {return m_suppliers;}
    const std::vector<AlgorithmNode*>& getConsumerNodes() const {return m_consumers;}

    void addOutputDataNode(DataNode* node);
    void addInputDataNode(DataNode* node);
    const std::vector<DataNode*>& getOutputDataNodes() const {return m_outputs;}
    const std::vector<DataNode*>& getInputDataNodes() const {return m_inputs;}
    void setRank(float& rank) {m_rank = rank;}
    const float& getRank() const {return m_rank;}

    const unsigned int& getAlgoIndex() const { return m_algoIndex; }
    bool dataDependenciesSatisfied(const int& slotNum) const;
    bool dataDependenciesSatisfied(AlgsExecutionStates& states) const;
    virtual int updateState(AlgsExecutionStates& states,
                            std::vector<int>& node_decisions) const;
    virtual bool promoteToControlReadyState(const int& slotNum,
                                            AlgsExecutionStates& states,
                                            std::vector<int>& node_decisions) const;
    bool promoteToDataReadyState(const int& slotNum, const AlgorithmNode* requestor = nullptr) const;
    virtual void updateDecision(const int& slotNum,
                                AlgsExecutionStates& states,
                                std::vector<int>& node_decisions,
                                const AlgorithmNode* requestor = nullptr) const;
    virtual void printState(std::stringstream& output,
                            AlgsExecutionStates& states,
                            const std::vector<int>& node_decisions,
                            const unsigned int& recursionLevel) const;
  private:
    unsigned int m_algoIndex;
    std::string m_algoName;
    bool m_inverted;
    bool m_allPass;
    std::vector<DecisionNode*> m_parents;

    std::vector<AlgorithmNode*> m_suppliers;
    std::vector<AlgorithmNode*> m_consumers;

    std::vector<DataNode*> m_outputs;
    std::vector<DataNode*> m_inputs;
    float m_rank;
    std::vector<IAlgorithm*> m_representatives;
  };

class DataNode {
public:
    DataNode(ExecutionFlowGraph& /*graph*/, const std::string& path): m_data_object_path(path) {}
    ~DataNode() {}
    const std::string& getPath() {return m_data_object_path;}
    void addProducerNode(AlgorithmNode* node) {
      if (std::find(m_producers.begin(),m_producers.end(),node) == m_producers.end())
        m_producers.push_back(node);
    }
    void addConsumerNode(AlgorithmNode* node) {
      if (std::find(m_consumers.begin(),m_consumers.end(),node) == m_consumers.end())
        m_consumers.push_back(node);
    }
    const std::vector<AlgorithmNode*>& getProducers() const {return m_producers;}
    const std::vector<AlgorithmNode*>& getConsumers() const {return m_consumers;}
private:
    std::string m_data_object_path;
    std::vector<AlgorithmNode*> m_producers;
    std::vector<AlgorithmNode*> m_consumers;
  };

typedef std::unordered_map<std::string,AlgorithmNode*> AlgoNodesMap;
typedef std::unordered_map<std::string,DecisionNode*> DecisionHubsMap;
typedef std::unordered_map<std::string,DataNode*> DataNodesMap;
typedef std::unordered_map<std::string,const DataObjectDescriptorCollection*> AlgoInputsMap;
typedef std::unordered_map<std::string,const DataObjectDescriptorCollection*> AlgoOutputsMap;

class ExecutionFlowManager;
struct IExecutionFlowGraph {
  virtual ~IExecutionFlowGraph() = default;
};

class ExecutionFlowGraph : public CommonMessaging<IExecutionFlowGraph> {
  friend ExecutionFlowManager;
public:
    ExecutionFlowGraph(const std::string& name, SmartIF<ISvcLocator> svc) :
     m_headNode(0), m_nodeCounter(0), m_svcLocator(svc), m_name(name), m_initTime(std::chrono::high_resolution_clock::now()),
     m_eventSlots(nullptr) {}
    ~ExecutionFlowGraph() override {
      if (m_headNode != 0) delete m_headNode;
    }
    StatusCode initialize(const std::unordered_map<std::string,unsigned int>& algname_index_map);
    StatusCode initialize(const std::unordered_map<std::string,unsigned int>& algname_index_map,
                          std::vector<EventSlot>& eventSlots);
    void registerIODataObjects(const Algorithm* algo);
    StatusCode buildDataDependenciesRealm();
    StatusCode buildAugmentedDataDependenciesRealm();
    void addHeadNode(const std::string& headName, bool modeOR, bool allPass, bool isLazy);
    StatusCode addAlgorithmNode(Algorithm* daughterAlgo, const std::string& parentName, bool inverted, bool allPass);
    template<class T>
    void attachAlgorithmsToNodes(const std::string& algo_name, const T& container) {
      auto node = getAlgorithmNode(algo_name);
      for (auto ialgoIt = container.unsafe_begin(); ialgoIt != container.unsafe_end(); ++ialgoIt)
        node->attachAlgorithm(*ialgoIt);
    }
    AlgorithmNode* getAlgorithmNode(const std::string& algoName) const;
    StatusCode addDataNode(const std::string& dataPath);
    DataNode* getDataNode(const std::string& dataPath) const;
    StatusCode addDecisionHubNode(Algorithm* daughterAlgo, const std::string& parentName, bool modeOR, bool allPass, bool isLazy);
    unsigned int getControlFlowNodeCounter() const {return m_nodeCounter;}
    void updateEventState(AlgsExecutionStates& states,
                          std::vector<int>& node_decisions) const;
    void updateDecision(const std::string& algo_name,
                        const int& slotNum,
                        AlgsExecutionStates& states,
                        std::vector<int>& node_decisions) const;
    void rankAlgorithms(IGraphVisitor& ranker) const;
    void printState(std::stringstream& output,
                    AlgsExecutionStates& states,
                    const std::vector<int>& node_decisions,
                    const unsigned int& recursionLevel) const {m_headNode->printState(output,states,node_decisions,recursionLevel);};
    const std::vector<AlgorithmNode*> getDataIndependentNodes() const;
    const std::string& name() const override {return m_name;}
    SmartIF<ISvcLocator>& serviceLocator() const override {return m_svcLocator;}
    const std::chrono::system_clock::time_point getInitTime() const {return m_initTime;};
    AlgsExecutionStates& getAlgoStates(const int& slotNum) const {return m_eventSlots->at(slotNum).algsStates;};
    std::vector<int>& getNodeDecisions(const int& slotNum) const {return m_eventSlots->at(slotNum).controlFlowState;}
    void dumpDataFlow() const;
    void dumpExecutionPlan();
    void addEdgeToExecutionPlan(const AlgorithmNode* u, const AlgorithmNode* v);

private:
    DecisionNode* m_headNode;
    AlgoNodesMap m_algoNameToAlgoNodeMap;
    DecisionHubsMap m_decisionNameToDecisionHubMap;
    DataNodesMap m_dataPathToDataNodeMap;
    AlgoInputsMap m_algoNameToAlgoInputsMap;
    AlgoOutputsMap m_algoNameToAlgoOutputsMap;
    unsigned int m_nodeCounter;
    mutable SmartIF<ISvcLocator> m_svcLocator;
    const std::string m_name;
    const std::chrono::system_clock::time_point m_initTime;
    std::vector<EventSlot>* m_eventSlots;
    boost::ExecPlan m_ExecPlan;
    std::map<std::string,boost::AlgoVertex> m_exec_plan_map;
  };


} // namespace concurrency


#endif