flow_graph.h

Go to the documentation of this file.

/*
    Copyright 2005-2011 Intel Corporation.  All Rights Reserved.

    The source code contained or described herein and all documents related
    to the source code ("Material") are owned by Intel Corporation or its
    suppliers or licensors.  Title to the Material remains with Intel
    Corporation or its suppliers and licensors.  The Material is protected
    by worldwide copyright laws and treaty provisions.  No part of the
    Material may be used, copied, reproduced, modified, published, uploaded,
    posted, transmitted, distributed, or disclosed in any way without
    Intel's prior express written permission.

    No license under any patent, copyright, trade secret or other
    intellectual property right is granted to or conferred upon you by
    disclosure or delivery of the Materials, either expressly, by
    implication, inducement, estoppel or otherwise.  Any license under such
    intellectual property rights must be express and approved by Intel in
    writing.
*/

#ifndef __TBB_flow_graph_H
#define __TBB_flow_graph_H

#include "tbb_stddef.h"
#include "atomic.h"
#include "spin_mutex.h"
#include "null_mutex.h"
#include "spin_rw_mutex.h"
#include "null_rw_mutex.h"
#include "task.h"
#include "concurrent_vector.h"
#include "internal/_aggregator_impl.h"

// use the VC10 or gcc version of tuple if it is available.
#if TBB_IMPLEMENT_CPP0X && (!defined(_MSC_VER) || _MSC_VER < 1600)
#define TBB_PREVIEW_TUPLE 1
#include "compat/tuple"
#else
#include <tuple>
#endif

#include<list>
#include<queue>

namespace tbb {
namespace flow {

enum concurrency { unlimited = 0, serial = 1 };

namespace interface6 {

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

class continue_msg {};
        
template< typename T > class sender;
template< typename T > class receiver;
class continue_receiver;
        
template< typename T >
class sender {
public:
    typedef T output_type;
        
    typedef receiver<T> successor_type;
        
    virtual ~sender() {}
        
    virtual bool register_successor( successor_type &r ) = 0;
        
    virtual bool remove_successor( successor_type &r ) = 0;
        
    virtual bool try_get( T & ) { return false; }
        
    virtual bool try_reserve( T & ) { return false; }
        
    virtual bool try_release( ) { return false; }
        
    virtual bool try_consume( ) { return false; }
        
};
        
        
template< typename T >
class receiver {
public:
        
    typedef T input_type;
        
    typedef sender<T> predecessor_type;
        
    virtual ~receiver() {}
        
    virtual bool try_put( const T& t ) = 0;
        
    virtual bool register_predecessor( predecessor_type & ) { return false; }
        
    virtual bool remove_predecessor( predecessor_type & ) { return false; }
        
};
        

class continue_receiver : public receiver< continue_msg > {
public:
        
    typedef continue_msg input_type;
        
    typedef sender< continue_msg > predecessor_type;
        
    continue_receiver( int number_of_predecessors = 0 ) { 
        my_predecessor_count = my_initial_predecessor_count = number_of_predecessors;
        my_current_count = 0;
    }
        
    continue_receiver( const continue_receiver& src ) : receiver<continue_msg>() { 
        my_predecessor_count = my_initial_predecessor_count = src.my_initial_predecessor_count;
        my_current_count = 0;
    }
        
    virtual ~continue_receiver() { }
        
    /* override */ bool register_predecessor( predecessor_type & ) {
        spin_mutex::scoped_lock l(my_mutex);
        ++my_predecessor_count;
        return true;
    }
        

    /* override */ bool remove_predecessor( predecessor_type & ) {
        spin_mutex::scoped_lock l(my_mutex);
        --my_predecessor_count;
        return true;
    }
        

    /* override */ bool try_put( const input_type & ) {
        {
            spin_mutex::scoped_lock l(my_mutex);
            if ( ++my_current_count < my_predecessor_count ) 
                return true;
            else
                my_current_count = 0;
        }
        execute();
        return true;
    }
        
protected:
        
    spin_mutex my_mutex;
    int my_predecessor_count;
    int my_current_count;
    int my_initial_predecessor_count;
        

    virtual void execute() = 0;
        
};

#include "internal/_flow_graph_impl.h"
using namespace internal::graph_policy_namespace;


class graph : tbb::internal::no_copy {
        
    template< typename Body >
    class run_task : public task {
    public: 
        run_task( Body& body ) : my_body(body) {}
        task *execute() {
            my_body();
            return NULL;
        }
    private:
        Body my_body;
    };
        
    template< typename Receiver, typename Body >
    class run_and_put_task : public task {
    public: 
        run_and_put_task( Receiver &r, Body& body ) : my_receiver(r), my_body(body) {}
        task *execute() {
            my_receiver.try_put( my_body() );
            return NULL;
        }
    private:
        Receiver &my_receiver;
        Body my_body;
    };
        
public:
        
        
    graph() : my_root_task( new ( task::allocate_root( ) ) empty_task ) {
        my_root_task->set_ref_count(1);
    }
        

    ~graph() {
        wait_for_all();
        my_root_task->set_ref_count(0);
        task::destroy( *my_root_task );
    }
        
        

    void increment_wait_count() { 
        if (my_root_task)
            my_root_task->increment_ref_count();
    }
        

    void decrement_wait_count() { 
        if (my_root_task)
            my_root_task->decrement_ref_count(); 
    }
        

    template< typename Receiver, typename Body >
        void run( Receiver &r, Body body ) {
       task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
           run_and_put_task< Receiver, Body >( r, body ) );
    }
        

    template< typename Body >
    void run( Body body ) {
       task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
           run_task< Body >( body ) );
    }
        

    void wait_for_all() {
        if (my_root_task)
            my_root_task->wait_for_all();
        my_root_task->set_ref_count(1);
    }
        
    task * root_task() {
        return my_root_task;
    }
        
private:
        
    task *my_root_task;
        
};

#include "internal/_flow_graph_node_impl.h"

template < typename Output >
class source_node : public graph_node, public sender< Output > {
public:
        
    typedef Output output_type;           
        
    typedef receiver< Output > successor_type;
        
    template< typename Body >
    source_node( graph &g, Body body, bool is_active = true )
        : my_root_task(g.root_task()), my_active(is_active), init_my_active(is_active),
        my_body( new internal::source_body_leaf< output_type, Body>(body) ),
        my_reserved(false), my_has_cached_item(false) 
    { 
        my_successors.set_owner(this);
    }
        
    source_node( const source_node& src ) :
        graph_node(), sender<Output>(),
        my_root_task( src.my_root_task), my_active(src.init_my_active),
        init_my_active(src.init_my_active), my_body( src.my_body->clone() ),
        my_reserved(false), my_has_cached_item(false)
    {
        my_successors.set_owner(this);
    }

    ~source_node() { delete my_body; }
        
    /* override */ bool register_successor( receiver<output_type> &r ) {
        spin_mutex::scoped_lock lock(my_mutex);
        my_successors.register_successor(r);
        if ( my_active )
            spawn_put();
        return true;
    }
        
    /* override */ bool remove_successor( receiver<output_type> &r ) {
        spin_mutex::scoped_lock lock(my_mutex);
        my_successors.remove_successor(r);
        return true;
    }
        
    /*override */ bool try_get( output_type &v ) {
        spin_mutex::scoped_lock lock(my_mutex);
        if ( my_reserved )  
            return false;
        
        if ( my_has_cached_item ) {
            v = my_cached_item;
            my_has_cached_item = false;
        } else if ( (*my_body)(v) == false ) {
            return false;
        }
        return true;
    }
        
    /* override */ bool try_reserve( output_type &v ) {
        spin_mutex::scoped_lock lock(my_mutex);
        if ( my_reserved ) {
            return false;
        }
        
        if ( !my_has_cached_item && (*my_body)(my_cached_item) )  
            my_has_cached_item = true;
        
        if ( my_has_cached_item ) {
            v = my_cached_item;
            my_reserved = true;
            return true;
        } else {
            return false;
        }
    }
        

    /* override */ bool try_release( ) {
        spin_mutex::scoped_lock lock(my_mutex);
        __TBB_ASSERT( my_reserved && my_has_cached_item, "releasing non-existent reservation" );
        my_reserved = false;
        spawn_put();
        return true;
    }
        
    /* override */ bool try_consume( ) {
        spin_mutex::scoped_lock lock(my_mutex);
        __TBB_ASSERT( my_reserved && my_has_cached_item, "consuming non-existent reservation" );
        my_reserved = false;
        my_has_cached_item = false;
        if ( !my_successors.empty() ) {
            spawn_put();
        }
        return true;
    }
        
    void activate() {
        spin_mutex::scoped_lock lock(my_mutex);
        my_active = true;
        if ( !my_successors.empty() )
            spawn_put();
    }
        
private:
        
    task *my_root_task;
    spin_mutex my_mutex;
    bool my_active;
    bool init_my_active;
    internal::source_body<output_type> *my_body;
    internal::broadcast_cache< output_type > my_successors;
    bool my_reserved;
    bool my_has_cached_item;
    output_type my_cached_item;
        
    friend class internal::source_task< source_node< output_type > >;
        
    /* override */ void apply_body( ) {
        output_type v;
        if ( try_reserve(v) == false )
            return;
        
        if ( my_successors.try_put( v ) ) 
            try_consume();
        else
            try_release();
    }
        
    /* override */ void spawn_put( ) {
        task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
           internal::source_task< source_node< output_type > >( *this ) ); 
    }
        
};
        
template < typename Input, typename Output = continue_msg, graph_buffer_policy = queueing, typename Allocator=cache_aligned_allocator<Input> >
class function_node : public graph_node, public internal::function_input<Input,Output,Allocator>, public internal::function_output<Output> {
public:
        
    typedef Input input_type;
    typedef Output output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
    typedef internal::function_output<output_type> fOutput_type;
        
    template< typename Body >
    function_node( graph &g, size_t concurrency, Body body )
    : internal::function_input<input_type,output_type,Allocator>( g, concurrency, body ) {
    }

    function_node( const function_node& src ) : 
        graph_node(), internal::function_input<input_type,output_type,Allocator>( src ),
        fOutput_type() {}
        
protected:

    /* override */ internal::broadcast_cache<output_type> &successors () { return fOutput_type::my_successors; }
        
};

template < typename Input, typename Output, typename Allocator >
class function_node<Input,Output,queueing,Allocator> : public graph_node, public internal::function_input<Input,Output,Allocator>, public internal::function_output<Output> {
public:
        
    typedef Input input_type;
    typedef Output output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
    typedef internal::function_input<input_type,output_type,Allocator> fInput_type;
    typedef internal::function_input_queue<input_type, Allocator> queue_type;
    typedef internal::function_output<output_type> fOutput_type;
        
    template< typename Body >
    function_node( graph &g, size_t concurrency, Body body ) : fInput_type( g, concurrency, body, new queue_type() ) {
    }

    function_node( const function_node& src ) : 
        graph_node(), fInput_type( src, new queue_type() ) , fOutput_type() { }

protected:

    /* override */ internal::broadcast_cache<output_type> &successors () { return fOutput_type::my_successors; }
        
};

#include "tbb/internal/_flow_graph_types_impl.h"

#if TBB_PREVIEW_GRAPH_NODES
// Output is a tuple of output types.
template < typename Input, typename Output, graph_buffer_policy = queueing, typename Allocator=cache_aligned_allocator<Input> >
class multioutput_function_node : 
    public graph_node, 
    public internal::multioutput_function_input
    <  
        Input, 
        typename internal::wrap_tuple_elements<
            std::tuple_size<Output>::value,  // #elements in tuple
            internal::function_output,  // wrap this around each element
            Output // the tuple providing the types
        >::type,
        Allocator
    > {
private:
    static const int N = std::tuple_size<Output>::value;
public:
    typedef Input input_type;
    typedef typename internal::wrap_tuple_elements<N,internal::function_output, Output>::type ports_type;
private:
    typedef typename internal::multioutput_function_input<input_type, ports_type, Allocator> base_type;
    typedef typename internal::function_input_queue<input_type,Allocator> queue_type;
public:
    template<typename Body>
    multioutput_function_node( graph &g, size_t concurrency, Body body ) : base_type(g,concurrency, body) {}
    multioutput_function_node( const multioutput_function_node &other) :
        graph_node(), base_type(other) {}
    // all the guts are in multioutput_function_input...

};  // multioutput_function_node
        
template < typename Input, typename Output, typename Allocator >
class multioutput_function_node<Input,Output,queueing,Allocator> : public graph_node, public internal::multioutput_function_input<Input, 
    typename internal::wrap_tuple_elements<std::tuple_size<Output>::value, internal::function_output, Output>::type, Allocator> {
    static const int N = std::tuple_size<Output>::value;
public:
    typedef Input input_type;
    typedef typename internal::wrap_tuple_elements<N, internal::function_output, Output>::type ports_type;
private:
    typedef typename internal::multioutput_function_input<input_type, ports_type, Allocator> base_type;
    typedef typename internal::function_input_queue<input_type,Allocator> queue_type;
public:

    template<typename Body>
    multioutput_function_node( graph &g, size_t concurrency, Body body) : base_type(g,concurrency, body, new queue_type()) {}
    multioutput_function_node( const multioutput_function_node &other) :
        graph_node(), base_type(other, new queue_type()) {}

};  // multioutput_function_node

//  successors.  The node has unlimited concurrency, so though it is marked as
//  "rejecting" it does not reject inputs.
template<typename TupleType, typename Allocator=cache_aligned_allocator<TupleType> >
class split_node : public multioutput_function_node<TupleType, TupleType, rejecting, Allocator> {
    static const int N = std::tuple_size<TupleType>::value;
    typedef multioutput_function_node<TupleType,TupleType,rejecting,Allocator> base_type;
public:
    typedef typename base_type::ports_type ports_type;
private:

    struct splitting_body {
        void operator()(const TupleType& t, ports_type &p) {
            internal::emit_element<N>::emit_this(t, p);
        }
    };
public:
    typedef TupleType input_type;
    typedef Allocator allocator_type;
    split_node(graph &g) : base_type(g, unlimited, splitting_body()) { }
    split_node( const split_node & other) : base_type(other) { }
};
#endif  // TBB_PREVIEW_GRAPH_NODES

template <typename Output>
class continue_node : public graph_node, public internal::continue_input<Output>, public internal::function_output<Output> {
public:
        
    typedef continue_msg input_type;
    typedef Output output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
    typedef internal::function_output<output_type> fOutput_type;
        
     template <typename Body >
     continue_node( graph &g, Body body )
             : internal::continue_input<output_type>( g, body ) {
     }
        
    template <typename Body >
    continue_node( graph &g, int number_of_predecessors, Body body )
        : internal::continue_input<output_type>( g, number_of_predecessors, body )
    {
    }
 
    continue_node( const continue_node& src ) :
        graph_node(), internal::continue_input<output_type>(src),
        internal::function_output<Output>() { }

protected:
        
    /* override */ internal::broadcast_cache<output_type> &successors () { return fOutput_type::my_successors; }
        
};
        
template< typename T >
class overwrite_node : public graph_node, public receiver<T>, public sender<T> {
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    overwrite_node() : my_buffer_is_valid(false) {
        my_successors.set_owner( this );
    }

    // Copy constructor; doesn't take anything from src; default won't work
    overwrite_node( const overwrite_node& ) : 
        graph_node(), receiver<T>(), sender<T>(), my_buffer_is_valid(false) {
        my_successors.set_owner( this );
    }
        
    ~overwrite_node() {}
        
    /* override */ bool register_successor( successor_type &s ) {
        spin_mutex::scoped_lock l( my_mutex );
        if ( my_buffer_is_valid ) {
            // We have a valid value that must be forwarded immediately.
            if ( s.try_put( my_buffer ) || !s.register_predecessor( *this  ) ) {
                // We add the successor: it accepted our put or it rejected it but won't let use become a predecessor
                my_successors.register_successor( s );
                return true;
            } else {
                // We don't add the successor: it rejected our put and we became its predecessor instead
                return false;
            }
        } else {
            // No valid value yet, just add as successor
            my_successors.register_successor( s );
            return true;
        }
    }
        
    /* override */ bool remove_successor( successor_type &s ) {
        spin_mutex::scoped_lock l( my_mutex );
        my_successors.remove_successor(s);
        return true;
    }
        
    /* override */ bool try_put( const T &v ) {
        spin_mutex::scoped_lock l( my_mutex );
        my_buffer = v;
        my_buffer_is_valid = true;
        my_successors.try_put(v);
        return true;
    }
        
    /* override */ bool try_get( T &v ) {
        spin_mutex::scoped_lock l( my_mutex );
        if ( my_buffer_is_valid ) {
            v = my_buffer;
            return true;
        } else {
            return false;
        }
    }
        
    bool is_valid() {
       spin_mutex::scoped_lock l( my_mutex );
       return my_buffer_is_valid;
    }
        
    void clear() {
       spin_mutex::scoped_lock l( my_mutex );
       my_buffer_is_valid = false;
    }
        
protected:
        
    spin_mutex my_mutex;
    internal::broadcast_cache< T, null_rw_mutex > my_successors;
    T my_buffer;
    bool my_buffer_is_valid;
        
};
        
template< typename T >
class write_once_node : public overwrite_node<T> {
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    write_once_node() : overwrite_node<T>() {}

    write_once_node( const write_once_node& src ) : overwrite_node<T>(src) {}

    /* override */ bool try_put( const T &v ) {
        spin_mutex::scoped_lock l( this->my_mutex );
        if ( this->my_buffer_is_valid ) {
            return false;
        } else {
            this->my_buffer = v;
            this->my_buffer_is_valid = true;
            this->my_successors.try_put(v);
            return true;
        }
    }
};
        
template <typename T>
class broadcast_node : public graph_node, public receiver<T>, public sender<T> {
        
    internal::broadcast_cache<T> my_successors;
        
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    broadcast_node( ) {
        my_successors.set_owner( this );
    }
        
    // Copy constructor
    broadcast_node( const broadcast_node& ) : graph_node(), receiver<T>(), sender<T>() {
        my_successors.set_owner( this );
    }
        
    virtual bool register_successor( receiver<T> &r ) {
        my_successors.register_successor( r );
        return true;
    }
        
    virtual bool remove_successor( receiver<T> &r ) {
        my_successors.remove_successor( r );
        return true;
    }
        
    /* override */ bool try_put( const T &t ) {
        my_successors.try_put(t);
        return true;
    }
        
};

#include "internal/_flow_graph_item_buffer_impl.h"

template <typename T, typename A=cache_aligned_allocator<T> >
class buffer_node : public graph_node, public reservable_item_buffer<T, A>, public receiver<T>, public sender<T> {
public:
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
    typedef buffer_node<T, A> my_class;
protected:
    typedef size_t size_type;
    internal::round_robin_cache< T, null_rw_mutex > my_successors;
        
    task *my_parent;
        
    friend class internal::forward_task< buffer_node< T, A > >;
        
    enum op_type {reg_succ, rem_succ, req_item, res_item, rel_res, con_res, put_item, try_fwd};
    enum op_stat {WAIT=0, SUCCEEDED, FAILED};
        
    // implements the aggregator_operation concept
    class buffer_operation : public internal::aggregated_operation< buffer_operation > {
    public:
        char type;
        T *elem;
        successor_type *r;
        buffer_operation(const T& e, op_type t) :
            type(char(t)), elem(const_cast<T*>(&e)), r(NULL) {}
        buffer_operation(op_type t) : type(char(t)), r(NULL) {}
    };
        
    bool forwarder_busy;
    typedef internal::aggregating_functor<my_class, buffer_operation> my_handler;
    friend class internal::aggregating_functor<my_class, buffer_operation>;
    internal::aggregator< my_handler, buffer_operation> my_aggregator;
        
    virtual void handle_operations(buffer_operation *op_list) {
        buffer_operation *tmp;
        bool try_forwarding=false;
        while (op_list) {
            tmp = op_list;
            op_list = op_list->next;
            switch (tmp->type) {
            case reg_succ: internal_reg_succ(tmp);  try_forwarding = true; break;
            case rem_succ: internal_rem_succ(tmp); break;
            case req_item: internal_pop(tmp); break;
            case res_item: internal_reserve(tmp); break;
            case rel_res:  internal_release(tmp);  try_forwarding = true; break;
            case con_res:  internal_consume(tmp);  try_forwarding = true; break;
            case put_item: internal_push(tmp);  try_forwarding = true; break;
            case try_fwd:  internal_forward(tmp); break;
            }
        }
        if (try_forwarding && !forwarder_busy) {
            forwarder_busy = true;
            task::enqueue(*new(task::allocate_additional_child_of(*my_parent)) internal::forward_task< buffer_node<input_type, A> >(*this));
        }
    }
        
    virtual void forward() {
        buffer_operation op_data(try_fwd);
        do {
            op_data.status = WAIT;
            my_aggregator.execute(&op_data);
        } while (op_data.status == SUCCEEDED);
    }
        
    virtual void internal_reg_succ(buffer_operation *op) {
        my_successors.register_successor(*(op->r));
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
    virtual void internal_rem_succ(buffer_operation *op) {
        my_successors.remove_successor(*(op->r));
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
    virtual void internal_forward(buffer_operation *op) {
        T i_copy;
        bool success = false; // flagged when a successor accepts
        size_type counter = my_successors.size();
        // Try forwarding, giving each successor a chance
        while (counter>0 && !this->buffer_empty() && this->item_valid(this->my_tail-1)) {
            this->fetch_back(i_copy);
            if( my_successors.try_put(i_copy) ) {
                this->invalidate_back();
                --(this->my_tail);
                success = true; // found an accepting successor
            }
            --counter;
        }
        if (success && !counter)
            __TBB_store_with_release(op->status, SUCCEEDED);
        else {
            __TBB_store_with_release(op->status, FAILED);
            forwarder_busy = false;
        }
    }
        
    virtual void internal_push(buffer_operation *op) {
        this->push_back(*(op->elem));
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
    virtual void internal_pop(buffer_operation *op) {
        if(this->pop_back(*(op->elem))) {
            __TBB_store_with_release(op->status, SUCCEEDED);
        }
        else {
            __TBB_store_with_release(op->status, FAILED);
        }
    }
        
    virtual void internal_reserve(buffer_operation *op) {
        if(this->reserve_front(*(op->elem))) {
            __TBB_store_with_release(op->status, SUCCEEDED);
        }
        else {
            __TBB_store_with_release(op->status, FAILED);
        }
    }
        
    virtual void internal_consume(buffer_operation *op) {
        this->consume_front();
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
    virtual void internal_release(buffer_operation *op) {
        this->release_front();
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
public:
    buffer_node( graph &g ) : reservable_item_buffer<T>(),
        my_parent( g.root_task() ), forwarder_busy(false) {
        my_successors.set_owner(this);
        my_aggregator.initialize_handler(my_handler(this));
    }

    buffer_node( const buffer_node& src ) :
        graph_node(), reservable_item_buffer<T>(), receiver<T>(), sender<T>(),
        my_parent( src.my_parent ) {
        forwarder_busy = false;
        my_successors.set_owner(this);
        my_aggregator.initialize_handler(my_handler(this));
    }

    virtual ~buffer_node() {}
        
    //
    // message sender implementation
    //
        

    /* override */ bool register_successor( receiver<output_type> &r ) {
        buffer_operation op_data(reg_succ);
        op_data.r = &r;
        my_aggregator.execute(&op_data);
        return true;
    }
        

    /* override */ bool remove_successor( receiver<output_type> &r ) {
        r.remove_predecessor(*this);
        buffer_operation op_data(rem_succ);
        op_data.r = &r;
        my_aggregator.execute(&op_data);
        return true;
    }
        

    /* override */ bool try_get( T &v ) {
        buffer_operation op_data(req_item);
        op_data.elem = &v;
        my_aggregator.execute(&op_data);
        return (op_data.status==SUCCEEDED);
    }
        

    /* override */ bool try_reserve( T &v ) {
        buffer_operation op_data(res_item);
        op_data.elem = &v;
        my_aggregator.execute(&op_data);
        return (op_data.status==SUCCEEDED);
    }
        

    /* override */ bool try_release() {
        buffer_operation op_data(rel_res);
        my_aggregator.execute(&op_data);
        return true;
    }
        

    /* override */ bool try_consume() {
        buffer_operation op_data(con_res);
        my_aggregator.execute(&op_data);
        return true;
    }
        

    /* override */ bool try_put(const T &t) {
        buffer_operation op_data(t, put_item);
        my_aggregator.execute(&op_data);
        return true;
    }
};
        
        
template <typename T, typename A=cache_aligned_allocator<T> >
class queue_node : public buffer_node<T, A> {
protected:
typedef typename buffer_node<T, A>::size_type size_type;
typedef typename buffer_node<T, A>::buffer_operation queue_operation;
        
    enum op_stat {WAIT=0, SUCCEEDED, FAILED};
        
    /* override */ void internal_forward(queue_operation *op) {
        T i_copy;
        bool success = false; // flagged when a successor accepts
        size_type counter = this->my_successors.size();
        if (this->my_reserved || !this->item_valid(this->my_head)){
            __TBB_store_with_release(op->status, FAILED);
            this->forwarder_busy = false;
            return;
        }
        // Keep trying to send items while there is at least one accepting successor
        while (counter>0 && this->item_valid(this->my_head)) {
            this->fetch_front(i_copy);
            if(this->my_successors.try_put(i_copy)) {
                 this->invalidate_front();
                 ++(this->my_head);
                success = true; // found an accepting successor
            }
            --counter;
        }
        if (success && !counter)
            __TBB_store_with_release(op->status, SUCCEEDED);
        else {
            __TBB_store_with_release(op->status, FAILED);
            this->forwarder_busy = false;
        }
    }
        
    /* override */ void internal_pop(queue_operation *op) {
        if ( this->my_reserved || !this->item_valid(this->my_head)){
            __TBB_store_with_release(op->status, FAILED);
        }
        else {
            this->pop_front(*(op->elem));
            __TBB_store_with_release(op->status, SUCCEEDED);
        }
    }
    /* override */ void internal_reserve(queue_operation *op) {
        if (this->my_reserved || !this->item_valid(this->my_head)) {
            __TBB_store_with_release(op->status, FAILED);
        }
        else {
            this->my_reserved = true;
            this->fetch_front(*(op->elem));
            this->invalidate_front();
            __TBB_store_with_release(op->status, SUCCEEDED);
        }
    }
    /* override */ void internal_consume(queue_operation *op) {
        this->consume_front();
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
        
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    queue_node( graph &g ) : buffer_node<T, A>(g) {}

    queue_node( const queue_node& src) : buffer_node<T, A>(src) {}
};
        
template< typename T, typename A=cache_aligned_allocator<T> >
class sequencer_node : public queue_node<T, A> {
    internal::function_body< T, size_t > *my_sequencer;
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    template< typename Sequencer >
    sequencer_node( graph &g, const Sequencer& s ) : queue_node<T, A>(g),
        my_sequencer(new internal::function_body_leaf< T, size_t, Sequencer>(s) ) {}

    sequencer_node( const sequencer_node& src ) : queue_node<T, A>(src),
        my_sequencer( src.my_sequencer->clone() ) {}
        
    ~sequencer_node() { delete my_sequencer; }
protected:
    typedef typename buffer_node<T, A>::size_type size_type;
    typedef typename buffer_node<T, A>::buffer_operation sequencer_operation;
        
    enum op_stat {WAIT=0, SUCCEEDED, FAILED};
        
private:
    /* override */ void internal_push(sequencer_operation *op) {
        size_type tag = (*my_sequencer)(*(op->elem));
        
        this->my_tail = (tag+1 > this->my_tail) ? tag+1 : this->my_tail;
        
        if(this->size() > this->capacity())
            this->grow_my_array(this->size());  // tail already has 1 added to it
        this->item(tag) = std::make_pair( *(op->elem), true );
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
};
        
template< typename T, typename Compare = std::less<T>, typename A=cache_aligned_allocator<T> >
class priority_queue_node : public buffer_node<T, A> {
public:
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
    priority_queue_node( graph &g ) : buffer_node<T, A>(g), mark(0) {}

    priority_queue_node( const priority_queue_node &src ) : buffer_node<T, A>(src), mark(0) {}
        
protected:
    typedef typename buffer_node<T, A>::size_type size_type;
    typedef typename buffer_node<T, A>::item_type item_type;
    typedef typename buffer_node<T, A>::buffer_operation prio_operation;
        
    enum op_stat {WAIT=0, SUCCEEDED, FAILED};
        
    /* override */ void handle_operations(prio_operation *op_list) {
        prio_operation *tmp /*, *pop_list*/ ;
        bool try_forwarding=false;
        while (op_list) {
            tmp = op_list;
            op_list = op_list->next;
            switch (tmp->type) {
            case buffer_node<T, A>::reg_succ: this->internal_reg_succ(tmp); try_forwarding = true; break;
            case buffer_node<T, A>::rem_succ: this->internal_rem_succ(tmp); break;
            case buffer_node<T, A>::put_item: internal_push(tmp); try_forwarding = true; break;
            case buffer_node<T, A>::try_fwd: internal_forward(tmp); break;
            case buffer_node<T, A>::rel_res: internal_release(tmp); try_forwarding = true; break;
            case buffer_node<T, A>::con_res: internal_consume(tmp); try_forwarding = true; break;
            case buffer_node<T, A>::req_item: internal_pop(tmp); break;
            case buffer_node<T, A>::res_item: internal_reserve(tmp); break;
            }
        }
        // process pops!  for now, no special pop processing
        if (mark<this->my_tail) heapify();
        if (try_forwarding && !this->forwarder_busy) {
            this->forwarder_busy = true;
            task::enqueue(*new(task::allocate_additional_child_of(*(this->my_parent))) internal::forward_task< buffer_node<input_type, A> >(*this));
        }
    }
        
    /* override */ void internal_forward(prio_operation *op) {
        T i_copy;
        bool success = false; // flagged when a successor accepts
        size_type counter = this->my_successors.size();
        
        if (this->my_reserved || this->my_tail == 0) {
            __TBB_store_with_release(op->status, FAILED);
            this->forwarder_busy = false;
            return;
        }
        // Keep trying to send while there exists an accepting successor
        while (counter>0 && this->my_tail > 0) {
            i_copy = this->my_array[0].first;
            bool msg = this->my_successors.try_put(i_copy);
            if ( msg == true ) {
                 if (mark == this->my_tail) --mark;
                --(this->my_tail);
                this->my_array[0].first=this->my_array[this->my_tail].first;
                if (this->my_tail > 1) // don't reheap for heap of size 1
                    reheap();
                success = true; // found an accepting successor
            }
            --counter;
        }
        if (success && !counter)
            __TBB_store_with_release(op->status, SUCCEEDED);
        else {
            __TBB_store_with_release(op->status, FAILED);
            this->forwarder_busy = false;
        }
    }
        
    /* override */ void internal_push(prio_operation *op) {
        if ( this->my_tail >= this->my_array_size )
            this->grow_my_array( this->my_tail + 1 );
        this->my_array[this->my_tail] = std::make_pair( *(op->elem), true );
        ++(this->my_tail);
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
    /* override */ void internal_pop(prio_operation *op) {
        if ( this->my_reserved == true || this->my_tail == 0 ) {
            __TBB_store_with_release(op->status, FAILED);
        }
        else {
            if (mark<this->my_tail &&
                compare(this->my_array[0].first,
                        this->my_array[this->my_tail-1].first)) {
                // there are newly pushed elems; last one higher than top
                // copy the data
                *(op->elem) = this->my_array[this->my_tail-1].first;
                --(this->my_tail);
                __TBB_store_with_release(op->status, SUCCEEDED);
            }
            else { // extract and push the last element down heap
                *(op->elem) = this->my_array[0].first; // copy the data
                if (mark == this->my_tail) --mark;
                --(this->my_tail);
                __TBB_store_with_release(op->status, SUCCEEDED);
                this->my_array[0].first=this->my_array[this->my_tail].first;
                if (this->my_tail > 1) // don't reheap for heap of size 1
                    reheap();
            }
        }
    }
    /* override */ void internal_reserve(prio_operation *op) {
        if (this->my_reserved == true || this->my_tail == 0) {
            __TBB_store_with_release(op->status, FAILED);
        }
        else {
            this->my_reserved = true;
            *(op->elem) = reserved_item = this->my_array[0].first;
            if (mark == this->my_tail) --mark;
            --(this->my_tail);
            __TBB_store_with_release(op->status, SUCCEEDED);
            this->my_array[0].first = this->my_array[this->my_tail].first;
            if (this->my_tail > 1) // don't reheap for heap of size 1
                reheap();
        }
    }
    /* override */ void internal_consume(prio_operation *op) {
        this->my_reserved = false;
        __TBB_store_with_release(op->status, SUCCEEDED);
    }
    /* override */ void internal_release(prio_operation *op) {
        if (this->my_tail >= this->my_array_size)
            this->grow_my_array( this->my_tail + 1 );
        this->my_array[this->my_tail] = std::make_pair(reserved_item, true);
        ++(this->my_tail);
        this->my_reserved = false;
        __TBB_store_with_release(op->status, SUCCEEDED);
        heapify();
    }
private:
    Compare compare;
    size_type mark;
    input_type reserved_item;
        
    void heapify() {
        if (!mark) mark = 1;
        for (; mark<this->my_tail; ++mark) { // for each unheaped element
            size_type cur_pos = mark;
            input_type to_place = this->my_array[mark].first;
            do { // push to_place up the heap
                size_type parent = (cur_pos-1)>>1;
                if (!compare(this->my_array[parent].first, to_place))
                    break;
                this->my_array[cur_pos].first = this->my_array[parent].first;
                cur_pos = parent;
            } while( cur_pos );
            this->my_array[cur_pos].first = to_place;
        }
    }
        
    void reheap() {
        size_type cur_pos=0, child=1;
        while (child < mark) {
            size_type target = child;
            if (child+1<mark &&
                compare(this->my_array[child].first,
                        this->my_array[child+1].first))
                ++target;
            // target now has the higher priority child
            if (compare(this->my_array[target].first,
                        this->my_array[this->my_tail].first))
                break;
            this->my_array[cur_pos].first = this->my_array[target].first;
            cur_pos = target;
            child = (cur_pos<<1)+1;
        }
        this->my_array[cur_pos].first = this->my_array[this->my_tail].first;
    }
};
        

template< typename T >
class limiter_node : public graph_node, public receiver< T >, public sender< T > {
public:
        
    typedef T input_type;
    typedef T output_type;
    typedef sender< input_type > predecessor_type;
    typedef receiver< output_type > successor_type;
        
private:
        
    task *my_root_task;
    size_t my_threshold;
    size_t my_count;
    internal::predecessor_cache< T > my_predecessors;
    spin_mutex my_mutex;
    internal::broadcast_cache< T > my_successors;
    int init_decrement_predecessors;

    friend class internal::forward_task< limiter_node<T> >;
        
    // Let decrementer call decrement_counter()
    friend class internal::decrementer< limiter_node<T> >;
        
    void decrement_counter() {
        input_type v;
        
        // If we can't get / put an item immediately then drop the count
        if ( my_predecessors.get_item( v ) == false 
             || my_successors.try_put(v) == false ) {
            spin_mutex::scoped_lock lock(my_mutex);
            --my_count;
            if ( !my_predecessors.empty() ) 
                task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
                            internal::forward_task< limiter_node<T> >( *this ) );
        }
    }
        
    void forward() {
        {
            spin_mutex::scoped_lock lock(my_mutex);
            if ( my_count < my_threshold ) 
                ++my_count;
            else
                return;
        }
        decrement_counter();
    }
        
public:
        
    internal::decrementer< limiter_node<T> > decrement;
        
    limiter_node(graph &g, size_t threshold, int num_decrement_predecessors=0) : 
        my_root_task(g.root_task()), my_threshold(threshold), my_count(0), 
        init_decrement_predecessors(num_decrement_predecessors), 
        decrement(num_decrement_predecessors) 
    {
        my_predecessors.set_owner(this);
        my_successors.set_owner(this);
        decrement.set_owner(this);
    }
        
    limiter_node( const limiter_node& src ) : 
        graph_node(), receiver<T>(), sender<T>(),
        my_root_task(src.my_root_task), my_threshold(src.my_threshold), my_count(0), 
        init_decrement_predecessors(src.init_decrement_predecessors), 
        decrement(src.init_decrement_predecessors) 
    {
        my_predecessors.set_owner(this);
        my_successors.set_owner(this);
        decrement.set_owner(this);
    }

    /* override */ bool register_successor( receiver<output_type> &r ) {
        my_successors.register_successor(r);
        return true;
    }
        

    /* override */ bool remove_successor( receiver<output_type> &r ) {
        r.remove_predecessor(*this);
        my_successors.remove_successor(r);
        return true;
    }
        
    /* override */ bool try_put( const T &t ) {
        {
            spin_mutex::scoped_lock lock(my_mutex);
            if ( my_count >= my_threshold ) 
                return false;
            else
                ++my_count; 
        }
        
        bool msg = my_successors.try_put(t);
        
        if ( msg != true ) {
            spin_mutex::scoped_lock lock(my_mutex);
            --my_count;
            if ( !my_predecessors.empty() ) 
                task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
                            internal::forward_task< limiter_node<T> >( *this ) );
        }
        
        return msg;
    }
        
    /* override */ bool register_predecessor( predecessor_type &src ) {
        spin_mutex::scoped_lock lock(my_mutex);
        my_predecessors.add( src );
        if ( my_count < my_threshold && !my_successors.empty() ) 
            task::enqueue( * new ( task::allocate_additional_child_of( *my_root_task ) ) 
                           internal::forward_task< limiter_node<T> >( *this ) );
        return true;
    }
        
    /* override */ bool remove_predecessor( predecessor_type &src ) {
        my_predecessors.remove( src );
        return true;
    }
        
};

#include "internal/_flow_graph_join_impl.h"

using internal::reserving_port;
using internal::queueing_port;
using internal::tag_matching_port;
using internal::input_port;
using internal::tag_value;
using internal::NO_TAG;

template<typename OutputTuple, graph_buffer_policy JP=queueing> class join_node;

template<typename OutputTuple>
class join_node<OutputTuple,reserving>: public internal::unfolded_join_node<std::tuple_size<OutputTuple>::value, reserving_port, OutputTuple, reserving> {
private:
    static const int N = std::tuple_size<OutputTuple>::value;
    typedef typename internal::unfolded_join_node<N, reserving_port, OutputTuple, reserving> unfolded_type;
public:
    typedef OutputTuple output_type;
    typedef typename unfolded_type::input_ports_tuple_type input_ports_tuple_type;
    join_node(graph &g) : unfolded_type(g) { }
    join_node(const join_node &other) : unfolded_type(other) {}
};

template<typename OutputTuple>
class join_node<OutputTuple,queueing>: public internal::unfolded_join_node<std::tuple_size<OutputTuple>::value, queueing_port, OutputTuple, queueing> {
private:
    static const int N = std::tuple_size<OutputTuple>::value;
    typedef typename internal::unfolded_join_node<N, queueing_port, OutputTuple, queueing> unfolded_type;
public:
    typedef OutputTuple output_type;
    typedef typename unfolded_type::input_ports_tuple_type input_ports_tuple_type;
    join_node(graph &g) : unfolded_type(g) { }
    join_node(const join_node &other) : unfolded_type(other) {}
};

// template for tag_matching join_node
template<typename OutputTuple>
class join_node<OutputTuple, tag_matching> : public internal::unfolded_join_node<std::tuple_size<OutputTuple>::value,
      tag_matching_port, OutputTuple, tag_matching> {
private:
    static const int N = std::tuple_size<OutputTuple>::value;
    typedef typename internal::unfolded_join_node<N, tag_matching_port, OutputTuple, tag_matching> unfolded_type;
public:
    typedef OutputTuple output_type;
    typedef typename unfolded_type::input_ports_tuple_type input_ports_tuple_type;
    template<typename B0, typename B1>
    join_node(graph &g, B0 b0, B1 b1) : unfolded_type(g, b0, b1) { }
    template<typename B0, typename B1, typename B2>
    join_node(graph &g, B0 b0, B1 b1, B2 b2) : unfolded_type(g, b0, b1, b2) { }
    template<typename B0, typename B1, typename B2, typename B3>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3) : unfolded_type(g, b0, b1, b2, b3) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4) : unfolded_type(g, b0, b1, b2, b3, b4) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4, typename B5>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4, B5 b5) : unfolded_type(g, b0, b1, b2, b3, b4, b5) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4, typename B5, typename B6>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4, B5 b5, B6 b6) : unfolded_type(g, b0, b1, b2, b3, b4, b5, b6) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4, typename B5, typename B6, typename B7>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4, B5 b5, B6 b6, B7 b7) : unfolded_type(g, b0, b1, b2, b3, b4, b5, b6, b7) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4, typename B5, typename B6, typename B7, typename B8>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4, B5 b5, B6 b6, B7 b7, B8 b8) : unfolded_type(g, b0, b1, b2, b3, b4, b5, b6, b7, b8) { }
    template<typename B0, typename B1, typename B2, typename B3, typename B4, typename B5, typename B6, typename B7, typename B8, typename B9>
    join_node(graph &g, B0 b0, B1 b1, B2 b2, B3 b3, B4 b4, B5 b5, B6 b6, B7 b7, B8 b8, B9 b9) : unfolded_type(g, b0, b1, b2, b3, b4, b5, b6, b7, b8, b9) { }
    join_node(const join_node &other) : unfolded_type(other) {}
};

#if TBB_PREVIEW_GRAPH_NODES
// or node
#include "internal/_flow_graph_or_impl.h"

template<typename InputTuple>
class or_node : public internal::unfolded_or_node<InputTuple> {
private:
    static const int N = std::tuple_size<InputTuple>::value;
public:
    typedef typename internal::or_output_type<InputTuple>::type output_type;
    typedef typename internal::unfolded_or_node<InputTuple> unfolded_type;
    or_node() : unfolded_type() { }
    // Copy constructor
    or_node( const or_node& /*other*/ ) : unfolded_type() { }
};
#endif  // TBB_PREVIEW_GRAPH_NODES

template< typename T >
inline void make_edge( sender<T> &p, receiver<T> &s ) {
    p.register_successor( s );
}
        
template< typename T >
inline void remove_edge( sender<T> &p, receiver<T> &s ) {
    p.remove_successor( s );
}

template< typename Body, typename Node >
Body copy_body( Node &n ) {
    return n.template copy_function_object<Body>();
}
        
        
} // interface6

    using interface6::graph;
    using interface6::graph_node;
    using interface6::continue_msg;
    using interface6::sender;
    using interface6::receiver;
    using interface6::continue_receiver;

    using interface6::source_node;
    using interface6::function_node;
#if TBB_PREVIEW_GRAPH_NODES
    using interface6::multioutput_function_node;
    using interface6::split_node;
    using interface6::internal::output_port;
    using interface6::or_node;
#endif
    using interface6::continue_node;
    using interface6::overwrite_node;
    using interface6::write_once_node;
    using interface6::broadcast_node;
    using interface6::buffer_node;
    using interface6::queue_node;
    using interface6::sequencer_node;
    using interface6::priority_queue_node;
    using interface6::limiter_node;
    using namespace interface6::internal::graph_policy_namespace;
    using interface6::join_node;
    using interface6::input_port;
    using interface6::copy_body; 
    using interface6::make_edge; 
    using interface6::remove_edge; 
    using interface6::internal::NO_TAG;
    using interface6::internal::tag_value;

} // flow
} // tbb

#endif // __TBB_flow_graph_H

---

Copyright © 2005-2011 Intel Corporation. All Rights Reserved.

Intel, Pentium, Intel Xeon, Itanium, Intel XScale and VTune are registered trademarks or trademarks of Intel Corporation or its subsidiaries in the United States and other countries.

* Other names and brands may be claimed as the property of others.