input_node.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 INPUT_NODE_H
#define INPUT_NODE_H
 
#include <memory>
#include <map>
#include <iostream>
 
#include "nrp_general_library/utils/nrp_logger.h"
 
#include "nrp_event_loop/computational_graph/computational_node_policies.h"
#include "nrp_event_loop/computational_graph/computational_node.h"
#include "nrp_event_loop/computational_graph/output_port.h"
 
template<class DATA>
struct DataPortHandle {
 
    DataPortHandle() = delete;
 
    DataPortHandle(const std::string& id, ComputationalNode* parent, size_t queue_size) :
            singlePort(new OutputPort<DATA>(id, parent)),
            listPort(new OutputPort<std::vector<const DATA*>>(id, parent))
    { _data.reserve(queue_size); }
 
    void publishLast()
    {
        if(_data.size())
            singlePort->publish(_data.back());
    }
 
    void publishAll()
    {
        if(_data.size())
            listPort->publish(&_data);
    }
 
    void publishNullandClear()
    {
        singlePort->publish(nullptr);
        listPort->publish(nullptr);
        _data.clear();
    }
 
    bool addMsg(const DATA* msg)
    {
        if(_data.size() < _data.capacity()) {
            _data.push_back(msg);
            return true;
        }
        else {
            std::string s = "Port handle '" + singlePort->id() + "' reached max size. Added msg will be dropped.";
            NRPLogger::info(s);
            return false;
        }
    }
 
    void clear()
    { _data.clear(); }
 
    size_t size()
    { return _data.size(); }
 
    std::shared_ptr<OutputPort<DATA>> singlePort;
    std::shared_ptr<OutputPort<std::vector<const DATA*>>> listPort;
 
private:
 
    std::vector<const DATA*> _data;
};
 
template <class DATA>
class InputNode : public ComputationalNode {
public:
 
    InputNode(const std::string &id, InputNodePolicies::MsgPublishPolicy msgPublishPolicy = InputNodePolicies::MsgPublishPolicy::LAST, InputNodePolicies::MsgCachePolicy msgCachePolicy = InputNodePolicies::MsgCachePolicy::KEEP_CACHE, size_t queue_size = 10) :
            ComputationalNode(id, ComputationalNode::Input),
            _msgPublishPolicy(msgPublishPolicy),
            _msgCachePolicy(msgCachePolicy),
            _queueSize(queue_size)
    { }
 
    void registerOutput(const std::string& id)
    {
        if(!_portMap.count(id))
            _portMap.emplace(id, DataPortHandle<DATA>(id, this, _queueSize));
        else {
            std::string s = "A port with id '" + id + "' is already registered with node '" + this->id() +
                    "', ignoring this request.";
            NRPLogger::info(s);
        }
    }
 
    OutputPort<DATA>* getSinglePort(const std::string& id)
    { return _portMap.count(id) ? _portMap.at(id).singlePort.get() : nullptr; }
 
    OutputPort<std::vector<const DATA*>>* getListPort(const std::string& id)
    { return _portMap.count(id) ? _portMap.at(id).listPort.get() : nullptr; }
 
    InputNodePolicies::MsgPublishPolicy msgPublishPolicy()
    { return _msgPublishPolicy; }
 
    InputNodePolicies::MsgCachePolicy msgCachePolicy()
    { return _msgCachePolicy; }
 
    void setMsgPublishPolicy(InputNodePolicies::MsgPublishPolicy msgPublishPolicy)
    { _msgPublishPolicy = msgPublishPolicy; }
 
    void setMsgCachePolicy(InputNodePolicies::MsgCachePolicy msgCachePolicy)
    { _msgCachePolicy = msgCachePolicy; }
 
protected:
 
    void compute() override final
    {
        for(auto& [id, port]: _portMap) {
            auto hasNewMsgs = this->updatePortData(id);
 
            if(hasNewMsgs) {
                if(_msgPublishPolicy == InputNodePolicies::MsgPublishPolicy::LAST)
                    port.publishLast();
                else
                    port.publishAll();
            }
            else if(_msgCachePolicy == InputNodePolicies::MsgCachePolicy::CLEAR_CACHE)
                port.publishNullandClear();
        }
    }
 
    virtual bool updatePortData(const std::string& id) = 0;
 
    InputNodePolicies::MsgPublishPolicy _msgPublishPolicy;
    InputNodePolicies::MsgCachePolicy _msgCachePolicy;
    std::map<std::string, DataPortHandle<DATA>> _portMap;
    size_t _queueSize;
 
    friend class ComputationalNodes_INPUT_NODE_UPDATE_POLICY_WITH_KEEP_CACHE_Test;
    friend class ComputationalNodes_INPUT_NODE_UPDATE_POLICY_WITH_CLEAR_CACHE_Test;
};
 
 
#endif //INPUT_NODE_H