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/pipelining/exception.h>
#include <tpie/pipelining/tokens.h>
#include <tpie/progress_indicator_base.h>
#include <tpie/progress_indicator_null.h>
#include <boost/any.hpp>
#include <tpie/pipelining/priority_type.h>
#include <tpie/pipelining/predeclare.h>

namespace tpie {

namespace pipelining {

namespace bits {

class proxy_progress_indicator : public tpie::progress_indicator_base {
    node & m_node;

public:
    proxy_progress_indicator(node & s)
        : progress_indicator_base(1)
        , m_node(s)
    {
    }

    inline void refresh();
};

} // namespace bits

class node {
public:
    enum STATE {
        STATE_FRESH,
        STATE_IN_PREPARE,
        STATE_AFTER_PREPARE,
        STATE_IN_PROPAGATE,
        STATE_AFTER_PROPAGATE,
        STATE_IN_BEGIN,
        STATE_AFTER_BEGIN,
        STATE_IN_END,
        STATE_AFTER_END
    };

    virtual ~node() {}

    inline memory_size_type get_minimum_memory() const {
        return m_minimumMemory;
    }

    inline memory_size_type get_maximum_memory() const {
        return m_maximumMemory;
    }

    inline memory_size_type get_available_memory() const {
        return m_availableMemory;
    }

    inline void set_memory_fraction(double f) {
        m_memoryFraction = f;
    }

    inline double get_memory_fraction() const {
        return m_memoryFraction;
    }

    inline bits::node_map::ptr get_node_map() const {
        return token.get_map();
    }

    inline node_token::id_t get_id() const {
        return token.id();
    }

    virtual void prepare() {
    }

    virtual void propagate() {
    }

    virtual void begin() {
    }

    virtual void go() {
        progress_indicator_null pi;
        go(pi);
        // if go didn't throw, it was overridden - but it shouldn't be
        log_warning() << "node subclass " << typeid(*this).name() << " uses old go() interface" << std::endl;
    }

    virtual void go(progress_indicator_base &) {
        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_namePriority;
    }

    inline const std::string & get_name() {
        return m_name;
    }

    inline void set_name(const std::string & name, priority_type priority = PRIORITY_USER) {
        m_name = name;
        m_namePriority = priority;
    }

    inline void set_breadcrumb(const std::string & breadcrumb) {
        m_name = m_name.empty() ? breadcrumb : (breadcrumb + " | " + m_name);
    }

    inline void add_successor(node * succ) {
        m_successors.push_back(succ);
    }

    inline stream_size_type get_steps() {
        return m_stepsTotal;
    }

    inline 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;
    }

protected:
    inline node()
        : token(this)
        , m_minimumMemory(0)
        , m_maximumMemory(std::numeric_limits<memory_size_type>::max())
        , m_availableMemory(0)
        , m_memoryFraction(0.0)
        , m_namePriority(PRIORITY_NO_NAME)
        , m_stepsTotal(0)
        , m_stepsLeft(0)
        , m_pi(0)
        , m_state(STATE_FRESH)
    {
    }

    inline node(const node & other)
        : token(other.token, this)
        , m_minimumMemory(other.m_minimumMemory)
        , m_maximumMemory(other.m_maximumMemory)
        , m_availableMemory(other.m_availableMemory)
        , m_memoryFraction(other.m_memoryFraction)
        , m_name(other.m_name)
        , m_namePriority(other.m_namePriority)
        , m_stepsTotal(other.m_stepsTotal)
        , m_stepsLeft(other.m_stepsLeft)
        , m_pi(other.m_pi)
        , m_state(other.m_state)
    {
        if (m_state != STATE_FRESH) 
            throw call_order_exception(
                "Tried to copy pipeline node after prepare had been called");
    }
    
    inline node(const node_token & token)
        : token(token, this, true)
        , m_minimumMemory(0)
        , m_maximumMemory(std::numeric_limits<memory_size_type>::max())
        , m_availableMemory(0)
        , m_memoryFraction(0.0)
        , m_namePriority(PRIORITY_NO_NAME)
        , m_stepsTotal(0)
        , m_stepsLeft(0)
        , m_pi(0)
        , m_state(STATE_FRESH)
    {
    }

    inline void add_push_destination(const node_token & dest) {
        bits::node_map::ptr m = token.map_union(dest);
        m->add_relation(token.id(), dest.id(), bits::pushes);
    }

    inline void add_push_destination(const node & dest) {
        if (get_state() != STATE_FRESH) {
            throw call_order_exception("add_push_destination called too late");
        }
        add_push_destination(dest.token);
    }

    void add_pull_source(const node_token & dest) {
        if (get_state() != STATE_FRESH) {
            throw call_order_exception("add_pull_source called too late");
        }
        bits::node_map::ptr m = token.map_union(dest);
        m->add_relation(token.id(), dest.id(), bits::pulls);
    }

    void add_pull_source(const node & dest) {
        add_pull_source(dest.token);
    }

    void add_pull_destination(const node_token & dest) {
        add_pull_source(dest);
    }

    void add_pull_destination(const node & dest) {
        add_pull_source(dest);
    }

    inline void add_dependency(const node_token & dest) {
        bits::node_map::ptr m = token.map_union(dest);
        m->add_relation(token.id(), dest.id(), bits::depends);
    }

    inline void add_dependency(const node & dest) {
        add_dependency(dest.token);
    }

    inline void set_minimum_memory(memory_size_type minimumMemory) {
        if (get_state() != STATE_FRESH && get_state() != STATE_IN_PREPARE) {
            throw call_order_exception("set_minimum_memory");
        }
        m_minimumMemory = minimumMemory;
    }

    inline void set_maximum_memory(memory_size_type maximumMemory) {
        if (get_state() != STATE_FRESH && get_state() != STATE_IN_PREPARE) {
            throw call_order_exception("set_maximum_memory");
        }
        m_maximumMemory = maximumMemory;
    }

    virtual void set_available_memory(memory_size_type availableMemory) {
        m_availableMemory = availableMemory;
    }

    // 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, bool explicitForward = true) {
        forward_any(key, boost::any(value), explicitForward);
    }

    void forward_any(std::string key, boost::any value, bool explicitForward = true) {
        switch (get_state()) {
            case STATE_IN_PREPARE:
                log_debug() << "forward in prepare" << std::endl;
                break;
            case STATE_IN_PROPAGATE:
                log_debug() << "forward in propagate" << std::endl;
                break;
            case STATE_IN_BEGIN:
                log_debug() << "forward in begin" << std::endl;
                break;
            case STATE_AFTER_BEGIN:
                log_debug() << "forward after begin" << std::endl;
                break;
            case STATE_AFTER_END:
                throw call_order_exception("forward");
            default:
                log_debug() << "forward in unknown state " << get_state() << std::endl;
                break;
        }

        for (size_t i = 0; i < m_successors.size(); ++i) {
            m_successors[i]->add_forwarded_data(key, value, explicitForward);
        }
    }

public:
    void add_forwarded_data(std::string key, boost::any value, bool explicitForward = true) {
        if (m_values.count(key) &&
            !explicitForward && m_values[key].second) return;
        m_values[key].first = value;
        m_values[key].second = explicitForward;
        forward_any(key, value, false);
    }

    inline bool can_fetch(std::string key) {
        return m_values.count(key) != 0;
    }

    inline boost::any fetch_any(std::string key) {
        if (m_values.count(key) != 0) {
            return m_values[key].first;
        } else {
            std::stringstream ss;
            ss << "Tried to fetch nonexistent key '" << key << '\'';
            throw invalid_argument_exception(ss.str());
        }
    }

    template <typename T>
    inline T fetch(std::string key) {
        if (m_values.count(key) == 0) {
            std::stringstream ss;
            ss << "Tried to fetch nonexistent key '" << key
               << "' of type " << typeid(T).name();
            throw invalid_argument_exception(ss.str());
        }
        try {
            return boost::any_cast<T>(m_values[key].first);
        } catch (boost::bad_any_cast m) {
            std::stringstream ss;
            ss << "Trying to fetch key '" << key << "' of type "
               << typeid(T).name() << " but forwarded data was of type "
               << m_values[key].first.type().name() << ". Message was: " << m.what();
            throw invalid_argument_exception(ss.str());
        }
    }

protected:
    const node_token & get_token() {
        return token;
    }

    void set_steps(stream_size_type steps) {
        switch (get_state()) {
            case STATE_FRESH:
            case STATE_IN_PREPARE:
            case STATE_IN_PROPAGATE:
                break;
            case STATE_IN_BEGIN:
                log_error() << "set_steps in begin(); use set_steps in propagate() instead." << std::endl;
                throw call_order_exception("set_steps");
            default:
                log_error() << "set_steps in unknown state " << get_state() << std::endl;
                throw call_order_exception("set_steps");
        }
        m_stepsTotal = m_stepsLeft = steps;
    }

    void step(stream_size_type steps = 1) {
        assert(get_state() == STATE_IN_END || get_state() == STATE_AFTER_BEGIN || get_state() == STATE_IN_END);
        if (m_stepsLeft < steps) {
            log_warning() << typeid(*this).name() << " ==== Too many steps!" << std::endl;
            m_stepsLeft = 0;
        } else {
            m_stepsLeft -= steps;
        }
        m_pi->step(steps);
    }

    progress_indicator_base * proxy_progress_indicator() {
        if (m_piProxy.get() != 0) return m_piProxy.get();
        progress_indicator_base * pi = new bits::proxy_progress_indicator(*this);
        m_piProxy.reset(pi);
        return pi;
    }

#ifdef DOXYGEN
    inline bool can_pull() const;

    inline item_type pull();

    inline void push(const item_type & item);
#endif

    friend class bits::phase;

private:
    node_token token;

    memory_size_type m_minimumMemory;
    memory_size_type m_maximumMemory;
    memory_size_type m_availableMemory;
    double m_memoryFraction;

    std::string m_name;
    priority_type m_namePriority;

    std::vector<node *> m_successors;
    typedef std::map<std::string, std::pair<boost::any, bool> > valuemap;
    valuemap m_values;

    stream_size_type m_stepsTotal;
    stream_size_type m_stepsLeft;
    progress_indicator_base * m_pi;
    STATE m_state;
    std::auto_ptr<progress_indicator_base> m_piProxy;

    friend class bits::proxy_progress_indicator;
};

namespace bits {

void proxy_progress_indicator::refresh() {
    double proxyMax = static_cast<double>(get_range());
    double proxyCur = static_cast<double>(get_current());
    double parentMax = static_cast<double>(m_node.m_stepsTotal);
    double parentCur = static_cast<double>(m_node.m_stepsTotal-m_node.m_stepsLeft);
    double missing = parentMax*proxyCur/proxyMax - parentCur;
    if (missing < 1.0) return;
    stream_size_type times = static_cast<stream_size_type>(1.0+missing);
    times = std::min(m_node.m_stepsLeft, times);
    m_node.step(times);
}

} // namespace bits

} // namespace pipelining

} // namespace tpie

#endif // __TPIE_PIPELINING_NODE_H__