node.h

Go to the documentation of this file.

#if !defined(FLOW_NODE_H)
     #define FLOW_NODE_H

#include "named.h"
#include "packet.h"
#include "pipe.h"

#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <utility>
#include <vector>
#include <utility>


namespace flow
{

// Forward declarations.
template<typename T>
class outpin;

template<typename T>
class consumer;

template<typename T>
class producer;

class graph;

namespace state
{

enum type
{
    started,    
    paused,     
    stopped     
};

}

template<typename T>
class pin : public named
{
protected:
    std::shared_ptr<std::pair<pipe<T>, std::unique_ptr<std::mutex>>> d_pipe_sp; 

    friend class outpin<T>;

public:
    pin(const std::string& name_r) : named(name_r) {}

    pin(const pin<T>& o) : named(o.name()), d_pipe_sp(o.d_pipe_sp) {}

    virtual ~pin() {}

    virtual void disconnect()
    {
        d_pipe_sp.reset();
    }
};

template<typename T>
class inpin : public pin<T>
{
    std::condition_variable *d_transition_cv_p;
    std::mutex *d_transition_m_p;

    using pin<T>::d_pipe_sp;

    virtual void disconnect()
    {
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            d_pipe_sp->first.rename(d_pipe_sp->first.input()->name() + "_to_" + "nothing");
        }

        pin<T>::disconnect();
    }

    friend class consumer<T>;

public:
    inpin(const std::string& name_r, std::condition_variable* transition_cv_p, std::mutex *transition_m_p)
        : pin<T>(name_r), d_transition_cv_p(transition_cv_p), d_transition_m_p(transition_m_p)
    {}

    virtual ~inpin() {}

    virtual std::string rename(const std::string& name_r)
    {
        if(d_pipe_sp)
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            if(d_pipe_sp->first.input())
            {
                d_pipe_sp->first.rename(d_pipe_sp->first.input()->name() + "_to_" + name_r);
            }
        }

        return named::rename(name_r);
    }

    virtual bool peek() const
    {
        if(d_pipe_sp)
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            return d_pipe_sp->first.length() != 0;
        }

        return false;
    }

    virtual std::unique_ptr<packet<T>> pop()
    {
        if(d_pipe_sp)
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            return d_pipe_sp->first.pop();
        }

        return std::unique_ptr<packet<T>>();
    }

    virtual void incoming()
    {
        std::unique_lock<std::mutex> ul(*d_transition_m_p);
        d_transition_cv_p->notify_one();
    }
};

template<typename T>
class outpin : public pin<T>
{
    using pin<T>::d_pipe_sp;
    
    virtual void disconnect()
    {
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            d_pipe_sp->first.rename(std::string("nothing") + "_to_" + d_pipe_sp->first.output()->name());
        }

        pin<T>::disconnect();
    }

    virtual void connect(inpin<T>& inpin_r, const size_t max_length = 0, const size_t max_weight = 0)
    {
        // Disconnect this outpin from it's pipe, if it has one.
        if(d_pipe_sp)
        {
            disconnect();
        }

        if(inpin_r.pin<T>::d_pipe_sp)
        {
            // The inpin already has a pipe, connect this outpin to it.
            std::lock_guard<std::mutex> lg(*inpin_r.pin<T>::d_pipe_sp->second);
            
            pipe<T> &inpin_pipe = inpin_r.pin<T>::d_pipe_sp->first;

            //... but first, disconnects it from it's other output pin.
            if(inpin_pipe.input())
            {
                inpin_pipe.input()->disconnect();
            }

            inpin_pipe.rename(pin<T>::name() + "_to_" + inpin_pipe.name());

            // Overwrite the pipe's parameters with new ones.
            inpin_pipe.cap_length(max_length);
            inpin_pipe.cap_length(max_weight);
        }
        else
        {
            // The inpin has no pipe, make a new one.
            pipe<T> p(pin<T>::name() + "_to_" + inpin_r.pin<T>::name(), this, &inpin_r, max_length, max_weight);
            d_pipe_sp = inpin_r.pin<T>::d_pipe_sp = std::make_shared<std::pair<pipe<T>, std::unique_ptr<std::mutex>>>(std::make_pair(std::move(p), std::unique_ptr<std::mutex>(new std::mutex()))); 
        }
    }

    friend class producer<T>;

public:
    outpin(const std::string& name_r) : pin<T>(name_r) {}

    virtual ~outpin() {}

    virtual std::string rename(const std::string& name_r)
    {
        if(d_pipe_sp)
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            if(d_pipe_sp->first.output())
            {
                d_pipe_sp->first.rename(name_r + "_to_" + d_pipe_sp->first.output()->name());
            }
        }

        return named::rename(name_r);
    }

    virtual bool push(std::unique_ptr<packet<T>> packet_p)
    {
        if(!d_pipe_sp) return false;

        inpin<T>* inpin_p = 0;
        {
            std::lock_guard<std::mutex> lg(*d_pipe_sp->second);
            if(d_pipe_sp->first.push(std::move(packet_p)))
            {
                inpin_p = d_pipe_sp->first.output();
            }
        }

        if(inpin_p)
        {
            inpin_p->incoming();
        }

        return inpin_p != 0;
    }
};

class node : public named
{
    state::type d_state_a; 

    virtual void transition(state::type s)
    {
        std::unique_lock<std::mutex> ul(d_transition_m);

        d_state_a = s;

        // Notify the concrete class.
        switch(s)
        {
        case state::started: started(); break;
        case state::paused: paused(); break;
        case state::stopped: stopped(); break;
        default: break;
        }

        // Notify the execution loop.
        d_transition_cv.notify_one();
    }

    friend class graph;

protected:
    std::condition_variable d_transition_cv;    
    std::mutex d_transition_m;                  

    virtual void sever() = 0;

public:
    node(const std::string& name_r) : named(name_r), d_state_a(state::paused)
    {}

    node(node&& node_rr) : named(std::move(node_rr)), d_state_a(node_rr.d_state_a)
    {}
    
    virtual ~node() {}

    virtual state::type state() const
    {
        return d_state_a;
    }
    
    virtual void started() {}

    virtual void paused() {}

    virtual void stopped() {}

    virtual void operator()() = 0;
};

namespace detail
{

// These base classes help flow::graph distinguish between types of nodes without having to know their template types.

class producer
{
public:
    virtual ~producer() {}
};

class transformer
{
public:
    virtual ~transformer() {}
};

class consumer
{
public:
    virtual ~consumer() {}
};

}

template<typename T>
class producer : public virtual node, public detail::producer
{
    typedef std::vector<outpin<T>> outputs_t;
    outputs_t d_outputs;

protected:
    virtual void connect(size_t p_pin, consumer<T>* consumer_p, size_t c_pin, const size_t max_length = 0, const size_t max_weight = 0)
    {
        output(p_pin).connect(consumer_p->input(c_pin), max_length, max_weight);
    }

    virtual void disconnect(size_t pin)
    {
        output(pin).disconnect();
    }

    virtual void sever()
    {
        for(auto& outpin : d_outputs)
        {
            outpin.disconnect();
        }
    }

    virtual void operator()()
    {
        state::type s(state());

        while(s != state::stopped)
        {
            if(s == state::paused)
            {
                std::unique_lock<std::mutex> ul(d_transition_m);
                d_transition_cv.wait(ul, [&s, this](){ return (s = this->state()) != state::paused; });
            }
            else
            {
                s = state();
            }

            if(s == state::started)
            {
                produce();
            }
        }
    }

    friend class graph;

public:
    producer(const std::string& name_r, const size_t outs) : node(name_r)
    {
        for(size_t i = 0; i != outs; ++i)
        {
            d_outputs.push_back(outpin<T>(name_r + "_out" + static_cast<char>('0' + i)));
        }
    }

    virtual ~producer() {}

    virtual size_t outs() const { return d_outputs.size(); }

    virtual outpin<T>& output(const size_t n) { return d_outputs[n]; }

    virtual const outpin<T>& output(const size_t n) const { return d_outputs[n]; }

    virtual outputs_t& outputs() { return d_outputs; }

    virtual const outputs_t& outputs() const { return d_outputs; }

    virtual std::string rename(const std::string& name_r)
    {
        for(size_t i = 0; i != outs(); ++i)
        {
            output(i).outpin<T>::rename(name_r + "_out" + static_cast<char>('0' + i));
        }

        return named::rename(name_r);
    }

    virtual void produce() = 0;
};

template<typename T>
class consumer : public virtual node, public detail::consumer
{
    typedef std::vector<inpin<T>> inputs_t;
    inputs_t d_inputs;

protected:
    virtual void disconnect(size_t pin)
    {
        input(pin).disconnect();
    }
    
    virtual void sever()
    {
        for(auto& inpin : d_inputs)
        {
            inpin.disconnect();
        }
    }

    virtual bool incoming()
    {
        for(auto& inpin : d_inputs)
        {
            if(inpin.peek())
            {
                return true;
            }
        }

        return false;
    }

    virtual void operator()()
    {
        state::type s(state());
        
        while(s != state::stopped)
        {
            bool p = false;

            if(s == state::paused)
            {
                std::unique_lock<std::mutex> ul(d_transition_m);
                d_transition_cv.wait(ul, [&s, this](){ return (s = this->state()) != state::paused; });
            }
            else if(s == state::started)
            {
                std::unique_lock<std::mutex> ul(d_transition_m);
                d_transition_cv.wait(ul, [&s, &p, this](){ return ((s = this->state()) != state::started) || (p = this->incoming()); });
            }

            if(p)
            {
                for(size_t i = 0; i != ins(); ++i)
                {
                    if(input(i).peek())
                    {
                        ready(i);
                    }
                }
            }
        }
    }

    friend class graph;

public:
    consumer(const std::string& name_r, const size_t ins) : node(name_r)
    {
        for(size_t i = 0; i != ins; ++i)
        {
            d_inputs.push_back(inpin<T>(name_r + "_in" + static_cast<char>('0' + i), &d_transition_cv, &d_transition_m));
        }
    }

    virtual ~consumer() {}

    virtual size_t ins() const { return d_inputs.size(); }

    virtual inpin<T>& input(const size_t n) { return d_inputs[n]; }

    virtual const inpin<T>& input(const size_t n) const { return d_inputs[n]; }

    virtual inputs_t& inputs() { return d_inputs; }

    virtual const inputs_t& inputs() const { return d_inputs; }

    virtual std::string rename(const std::string& name_r)
    {
        for(size_t i = 0; i != ins(); ++i)
        {
            input(i).inpin<T>::rename(name_r + "_in" + static_cast<char>('0' + i));
        }

        return named::rename(name_r);
    }

    virtual void ready(size_t n) = 0;
};

template<typename C, typename P>
class transformer : public consumer<C>, public producer<P>, public detail::transformer
{
    virtual void produce() {}

protected:
    virtual void sever()
    {
        consumer<C>::sever();

        producer<P>::sever();
    }

    virtual void operator()() { consumer<C>::operator()(); }
    
public:
    transformer(const std::string& name_r, const size_t ins, const size_t outs) : node(name_r), consumer<C>(name_r, ins), producer<P>(name_r, outs)
    {}

    virtual ~transformer() {}

    virtual std::string rename(const std::string& name_r)
    {
        for(size_t i = 0; i != producer<P>::outs(); ++i)
        {
            producer<P>::output(i).outpin<P>::rename(name_r + "_out" + static_cast<char>('0' + i));
        }

        for(size_t i = 0; i != consumer<C>::ins(); ++i)
        {
            consumer<C>::input(i).inpin<C>::rename(name_r + "_in" + static_cast<char>('0' + i));
        }

        return named::rename(name_r);
    }

    virtual void ready(size_t n) = 0;
};

}

#endif

/*
    (C) Copyright Thierry Seegers 2010-2012. Distributed under the following license:

    Boost Software License - Version 1.0 - August 17th, 2003

    Permission is hereby granted, free of charge, to any person or organization
    obtaining a copy of the software and accompanying documentation covered by
    this license (the "Software") to use, reproduce, display, distribute,
    execute, and transmit the Software, and to prepare derivative works of the
    Software, and to permit third-parties to whom the Software is furnished to
    do so, all subject to the following:

    The copyright notices in the Software and this entire statement, including
    the above license grant, this restriction and the following disclaimer,
    must be included in all copies of the Software, in whole or in part, and
    all derivative works of the Software, unless such copies or derivative
    works are solely in the form of machine-executable object code generated by
    a source language processor.

    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
    SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
    FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
    ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
    DEALINGS IN THE SOFTWARE.
*/