ReduceNode.hpp

#pragma once
#include <Dataflow/Core/Node.hpp>
 
#include <functional>
 
namespace Ra {
namespace Dataflow {
namespace Core {
namespace Functionals {
 
// TODO, allow to specify the type of the reduced information. This will allow, e.g, given a
//  collection of X, to compute a tuple containing mean and standard deviation of the collection
//  as reduction (using online Welford algo).
template <typename coll_t, typename v_t = typename coll_t::value_type>
class ReduceNode : public Node
{
  public:
    using ReduceOperator = std::function<v_t( const v_t&, const v_t& )>;
 
    explicit ReduceNode( const std::string& instanceName );
 
    ReduceNode( const std::string& instanceName, ReduceOperator op, v_t initialValue = v_t {} );
 
    void init() override;
    bool execute() override;
 
    void set_operator( ReduceOperator op, v_t initialValue = v_t {} );
 
  protected:
    ReduceNode( const std::string& instanceName,
                const std::string& typeName,
                ReduceOperator op,
                v_t initialValue );
 
    void toJsonInternal( nlohmann::json& data ) const override { Node::toJsonInternal( data ); }
    bool fromJsonInternal( const nlohmann::json& data ) override {
        return Node::fromJsonInternal( data );
    }
    bool fromJsonInternal( const nlohmann::json& data ) override  {…}
 
  private:
    RA_NODE_PORT_IN( v_t, init );
    RA_NODE_PORT_IN( coll_t, data );
    RA_NODE_PORT_IN( ReduceOperator, op );
    RA_NODE_PORT_OUT_WITH_DATA( v_t, result );
 
  public:
    static const std::string& node_typename();
};
class ReduceNode : public Node {…};
 
// -----------------------------------------------------------------
// ---------------------- inline methods ---------------------------
 
template <typename coll_t, typename v_t>
ReduceNode<coll_t, v_t>::ReduceNode( const std::string& instanceName ) :
    ReduceNode(
        instanceName,
        node_typename(),
        []( const v_t& a, const v_t& ) -> v_t { return a; },
        v_t {} ) {}
ReduceNode<coll_t, v_t>::ReduceNode( const std::string& instanceName ) : {…}
 
template <typename coll_t, typename v_t>
ReduceNode<coll_t, v_t>::ReduceNode( const std::string& instanceName,
                                     ReduceOperator op,
                                     v_t initialValue ) :
    ReduceNode( instanceName, node_typename(), op, initialValue ) {}
ReduceNode<coll_t, v_t>::ReduceNode( const std::string& instanceName, {…}
 
template <typename coll_t, typename v_t>
void ReduceNode<coll_t, v_t>::set_operator( ReduceOperator op, v_t initialValue ) {
    m_port_in_op->set_default_value( op );
    m_port_in_init->set_default_value( initialValue );
}
void ReduceNode<coll_t, v_t>::set_operator( ReduceOperator op, v_t initialValue ) {…}
 
template <typename coll_t, typename v_t>
void ReduceNode<coll_t, v_t>::init() {
    Node::init();
    m_result = m_port_in_init->data();
}
void ReduceNode<coll_t, v_t>::init() {…}
 
template <typename coll_t, typename v_t>
bool ReduceNode<coll_t, v_t>::execute() {
    const auto& f  = m_port_in_op->data();
    const auto& in = m_port_in_data->data();
 
    m_result = std::accumulate( in.begin(), in.end(), m_port_in_init->data(), f );
 
    return true;
}
bool ReduceNode<coll_t, v_t>::execute() {…}
 
template <typename coll_t, typename v_t>
const std::string& ReduceNode<coll_t, v_t>::node_typename() {
    static std::string demangledName =
        std::string { "Reduce<" } + Ra::Core::Utils::simplifiedDemangledType<coll_t>() + ">";
    return demangledName;
}
 
template <typename coll_t, typename v_t>
ReduceNode<coll_t, v_t>::ReduceNode( const std::string& instanceName,
                                     const std::string& typeName,
                                     ReduceOperator op,
                                     v_t initialValue ) :
    Node( instanceName, typeName ) {
    m_port_in_op->set_default_value( op );
    m_port_in_init->set_default_value( initialValue );
}
 
} // namespace Functionals
} // namespace Core
} // namespace Dataflow
} // namespace Ra