node.h

// -*- mode: c++; tab-width: 4; indent-tabs-mode: t; eval: (progn (c-set-style "stroustrup") (c-set-offset 'innamespace 0)); -*-
// vi:set ts=4 sts=4 sw=4 noet :
// Copyright 2012, The TPIE development team
// 
// This file is part of TPIE.
// 
// TPIE is free software: you can redistribute it and/or modify it under
// the terms of the GNU Lesser General Public License as published by the
// Free Software Foundation, either version 3 of the License, or (at your
// option) any later version.
// 
// TPIE is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
// License for more details.
// 
// You should have received a copy of the GNU Lesser General Public License
// along with TPIE.  If not, see <http://www.gnu.org/licenses/>
 
#ifndef __TPIE_PIPELINING_NODE_H__
#define __TPIE_PIPELINING_NODE_H__
 
#include <tpie/tpie_export.h>
#include <tpie/pipelining/exception.h>
#include <tpie/pipelining/tokens.h>
#include <tpie/progress_indicator_base.h>
#include <tpie/progress_indicator_null.h>
#include <tpie/pipelining/priority_type.h>
#include <tpie/pipelining/predeclare.h>
#include <tpie/pipelining/node_name.h>
#include <tpie/pipelining/node_traits.h>
#include <tpie/flags.h>
#include <tpie/memory.h>
#include <limits>
#include <tpie/resources.h>
 
namespace tpie::pipelining {
namespace bits {
 
class TPIE_EXPORT proxy_progress_indicator : public tpie::progress_indicator_base {
    node & m_node;
 
public:
    proxy_progress_indicator(node & s);
 
    void refresh() override;
};
 
} // namespace bits
 
struct node_resource_parameters {
    memory_size_type minimum = 0;
    memory_size_type maximum = std::numeric_limits<memory_size_type>::max();
    double fraction = 0.0;
 
    memory_size_type available = 0;
};
 
struct node_parameters {
    node_resource_parameters resource_parameters[resource_type::TOTAL_RESOURCE_TYPES];
 
    std::string name;
    priority_type namePriority = PRIORITY_NO_NAME;
 
    std::string phaseName;
    priority_type phaseNamePriority = PRIORITY_NO_NAME;
 
    stream_size_type stepsTotal = 0;
};
 
class TPIE_EXPORT node {
public:
    typedef boost::optional<any_noncopyable &> maybeany_t;
 
    enum PLOT {
        PLOT_SIMPLIFIED_HIDE=1,
        PLOT_BUFFERED=2,
        PLOT_PARALLEL=4
    };
 
    enum STATE {
        STATE_FRESH,
        STATE_IN_PREPARE,
        STATE_AFTER_PREPARE,
        STATE_IN_PROPAGATE,
        STATE_AFTER_PROPAGATE,
        STATE_IN_BEGIN,
        STATE_AFTER_BEGIN,
        STATE_IN_GO,
        STATE_IN_END,
        STATE_AFTER_END
    };
 
    virtual ~node() {}
 
    memory_size_type get_minimum_resource_usage(resource_type type) const {
        return m_parameters.resource_parameters[type].minimum;
    }
 
    memory_size_type get_maximum_resource_usage(resource_type type) const {
        return m_parameters.resource_parameters[type].maximum;
    }
 
    double get_resource_fraction(resource_type type) const {
        return m_parameters.resource_parameters[type].fraction;
    }
 
    memory_size_type get_available_of_resource(resource_type type) const {
        return m_parameters.resource_parameters[type].available;
    }
 
    void set_minimum_resource_usage(resource_type type, memory_size_type usage);
 
    void set_maximum_resource_usage(resource_type type, memory_size_type usage);
 
    void set_resource_fraction(resource_type type, double f);
 
    virtual void resource_available_changed(resource_type, memory_size_type) {
    }
 
    void _internal_set_available_of_resource(resource_type type, memory_size_type available);
 
    memory_size_type get_minimum_memory() const {
        return get_minimum_resource_usage(MEMORY);
    }
 
    memory_size_type get_maximum_memory() const {
        return get_maximum_resource_usage(MEMORY);
    }
 
    double get_memory_fraction() const {
        return get_resource_fraction(MEMORY);
    }
 
    memory_size_type get_available_memory() const {
        return get_available_of_resource(MEMORY);
    }
 
    void set_minimum_memory(memory_size_type minimumMemory) {
        set_minimum_resource_usage(MEMORY, minimumMemory);
    }
 
    void set_maximum_memory(memory_size_type maximumMemory) {
        set_maximum_resource_usage(MEMORY, maximumMemory);
    }
 
    void set_memory_fraction(double f) {
        set_resource_fraction(MEMORY, f);
    }
 
    virtual void set_available_memory(memory_size_type availableMemory) {
        unused(availableMemory);
    }
 
    memory_size_type get_used_memory() const {
        memory_size_type ans=0;
        for (const auto & p: m_buckets)
            if (p) ans += p->count;
        return ans;
    }
 
    bits::node_map::ptr get_node_map() const {
        return token.get_map();
    }
 
    node_token::id_t get_id() const {
        return token.id();
    }
 
    virtual void prepare() {
    }
 
    virtual void propagate() {
    }
 
    virtual void begin() {
    }
 
    virtual bool is_go_free() const {return false;}
    
    virtual void go() {
        log_warning() << "node subclass " << typeid(*this).name()
            << " is not an initiator node" << std::endl;
        throw not_initiator_node();
    }
 
    virtual void end() {
    }
 
    virtual bool can_evacuate() {
        return false;
    }
 
    virtual void evacuate() {
    }
 
    inline priority_type get_name_priority() {
        return m_parameters.namePriority;
    }
 
    const std::string & get_name();
 
    void set_name(const std::string & name, priority_type priority = PRIORITY_USER);
 
    priority_type get_phase_name_priority() {
        return m_parameters.phaseNamePriority;
    }
 
    const std::string & get_phase_name();
 
    void set_phase_name(const std::string & name, priority_type priority = PRIORITY_USER);
 
    
    void set_breadcrumb(const std::string & breadcrumb) {
        m_parameters.name = m_parameters.name.empty() ? breadcrumb : (breadcrumb + " | " + m_parameters.name);
    }
 
    stream_size_type get_steps() {
        return m_parameters.stepsTotal;
    }
 
    stream_size_type get_steps_left() {
        return m_stepsLeft;
    }
 
    void set_progress_indicator(progress_indicator_base * pi) {
        m_pi = pi;
    }
 
    progress_indicator_base * get_progress_indicator() {
        return m_pi;
    }
 
    STATE get_state() const {
        return m_state;
    }
 
    void set_state(STATE s) {
        m_state = s;
    }
 
    resource_type get_resource_being_assigned() const {
        return m_resourceBeingAssigned;
    }
 
    void set_resource_being_assigned(resource_type type) {
        m_resourceBeingAssigned = type;
    }
 
    flags<PLOT> get_plot_options() const {
        return m_plotOptions;
    }
 
    void set_plot_options(flags<PLOT> options) {
        m_plotOptions = options;
    }
protected:
#ifdef _WIN32
    // Disable warning C4355: 'this' : used in base member initializer list
    // node_token does not access members of the `node *`,
    // it merely uses it as a value in the node map.
    // Only after this node object is completely constructed are node members accessed.
#pragma warning( push )
#pragma warning( disable : 4355 )
#endif // _WIN32
    node();
 
    node(const node & other) = delete;
    node & operator=(const node & other) = delete;
 
    node(node && other);
    node & operator=(node && other);
 
    node(const node_token & token);
#ifdef _WIN32
#pragma warning( pop )
#endif // _WIN32
public:
    void add_push_destination(const node_token & dest);
 
    void add_push_destination(const node & dest);
 
    void add_pull_source(const node_token & dest);
 
    void add_pull_source(const node & dest);
 
    void add_dependency(const node_token & dest);
 
    void add_dependency(const node & dest);
 
    void add_memory_share_dependency(const node_token & dest);
 
    void add_memory_share_dependency(const node & dest);
 
    // Implementation note: If the type of the `value` parameter is changed
    // from `T` to `const T &`, this will yield linker errors if an application
    // attempts to pass a const reference to a static data member inside a
    // templated class.
    // See http://stackoverflow.com/a/5392050
    template <typename T>
    void forward(std::string key, T value, memory_size_type k = std::numeric_limits<memory_size_type>::max()) {
        forward_any(key, any_noncopyable(std::move(value)), k);
    }
 
    void forward_any(std::string key, any_noncopyable value, memory_size_type k = std::numeric_limits<memory_size_type>::max());
 
private:
    void add_forwarded_data(std::string key, node_token::id_t from_node);
 
    maybeany_t get_forwarded_data_maybe(std::string key);
 
public:
    bool can_fetch(std::string key) {
        return bool(fetch_maybe(key));
    }
 
    maybeany_t fetch_maybe(std::string key);
 
    any_noncopyable & fetch_any(std::string key);
 
    template <typename T>
    inline T & fetch(std::string key) {
        any_noncopyable &item = fetch_any(key);
        try {
            return any_cast<T>(item);
        } catch (const bad_any_noncopyable_cast & e) {
            std::stringstream ss;
            ss << "Trying to fetch key '" << key << "' of type "
               << typeid(T).name() << " but forwarded data was of type "
               << item.type().name() << ". Message was: " << e.what();
            throw invalid_argument_exception(ss.str());
        }
    }
 
    void no_forward_through();
 
    const node_token & get_token() const {
        return token;
    }
 
public:
    void set_steps(stream_size_type steps);
 
 
private:
    void step_overflow();
public:
    void step(stream_size_type steps = 1) {
        assert(get_state() == STATE_IN_END ||
               get_state() == STATE_IN_BEGIN ||
               get_state() == STATE_AFTER_BEGIN ||
               get_state() == STATE_IN_END ||
               get_state() == STATE_IN_GO);
        if (m_stepsLeft < steps)
            step_overflow();
        else
            m_stepsLeft -= steps;
        m_pi->step(steps);
    }
 
    progress_indicator_base * proxy_progress_indicator();
 
#ifdef DOXYGEN
    inline bool can_pull() const;
 
    inline item_type pull();
 
    inline void push(const item_type & item);
#endif
 
    void register_datastructure_usage(const std::string & name, double priority=1);
 
    void set_datastructure_memory_limits(const std::string & name, memory_size_type min, memory_size_type max=std::numeric_limits<memory_size_type>::max());
 
    memory_size_type get_datastructure_memory(const std::string & name);
 
    template<typename T>
    void set_datastructure(const std::string & name, T datastructure) {
        bits::node_map::datastructuremap_t & structures = get_node_map()->find_authority()->get_datastructures();
        bits::node_map::datastructuremap_t::iterator i = structures.find(name);
 
        if(i == structures.end())
            throw tpie::exception("attempted to set non-registered datastructure");
 
        i->second.second = move_if_movable<T>(datastructure);
    }
 
    template<typename T>
    T & get_datastructure(const std::string & name) {
        bits::node_map::datastructuremap_t & structures = get_node_map()->find_authority()->get_datastructures();
        bits::node_map::datastructuremap_t::iterator i = structures.find(name);
 
        if(i == structures.end())
            throw tpie::exception("attempted to get non-registered datastructure");
 
        return any_cast<T>(i->second.second);
    }
 
    void unset_datastructure(const std::string & name) {
        bits::node_map::datastructuremap_t & structures = get_node_map()->find_authority()->get_datastructures();
        bits::node_map::datastructuremap_t::iterator i = structures.find(name);
 
        if(i == structures.end()) return;
        i->second.second.reset();
    }
    
private:
    struct datastructure_info_t {
        datastructure_info_t() : min(0), max(std::numeric_limits<memory_size_type>::max()) {}
        memory_size_type min;
        memory_size_type max;
        double priority;
    };
 
    typedef std::map<std::string, datastructure_info_t> datastructuremap_t;
 
    const datastructuremap_t & get_datastructures() const {
        return m_datastructures;
    }
 
public:
    size_t buckets() const {return m_buckets.size();}
 
    std::unique_ptr<memory_bucket> & bucket(size_t i) {
        if (m_buckets.size() <= i) m_buckets.resize(i+1);
        if (!m_buckets[i]) m_buckets[i].reset(new memory_bucket());
        return m_buckets[i];
    }
        
    tpie::memory_bucket_ref allocator(size_t i=0) {
        return tpie::memory_bucket_ref(bucket(i).get());
    }
    
    friend class bits::memory_runtime;
 
    friend class bits::datastructure_runtime;
 
    friend class factory_base;
 
    friend class bits::pipeline_base;
private:
    node_token token;
 
    node_parameters m_parameters;
    std::vector<std::unique_ptr<memory_bucket> > m_buckets;
    
    std::map<std::string, any_noncopyable> m_forwardedFromHere;
    std::map<std::string, node_token::id_t> m_forwardedToHere;
 
    datastructuremap_t m_datastructures;
    stream_size_type m_stepsLeft;
    progress_indicator_base * m_pi;
    STATE m_state;
    resource_type m_resourceBeingAssigned = NO_RESOURCE;
    std::unique_ptr<progress_indicator_base> m_piProxy;
    flags<PLOT> m_plotOptions;
 
    friend class bits::proxy_progress_indicator;
};
 
 
TPIE_DECLARE_OPERATORS_FOR_FLAGS(node::PLOT);
 
} // namespace tpie::pipelining
 
#endif // __TPIE_PIPELINING_NODE_H__