helpers.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 2011, 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_HELPERS_H__
#define __TPIE_PIPELINING_HELPERS_H__
 
#include <iostream>
#include <tpie/pipelining/node.h>
#include <tpie/pipelining/pipe_base.h>
#include <tpie/pipelining/factory_helpers.h>
#include <tpie/memory.h>
#include <tpie/tpie_assert.h>
 
namespace tpie::pipelining {
namespace bits {
 
template <typename dest_t>
class ostream_logger_t : public node {
public:
    typedef typename push_type<dest_t>::type item_type;
 
    ostream_logger_t(dest_t dest, std::ostream & log) : dest(std::move(dest)), log(log), begun(false), ended(false) {
        add_push_destination(this->dest);
        set_name("Log", PRIORITY_INSIGNIFICANT);
    }
    void begin() override {
        node::begin();
        begun = true;
    }
    void end() override {
        node::end();
        ended = true;
    }
    void push(const item_type & item) {
        if (!begun) {
            log << "WARNING: push() called before begin(). Calling begin on rest of pipeline." << std::endl;
            begin();
        }
        if (ended) {
            log << "WARNING: push() called after end()." << std::endl;
            ended = false;
        }
        log << "pushing " << item << std::endl;
        dest.push(item);
    }
private:
    dest_t dest;
    std::ostream & log;
    bool begun;
    bool ended;
};
 
template <typename source_t>
class pull_peek_t : public node {
public:
    typedef typename pull_type<source_t>::type item_type;
 
    pull_peek_t(source_t source) : source(std::move(source)) {
        add_pull_source(this->source);
        set_plot_options(PLOT_SIMPLIFIED_HIDE);
    }
 
    void begin() override {
        could_pull = source.can_pull();
        if (could_pull) item=source.pull();
    }
 
    item_type pull() {
        item_type i = item;
        could_pull = source.can_pull();
        if (could_pull) item=source.pull();
        return i;
    }
 
    const item_type & peek() const {
        return item;
    }
 
    bool can_pull() const {
        return could_pull;
    }
 
private:
    item_type item;
    bool could_pull;
    source_t source;
};
 
template <typename T>
class dummydest_t : public node {
public:
    dummydest_t() : buffer(std::make_shared<T>()) {}
 
    typedef T item_type;
    std::shared_ptr<T> buffer;
    void push(const T & el) {
        *buffer = el;
    }
    T pull() {
        return *buffer;
    }
};
 
template <typename dest_t, typename fact2_t>
class fork_t: public node {
public:
    typedef typename fact2_t::constructed_type dest2_t;
    typedef typename push_type<dest_t>::type item_type;
 
    fork_t(dest_t dest, fact2_t fact2) : dest(std::move(dest)), dest2(fact2.construct()) {
        add_push_destination(this->dest);
        add_push_destination(dest2);
    }
    
    void push(const item_type & item) {
        dest.push(item);
        dest2.push(item);
    }
    
private:
    dest_t dest;
    dest2_t dest2;
};
 
 
template <typename source_t, typename dest_fact_t>
class pull_fork_t: public node {
public:
    typedef typename pull_type<source_t>::type item_type;
 
    pull_fork_t(source_t source, dest_fact_t dest_fact)
        : dest(dest_fact.construct())
        , source(std::move(source)) {
        add_pull_source(this->source);
        add_push_destination(dest);
    }
 
    bool can_pull() {return source.can_pull();}
 
    item_type pull() {
        item_type i=source.pull();
        dest.push(i);
        return i;
    }
 
private:
    typename dest_fact_t::constructed_type dest;
    source_t source;
};
 
 
template <typename T>
class null_sink_t: public node {
public:
    typedef T item_type;
 
    void push(const T &) {}
};
 
 
template <typename T>
class zero_source_t: public node {
public:
    typedef T item_type;
    void propagate() override { forward("items", stream_size_type(0)); };
 
    T pull() {tpie_unreachable();}
    bool can_pull() {return false;}
};
 
 
template <typename IT>
class pull_input_iterator_t: public node {
    IT i;
    IT till;
public:
    typedef typename IT::value_type item_type;
    pull_input_iterator_t(IT from, IT to)
        : i(from)
        , till(to)
    {
    }
 
    bool can_pull() {
        return i != till;
    }
 
    item_type pull() {
        return *i++;
    }
};
 
template <typename dest_t, typename IT>
class push_input_iterator_t: public node {
    IT i;
    IT till;
    dest_t dest;
public:
    push_input_iterator_t(dest_t dest, IT from, IT to)
        : i(from)
        , till(to)
        , dest(std::move(dest)) {
        add_push_destination(dest);
    }
    
    void go() override {
        while (i != till) {
            dest.push(*i);
            ++i;
        }
    }
};
 
template <typename Iterator, typename Item = void>
class push_output_iterator_t;
 
template <typename Iterator>
class push_output_iterator_t<Iterator, void> : public node {
    Iterator i;
public:
    typedef typename Iterator::value_type item_type;
    push_output_iterator_t(Iterator to)
        : i(to)
    {
    }
 
    void push(const item_type & item) {
        *i = item;
        ++i;
    }
};
 
template <typename Iterator, typename Item>
class push_output_iterator_t : public node {
    Iterator i;
public:
    typedef Item item_type;
    push_output_iterator_t(Iterator to)
        : i(to)
    {
    }
 
    void push(const item_type & item) {
        *i = item;
        ++i;
    }
};
 
template <typename dest_t, typename IT>
class pull_output_iterator_t: public node {
    IT i;
    dest_t dest;
public:
    pull_output_iterator_t(dest_t dest, IT to)
        : i(to)
        , dest(std::move(dest)) {
        add_pull_source(dest);
    }
    
    void go() override {
        while (dest.can_pull()) {
            *i = dest.pull();
            ++i;
        }
    }
};
 
template <typename dest_t, typename F>
class preparer_t: public node {
private:
    F functor;
    dest_t dest;
public:
    typedef typename push_type<dest_t>::type item_type;
    preparer_t(dest_t dest, const F & functor): functor(functor), dest(std::move(dest)) {
        add_push_destination(this->dest);
    }
    
    void prepare() override {
        functor(*static_cast<node*>(this));
    }
    
    void push(const item_type & item) {dest.push(item);}
};
 
template <typename dest_t, typename F>
class propagater_t: public node {
private:
    F functor;
    dest_t dest;
public:
    typedef typename push_type<dest_t>::type item_type;
    propagater_t(dest_t dest, const F & functor): functor(functor), dest(std::move(dest)) {
        add_push_destination(this->dest);
    }
    
    void propagate() override {
        functor(*static_cast<node*>(this));
    }
    
    void push(const item_type & item) {dest.push(item);}
};
 
template <typename dest_t, typename src_fact_t>
struct zip_t: public node {
    typedef typename src_fact_t::constructed_type src_t;
    typedef typename push_type<dest_t>::type::first_type item_type;
    
    zip_t(dest_t dest, src_fact_t src_fact)
        : src(src_fact.construct()), dest(std::move(dest)) {
        add_push_destination(this->dest);
        add_pull_source(src);
    }
 
    void push(const item_type & item) {
        tp_assert(src.can_pull(), "We should be able to pull");
        dest.push(std::make_pair(item, src.pull()));
    }
private:
    src_t src;
    dest_t dest;
};
 
template <typename dest1_t, typename fact2_t>
struct unzip_t: public node {
    typedef typename fact2_t::constructed_type dest2_t;
    typedef typename push_type<dest1_t>::type first_type;
    typedef typename push_type<dest2_t>::type second_type;
    typedef std::pair<first_type, second_type> item_type;
    
    unzip_t(dest1_t dest1, fact2_t fact2) : dest1(std::move(dest1)), dest2(fact2.construct()) {
        add_push_destination(this->dest1);
        add_push_destination(dest2);
    }
    
    void push(const item_type & item) {
        dest2.push(item.second);
            dest1.push(item.first);
    }
private:
    dest1_t dest1;
    dest2_t dest2;
};
 
template <typename dest_t, typename T>
class item_type_t: public node {
private:
    dest_t dest;
public:
    typedef T item_type;
    item_type_t(dest_t dest): dest(std::move(dest)) {}
    void push(const item_type & item) {dest.push(item);}
};
 
 
template <typename dest_t, typename fact_t>
class pull_source_t: public node {
public:
    typedef typename fact_t::constructed_type source_t;
    typedef typename push_type<dest_t>::type item_type;
    
    pull_source_t(dest_t dest, fact_t fact)
        : dest(std::move(dest)), src(fact.construct()) {
        add_push_destination(dest);
        add_pull_source(src);
    }
 
    void propagate() override {
        size_t size = fetch<stream_size_type>("items");
        set_steps(size);
    }
    
    void go() override {
        while (src.can_pull()) {
            dest.push(src.pull());
            step();
        }
    }
    
private:
    dest_t dest;
    source_t src;
};
 
template <typename dest_t, typename equal_t>
class unique_t : public node {
public:
    typedef typename push_type<dest_t>::type item_type;
 
    unique_t(dest_t dest, equal_t equal)
        : equal(equal), dest(std::move(dest)) {}
 
    void begin() override {
        first = true;
    }
 
    void push(const item_type & item) {
        if (!first && equal(prev, item))
            return;
        first = false;
        dest.push(item);
        prev = item;
    }
 
private:
    bool first;
    equal_t equal;
    item_type prev;
    dest_t dest;
};
 
template <typename src_t, typename equal_t>
class pull_unique_t : public node {
public:
    typedef typename pull_type<src_t>::type item_type;
 
    pull_unique_t(src_t src, equal_t equal)
        : equal(equal), src(std::move(src)) {}
 
    void begin() override {
        has_item = src.can_pull();
        if (!has_item) return;
        item = src.pull();
    }
 
    bool can_pull() {
        return has_item;
    }
 
    item_type over() {
        has_item = false;
        return item;
    }
 
    item_type pull() {
        if (!src.can_pull()) return over();
 
        item_type next;
        do next = src.pull();
        while (src.can_pull() && equal(item, next));
 
        if (equal(item, next)) return over();
        std::swap(next, item);
        return next;
    }
 
private:
    bool has_item;
    equal_t equal;
    item_type item;
    src_t src;
};
 
} // namespace bits
 
inline pipe_middle<factory<bits::ostream_logger_t, std::ostream &> >
cout_logger() {
    return {std::cout};
}
 
typedef pullpipe_middle<factory<bits::pull_peek_t> > pull_peek;
 
 
template <typename fact_t>
pipe_middle<tfactory<bits::fork_t, Args<fact_t>, fact_t> >
fork(pipe_end<fact_t> to) {
    return {std::move(to.factory)};
}
 
template <typename dest_fact_t>
pullpipe_middle<tfactory<bits::pull_fork_t, Args<dest_fact_t>, dest_fact_t> >
pull_fork(dest_fact_t dest_fact) {
    return {std::move(dest_fact)};
}
 
template <typename fact_t>
pipe_middle<tfactory<bits::unzip_t, Args<fact_t>, fact_t> >
unzip(pipe_end<fact_t> to) {
    return {std::move(to.factory)};
}
 
template <typename fact_t>
pipe_middle<tfactory<bits::zip_t, Args<fact_t>, fact_t> >
zip(pullpipe_begin<fact_t> from) {
    return {std::move(from.factory)};
}
 
template <typename T>
inline pipe_end<termfactory<bits::null_sink_t<T>>>
null_sink() {return {};}
 
template <typename T>
inline pullpipe_begin<termfactory<bits::zero_source_t<T>>>
zero_source() {return {}; }
 
 
template <template <typename dest_t> class Fact, typename... T>
pipe_begin<factory<Fact, T...> > make_pipe_begin(T... t) {
    return {t...};
}
 
template <template <typename dest_t> class Fact, typename... T>
pipe_middle<factory<Fact, T...> > make_pipe_middle(T... t) {
    return {t...};
}
 
template <typename Fact, typename... T>
pipe_end<termfactory<Fact, T...> > make_pipe_end(T ... t) {
    return {t...};
}
 
template <typename IT>
pullpipe_begin<termfactory<bits::pull_input_iterator_t<IT>, IT, IT> > pull_input_iterator(IT begin, IT end) {
    return {begin, end};
}
 
template <typename IT>
pipe_begin<tfactory<bits::push_input_iterator_t, Args<IT>, IT, IT> > push_input_iterator(IT begin, IT end) {
    return {begin, end};
}
 
template <typename IT>
pipe_end<termfactory<bits::push_output_iterator_t<IT>, IT> > push_output_iterator(IT to) {
    return {to};
}
 
template <typename Item, typename IT>
pipe_end<termfactory<bits::push_output_iterator_t<IT, Item>, IT> > typed_push_output_iterator(IT to) {
    return {to};
}
 
template <typename IT>
pullpipe_end<tfactory<bits::pull_output_iterator_t, Args<IT>, IT> > pull_output_iterator(IT to) {
    return {to};
}
 
template <typename F>
pipe_middle<tfactory<bits::preparer_t, Args<F>, F> > preparer(const F & functor) {
    return {functor};
}
 
template <typename F>
pipe_middle<tfactory<bits::propagater_t, Args<F>, F> > propagater(const F & functor) {
    return {functor};
}
 
template <typename T>
pipe_middle<tfactory<bits::item_type_t, Args<T> > > item_type() {
    return {};
}
 
template <typename fact_t>
pipe_begin<tfactory<bits::pull_source_t, Args<fact_t>, fact_t> >
pull_source(pullpipe_begin<fact_t> from) {
    return {std::move(from.factory)};
}
 
template <typename equal_t>
pipe_middle<tfactory<bits::unique_t, Args<equal_t>, equal_t> > unique(equal_t equal) {
    return {equal};
}
 
template <typename equal_t>
pullpipe_middle<tfactory<bits::pull_unique_t, Args<equal_t>, equal_t> > pull_unique(equal_t equal) {
    return {equal};
}
 
} //namespace tpie::pipelining
#endif //__TPIE_PIPELINING_HELPERS_H__