virtual.h

Go to the documentation of this file.

// -*- mode: c++; tab-width: 4; indent-tabs-mode: t; c-file-style: "stroustrup"; -*-
// vi:set ts=4 sts=4 sw=4 noet cino+=(0 :
// 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_VIRTUAL_H__
#define __TPIE_PIPELINING_VIRTUAL_H__

namespace tpie {

namespace pipelining {

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

namespace bits {

template <typename T>
struct maybe_add_const_ref {
    typedef const T & type;
};

template <typename T>
struct maybe_add_const_ref<const T &> {
    typedef const T & type;
};

template <typename T>
struct maybe_add_const_ref<const T *> {
    typedef const T * type;
};

template <typename T>
struct maybe_add_const_ref<T &> {
    typedef T & type;
};

template <typename T>
struct maybe_add_const_ref<T *> {
    typedef T * type;
};

template <typename Input>
class virtsrc : public node {
    typedef typename maybe_add_const_ref<Input>::type input_type;

public:
    virtual const node_token & get_token() = 0;
    virtual void push(input_type v) = 0;
};

template <typename dest_t>
class virtsrc_impl : public virtsrc<typename dest_t::item_type> {
public:
    typedef typename dest_t::item_type item_type;

private:
    typedef typename maybe_add_const_ref<item_type>::type input_type;
    dest_t dest;

public:
    virtsrc_impl(const dest_t & dest)
        : dest(dest)
    {
        node::add_push_destination(dest);
        this->set_name("Virtual source", PRIORITY_INSIGNIFICANT);
    }

    const node_token & get_token() {
        return node::get_token();
    }

    void push(input_type v) {
        dest.push(v);
    }
};

template <typename Output>
class virtrecv : public node {
    virtrecv *& m_self;
    virtsrc<Output> * m_virtdest;

public:
    typedef Output item_type;

    virtrecv(virtrecv *& self)
        : m_self(self)
        , m_virtdest(0)
    {
        m_self = this;
        set_name("Virtual destination", PRIORITY_INSIGNIFICANT);
    }

    virtrecv(const virtrecv & other)
        : node(other)
        , m_self(other.m_self)
        , m_virtdest(other.m_virtdest)
    {
        m_self = this;
    }

    void begin() {
        node::begin();
        if (m_virtdest == 0) {
            throw tpie::exception("No virtual destination");
        }
    }

    void push(typename maybe_add_const_ref<Output>::type v) {
        m_virtdest->push(v);
    }

    void set_destination(virtsrc<Output> * dest) {
        if (m_virtdest != 0) {
            throw tpie::exception("Virtual destination set twice");
        }

        m_virtdest = dest;
        add_push_destination(dest->get_token());
    }
};

class virt_node {
public:
    typedef boost::shared_ptr<virt_node> ptr;

private:
    std::auto_ptr<node> m_pipeSegment;
    std::auto_ptr<virtual_container> m_container;
    ptr m_left;
    ptr m_right;

public:
    static ptr take_own(node * pipe) {
        virt_node * n = new virt_node();
        n->m_pipeSegment.reset(pipe);
        ptr res(n);
        return res;
    }

    static ptr combine(ptr left, ptr right) {
        virt_node * n = new virt_node();
        n->m_left = left;
        n->m_right = right;
        ptr res(n);
        return res;
    }

    void set_container(virtual_container * ctr) {
        m_container.reset(ctr);
    }
};

template <typename T, typename U, typename Result>
struct assert_types_equal_and_return {
    static Result go(...) {
        throw virtual_chunk_missing_middle();
    }
};

template <typename T, typename Result>
struct assert_types_equal_and_return<T, T, Result> {
    static Result go(Result r) {
        return r;
    }
};

class virtual_chunk_base : public pipeline_base {
    // pipeline_base has virtual dtor and shared_ptr to m_segmap
protected:
    virt_node::ptr m_node;
public:
    virtual_chunk_base() {}

    virt_node::ptr get_node() const { return m_node; }
    virtual_chunk_base(node_map::ptr segmap, virt_node::ptr ptr)
        : m_node(ptr)
    {
        this->m_segmap = segmap;
    }

    virtual_chunk_base(node_map::ptr segmap) {
        this->m_segmap = segmap;
    }

    void set_container(virtual_container * ctr) {
        m_node->set_container(ctr);
    }

    operator bool() { return m_node; }
};

} // namespace bits

// Predeclare
template <typename Input>
class virtual_chunk_end;

// Predeclare
template <typename Input, typename Output>
class virtual_chunk;

// Predeclare
template <typename Output>
class virtual_chunk_begin;

namespace bits {

class access {
    template <typename>
    friend class pipelining::virtual_chunk_end;
    template <typename, typename>
    friend class pipelining::virtual_chunk;
    template <typename>
    friend class pipelining::virtual_chunk_begin;

    template <typename Input>
    static virtsrc<Input> * get_source(const virtual_chunk_end<Input> &);
    template <typename Input, typename Output>
    static virtsrc<Input> * get_source(const virtual_chunk<Input, Output> &);
    template <typename Input, typename Output>
    static virtrecv<Output> * get_destination(const virtual_chunk<Input, Output> &);
    template <typename Output>
    static virtrecv<Output> * get_destination(const virtual_chunk_begin<Output> &);
};

} // namespace bits

template <typename Input>
class virtual_chunk_end : public bits::virtual_chunk_base {
    friend class bits::access;
    typedef bits::access acc;
    typedef bits::virtsrc<Input> src_type;
    src_type * m_src;

    src_type * get_source() const { return m_src; }

public:
    virtual_chunk_end()
        : m_src(0)
    {}

    template <typename fact_t>
    virtual_chunk_end(const pipe_end<fact_t> & pipe, virtual_container * ctr = 0) {
        *this = pipe;
        set_container(ctr);
    }

    template <typename Mid>
    virtual_chunk_end(const virtual_chunk<Input, Mid> & left,
                      const virtual_chunk_end<Mid> & right);

    template <typename fact_t>
    virtual_chunk_end & operator=(const pipe_end<fact_t> & pipe) {
        if (this->m_node) {
            log_error() << "Virtual chunk assigned twice" << std::endl;
            throw tpie::exception("Virtual chunk assigned twice");
        }

        typedef typename fact_t::constructed_type constructed_type;
        m_src = new bits::virtsrc_impl<constructed_type>(pipe.factory.construct());
        this->m_node = bits::virt_node::take_own(m_src);
        this->m_segmap = m_src->get_node_map();

        return *this;
    }
};

template <typename Input, typename Output=Input>
class virtual_chunk : public bits::virtual_chunk_base {
    friend class bits::access;
    typedef bits::access acc;
    typedef bits::virtsrc<Input> src_type;
    typedef bits::virtrecv<Output> recv_type;
    src_type * m_src;
    recv_type * m_recv;
    src_type * get_source() const { return m_src; }
    recv_type * get_destination() const { return m_recv; }

public:
    virtual_chunk()
        : m_src(0)
        , m_recv(0)
    {}

    template <typename fact_t>
    virtual_chunk(const pipe_middle<fact_t> & pipe, virtual_container * ctr = 0) {
        *this = pipe;
        set_container(ctr);
    }

    template <typename Mid>
    virtual_chunk(const virtual_chunk<Input, Mid> & left,
                  const virtual_chunk<Mid, Output> & right)
        : virtual_chunk_base(left.get_node_map(), bits::virt_node::combine(left.get_node(), right.get_node()))
    {
        m_src = acc::get_source(left);
        m_recv = acc::get_destination(right);
    }

    template <typename fact_t>
    virtual_chunk & operator=(const pipe_middle<fact_t> & pipe) {
        if (this->m_node) {
            log_error() << "Virtual chunk assigned twice" << std::endl;
            throw tpie::exception("Virtual chunk assigned twice");
        }
        typedef typename fact_t::template constructed<recv_type>::type constructed_type;
        recv_type temp(m_recv);
        m_src = new bits::virtsrc_impl<constructed_type>(pipe.factory.construct(temp));
        this->m_node = bits::virt_node::take_own(m_src);
        this->m_segmap = temp.get_node_map();

        return *this;
    }

    template <typename NextOutput>
    virtual_chunk<Input, NextOutput> operator|(virtual_chunk<Output, NextOutput> dest) {
        if (!*this) {
            return *bits::assert_types_equal_and_return<Input, Output, virtual_chunk<Input, NextOutput> *>
                ::go(&dest);
        }
        m_recv->set_destination(acc::get_source(dest));
        return virtual_chunk<Input, NextOutput>(*this, dest);
    }

    virtual_chunk_end<Input> operator|(virtual_chunk_end<Output> dest) {
        if (!*this) {
            return *bits::assert_types_equal_and_return<Input, Output, virtual_chunk_end<Input> *>
                ::go(&dest);
        }
        m_recv->set_destination(acc::get_source(dest));
        return virtual_chunk_end<Input>(*this, dest);
    }
};

template <typename Input>
template <typename Mid>
virtual_chunk_end<Input>::virtual_chunk_end(const virtual_chunk<Input, Mid> & left,
                                            const virtual_chunk_end<Mid> & right)
    : virtual_chunk_base(left.get_node_map(),
                         bits::virt_node::combine(left.get_node(), right.get_node()))
{
    m_src = acc::get_source(left);
}

template <typename Output>
class virtual_chunk_begin : public bits::virtual_chunk_base {
    friend class bits::access;
    typedef bits::access acc;
    typedef bits::virtrecv<Output> recv_type;
    recv_type * m_recv;
    recv_type * get_destination() const { return m_recv; }

public:
    virtual_chunk_begin()
        : m_recv(0)
    {}

    template <typename fact_t>
    virtual_chunk_begin(const pipe_begin<fact_t> & pipe, virtual_container * ctr = 0) {
        *this = pipe;
        set_container(ctr);
    }

    template <typename Mid>
    virtual_chunk_begin(const virtual_chunk_begin<Mid> & left,
                        const virtual_chunk<Mid, Output> & right)
        : virtual_chunk_base(left.get_node_map(),
                             bits::virt_node::combine(left.get_node(), right.get_node()))
    {
        m_recv = acc::get_destination(right);
    }

    template <typename fact_t>
    virtual_chunk_begin & operator=(const pipe_begin<fact_t> & pipe) {
        if (this->m_node) {
            log_error() << "Virtual chunk assigned twice" << std::endl;
            throw tpie::exception("Virtual chunk assigned twice");
        }
        typedef typename fact_t::template constructed<recv_type>::type constructed_type;
        recv_type temp(m_recv);
        this->m_node = bits::virt_node::take_own(new constructed_type(pipe.factory.construct(temp)));
        this->m_segmap = m_recv->get_node_map();
        return *this;
    }

    template <typename NextOutput>
    virtual_chunk_begin<NextOutput> operator|(virtual_chunk<Output, NextOutput> dest) {
        if (!*this) throw virtual_chunk_missing_begin();
        if (!dest) {
            return *bits::assert_types_equal_and_return<Output, NextOutput, virtual_chunk_begin<NextOutput> *>
                ::go(this);
        }
        m_recv->set_destination(acc::get_source(dest));
        return virtual_chunk_begin<NextOutput>(*this, dest);
    }

    virtual_chunk_base operator|(virtual_chunk_end<Output> dest) {
        if (!*this) throw virtual_chunk_missing_begin();
        if (!dest) throw virtual_chunk_missing_end();
        m_recv->set_destination(acc::get_source(dest));
        return virtual_chunk_base(this->m_segmap,
                                  bits::virt_node::combine(get_node(), dest.get_node()));
    }
};

namespace bits {

template <typename Input>
virtsrc<Input> * access::get_source(const virtual_chunk_end<Input> & chunk) {
    return chunk.get_source();
}

template <typename Input, typename Output>
virtsrc<Input> * access::get_source(const virtual_chunk<Input, Output> & chunk) {
    return chunk.get_source();
}

template <typename Input, typename Output>
virtrecv<Output> * access::get_destination(const virtual_chunk<Input, Output> & chunk) {
    return chunk.get_destination();
}

template <typename Output>
virtrecv<Output> * access::get_destination(const virtual_chunk_begin<Output> & chunk) {
    return chunk.get_destination();
}

} // namespace bits

} // namespace pipelining

} // namespace tpie

#endif // __TPIE_PIPELINING_VIRTUAL_H__