computational_graph.h

Go to the documentation of this file.

/* * NRP Core - Backend infrastructure to synchronize simulations
 *
 * Copyright 2020-2023 NRP Team
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * This project has received funding from the European Union’s Horizon 2020
 * Framework Programme for Research and Innovation under the Specific Grant
 * Agreement No. 945539 (Human Brain Project SGA3).
 */
 
#ifndef COMPUTATION_GRAPH_H
#define COMPUTATION_GRAPH_H
 
#include "nrp_general_library/utils/nrp_exceptions.h"
#include "nrp_general_library/utils/nrp_logger.h"
#include "nrp_event_loop/computational_graph/ngraph/ngraph.hpp"
#include "nrp_event_loop/computational_graph/computational_node.h"
 
class ComputationalGraph :
        private NGraph::tGraph<ComputationalNode *>
{
public:
 
    typedef std::vector<ComputationalGraph::vertex> comp_layer;
 
    enum GraphState {EMPTY, CONFIGURING, READY, COMPUTING};
 
    enum ExecMode {ALL_NODES, OUTPUT_DRIVEN};
 
    void insert_edge(const vertex &a, const vertex& b)
    {
        if(this->_state > GraphState::EMPTY)
            throw NRPException::logCreate("Inserting edges while the graph is being configured or is configured is not allowed");
 
        // Enforce edge rules
        if(a == b)
            throw NRPException::logCreate("Attempt to insert edge from node \"" + a->id() + "\" to itself. This is not allowed");
        if(a->type() == ComputationalNode::Output)
            throw NRPException::logCreate("Attempt to insert edge with Output node \"" + a->id() + "\" as source. This is not allowed");
        if(b->type() == ComputationalNode::Input)
            throw NRPException::logCreate("Attempt to insert edge with Input node \"" + b->id() + "\" as target. This is not allowed");
 
        NGraph::tGraph<ComputationalNode *>::insert_edge(a, b);
    }
 
    void clear()
    {
        if(this->_state == GraphState::CONFIGURING)
            throw NRPException::logCreate("Graph can't be cleared while is being configured");
        else if( this->_state == GraphState::COMPUTING)
            throw NRPException::logCreate("Graph can't be cleared while is computing");
 
        clearLayers();
        NGraph::tGraph<ComputationalNode *>::clear();
 
        this->_state = GraphState::EMPTY;
    }
 
    void configure()
    {
        if(this->_state > GraphState::EMPTY)
            throw NRPException::logCreate("Graph is already configured. Please reset the graph first by calling 'clear()'");
 
        this->_state = GraphState::CONFIGURING;
 
        try {
            // Configure nodes
            for (const auto &e: *this)
                e.first->configure();
 
            // Clear layers
            clearLayers();
 
            // Reset isVisited
            for (const auto &e: *this)
                e.first->setVisited(false);
 
            // Creates input and output layers
            setIOLayers();
 
            // Creates other layers
            comp_layer layer;
            setFirstLayer(layer);
            while (!layer.empty()) {
                _compLayers.push_back(layer);
                layer = comp_layer();
                setNextLayer(_compLayers.back(), layer);
            }
 
            if (checkForCycles())
                throw NRPException::logCreate("Cycle(s) found in the graph. Cycles are not supported");
 
 
            this->_state = GraphState::READY;
        }
        catch(const std::exception& e) {
            this->_state = GraphState::READY;
            clear();
            throw;
        }
    }
 
    void compute()
    {
        if (this->_state < GraphState::READY)
            throw NRPException::logCreate("Graph is not configured. Please (re-)build the graph by calling 'configure()'");
        else if (this->_state == GraphState::COMPUTING)
            throw NRPException::logCreate("'compute' can't be called while graph is already computing");
 
        this->_state = GraphState::COMPUTING;
 
        // Inform OutputNodes that it is a new execution cycle, currently they are the only type of nodes using this
        // information
        sendCycleStartSignal();
        
        try {
            // TODO: each of these loops could be possibly parallelized
 
            // Input nodes are always executed, regardless of the execution mode
            for (auto &node: _inputLayer)
                node->compute();
 
            // Functional nodes and output nodes are executed if they have been marked for execution or the CG is
            // being run in input controlled execution mode
            for (auto &layer: _compLayers)
                for (auto &node: layer)
                    if (this->_execMode == ExecMode::ALL_NODES || node->doCompute()) {
                        node->compute();
                        node->setDoCompute(false);
                    }
 
            for (auto &node: _outputLayer)
                if (this->_execMode == ExecMode::ALL_NODES || node->doCompute()) {
                    node->compute();
                    node->setDoCompute(false);
                }
 
            this->_state = GraphState::READY;
        }
        catch(const std::exception& e) {
            this->_state = GraphState::READY;
            throw;
        }
    }
 
    GraphState getState() const
    { return this->_state; }
 
    void setExecMode(ExecMode mode)
    {
        if( this->_state == GraphState::COMPUTING)
            throw NRPException::logCreate("Graph execution mode can't be changed while computing");
 
        _execMode = mode;
    }
 
    ExecMode getExecMode()
    { return _execMode; }
 
private:
 
    void sendCycleStartSignal()
    {
        // Inform OutputNodes that it is a new execution cycle, currently they are the only type of nodes using this
        // information
        for (auto &node: _outputLayer) {
            node->graphCycleStartCB();
            // In OUTPUT_DRIVEN mode, Output nodes that will execute this cycle propagates the "execution signal" back 
            // in the graph
            if(this->_execMode == ExecMode::OUTPUT_DRIVEN && node->doCompute())
                propagateExecSignalBack(node);
        }
    }
 
    void propagateExecSignalBack(const ComputationalGraph::vertex& v)
    {
        for(auto &node : this->in_neighbors(v))
            // If the node is already marked for execution it means it has already been processed
            if(!node->doCompute()) {
                node->setDoCompute(true);
                propagateExecSignalBack(node);
            }
    }
 
    void clearLayers()
    {
        _inputLayer.clear();
        _outputLayer.clear();
        _compLayers.clear();
    }
 
    void setIOLayers()
    {
        for(const auto &e : *this) {
            if(!e.first->isVisited() && e.first->type() == ComputationalNode::NodeType::Input) {
                e.first->setVisited(true);
                _inputLayer.push_back(e.first);
            }
 
            if(!e.first->isVisited() && e.first->type() == ComputationalNode::NodeType::Output) {
                e.first->setVisited(true);
                _outputLayer.push_back(e.first);
            }
        }
    }
 
    void setFirstLayer(comp_layer& layer)
    {
        for(const auto &e : *this) {
            auto v = e.first;
            if(v->isVisited() || v->type() != ComputationalNode::NodeType::Functional)
                continue;
 
            bool isFirst = true;
            for(const auto &i : this->in_neighbors(v))
                if(i->type() != ComputationalNode::NodeType::Input) {
                    isFirst = false;
                    break;
                }
 
            if(isFirst) {
                v->setVisited(true);
                layer.push_back(v);
            }
        }
    }
 
    void setNextLayer(const comp_layer& prev_layer, comp_layer& layer)
    {
        // build output set
        ComputationalGraph::vertex_set output_set;
        for(const auto &v : prev_layer)
            output_set.insert(this->out_neighbors(v).begin(), this->out_neighbors(v).end());
 
        // build layer
        for(const auto &v : output_set) {
            if(v->type() == ComputationalNode::NodeType::Output)
                continue;
 
            bool isNext = true;
            for(const auto &i : this->in_neighbors(v))
                if(i->type() != ComputationalNode::NodeType::Input && !i->isVisited()) {
                    isNext = false;
                    break;
                }
 
            if(isNext) {
                v->setVisited(true);
                layer.push_back(v);
            }
        }
    }
 
    bool checkForCycles()
    {
        for(const auto &e : *this)
            if(!e.first->isVisited())
                return true;
 
        return false;
    }
 
    comp_layer _inputLayer;
    comp_layer _outputLayer;
    std::vector<comp_layer> _compLayers;
 
    GraphState _state = GraphState::EMPTY;
 
    ExecMode _execMode = ExecMode::ALL_NODES;
 
 
};
 
 
#endif //COMPUTATION_GRAPH_H