_flow_graph_node_impl.h
Go to the documentation of this file.
/*
Copyright (c) 2005-2020 Intel Corporation
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef __TBB__flow_graph_node_impl_H
#define __TBB__flow_graph_node_impl_H
#ifndef __TBB_flow_graph_H
#error Do not #include this internal file directly; use public TBB headers instead.
#endif
#include "_flow_graph_item_buffer_impl.h"
namespace internal {
using tbb::internal::aggregated_operation;
using tbb::internal::aggregating_functor;
using tbb::internal::aggregator;
template< typename T, typename A >
class function_input_queue : public item_buffer<T,A> {
public:
bool empty() const {
return this->buffer_empty();
}
const T& front() const {
return this->item_buffer<T, A>::front();
}
bool pop( T& t ) {
return this->pop_front( t );
}
void pop() {
this->destroy_front();
}
bool push( T& t ) {
return this->push_back( t );
}
};
// The only up-ref is apply_body_impl, which should implement the function
// call and any handling of the result.
template< typename Input, typename Policy, typename A, typename ImplType >
class function_input_base : public receiver<Input>, tbb::internal::no_assign {
enum op_type {reg_pred, rem_pred, try_fwd, tryput_bypass, app_body_bypass, occupy_concurrency
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
, add_blt_pred, del_blt_pred,
blt_pred_cnt, blt_pred_cpy // create vector copies of preds and succs
#endif
};
typedef function_input_base<Input, Policy, A, ImplType> class_type;
public:
typedef Input input_type;
typedef typename receiver<input_type>::predecessor_type predecessor_type;
typedef predecessor_cache<input_type, null_mutex > predecessor_cache_type;
typedef function_input_queue<input_type, A> input_queue_type;
typedef typename tbb::internal::allocator_rebind<A, input_queue_type>::type queue_allocator_type;
__TBB_STATIC_ASSERT(!((internal::has_policy<queueing, Policy>::value) && (internal::has_policy<rejecting, Policy>::value)),
"queueing and rejecting policies can't be specified simultaneously");
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
typedef typename predecessor_cache_type::built_predecessors_type built_predecessors_type;
typedef typename receiver<input_type>::predecessor_list_type predecessor_list_type;
#endif
function_input_base(
graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(size_t max_concurrency, node_priority_t priority)
) : my_graph_ref(g), my_max_concurrency(max_concurrency)
, __TBB_FLOW_GRAPH_PRIORITY_ARG1(my_concurrency(0), my_priority(priority))
, my_queue(!internal::has_policy<rejecting, Policy>::value ? new input_queue_type() : NULL)
, forwarder_busy(false)
{
my_predecessors.set_owner(this);
my_aggregator.initialize_handler(handler_type(this));
}
function_input_base( const function_input_base& src)
: receiver<Input>(), tbb::internal::no_assign()
, my_graph_ref(src.my_graph_ref), my_max_concurrency(src.my_max_concurrency)
, __TBB_FLOW_GRAPH_PRIORITY_ARG1(my_concurrency(0), my_priority(src.my_priority))
, my_queue(src.my_queue ? new input_queue_type() : NULL), forwarder_busy(false)
{
my_predecessors.set_owner(this);
my_aggregator.initialize_handler(handler_type(this));
}
// The queue is allocated by the constructor for {multi}function_node.
// TODO: pass the graph_buffer_policy to the base so it can allocate the queue instead.
// This would be an interface-breaking change.
virtual ~function_input_base() {
if ( my_queue ) delete my_queue;
}
task* try_put_task( const input_type& t) __TBB_override {
return try_put_task_impl(t, internal::has_policy<lightweight, Policy>());
}
bool register_predecessor( predecessor_type &src ) __TBB_override {
operation_type op_data(reg_pred);
op_data.r = &src;
my_aggregator.execute(&op_data);
return true;
}
bool remove_predecessor( predecessor_type &src ) __TBB_override {
operation_type op_data(rem_pred);
op_data.r = &src;
my_aggregator.execute(&op_data);
return true;
}
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
void internal_add_built_predecessor( predecessor_type &src) __TBB_override {
operation_type op_data(add_blt_pred);
op_data.r = &src;
my_aggregator.execute(&op_data);
}
void internal_delete_built_predecessor( predecessor_type &src) __TBB_override {
operation_type op_data(del_blt_pred);
op_data.r = &src;
my_aggregator.execute(&op_data);
}
size_t predecessor_count() __TBB_override {
operation_type op_data(blt_pred_cnt);
my_aggregator.execute(&op_data);
return op_data.cnt_val;
}
void copy_predecessors(predecessor_list_type &v) __TBB_override {
operation_type op_data(blt_pred_cpy);
op_data.predv = &v;
my_aggregator.execute(&op_data);
}
built_predecessors_type &built_predecessors() __TBB_override {
return my_predecessors.built_predecessors();
}
#endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
protected:
void reset_function_input_base( reset_flags f) {
my_concurrency = 0;
if(my_queue) {
my_queue->reset();
}
reset_receiver(f);
forwarder_busy = false;
}
graph& my_graph_ref;
const size_t my_max_concurrency;
size_t my_concurrency;
__TBB_FLOW_GRAPH_PRIORITY_EXPR( node_priority_t my_priority; )
input_queue_type *my_queue;
predecessor_cache<input_type, null_mutex > my_predecessors;
void reset_receiver( reset_flags f) __TBB_override {
if( f & rf_clear_edges) my_predecessors.clear();
else
my_predecessors.reset();
__TBB_ASSERT(!(f & rf_clear_edges) || my_predecessors.empty(), "function_input_base reset failed");
}
graph& graph_reference() const __TBB_override {
return my_graph_ref;
}
task* try_get_postponed_task(const input_type& i) {
operation_type op_data(i, app_body_bypass); // tries to pop an item or get_item
my_aggregator.execute(&op_data);
return op_data.bypass_t;
}
private:
friend class apply_body_task_bypass< class_type, input_type >;
friend class forward_task_bypass< class_type >;
class operation_type : public aggregated_operation< operation_type > {
public:
char type;
union {
input_type *elem;
predecessor_type *r;
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
size_t cnt_val;
predecessor_list_type *predv;
#endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
};
tbb::task *bypass_t;
operation_type(const input_type& e, op_type t) :
type(char(t)), elem(const_cast<input_type*>(&e)) {}
operation_type(op_type t) : type(char(t)), r(NULL) {}
};
bool forwarder_busy;
typedef internal::aggregating_functor<class_type, operation_type> handler_type;
friend class internal::aggregating_functor<class_type, operation_type>;
aggregator< handler_type, operation_type > my_aggregator;
task* perform_queued_requests() {
task* new_task = NULL;
if(my_queue) {
if(!my_queue->empty()) {
++my_concurrency;
new_task = create_body_task(my_queue->front());
my_queue->pop();
}
}
else {
input_type i;
if(my_predecessors.get_item(i)) {
++my_concurrency;
new_task = create_body_task(i);
}
}
return new_task;
}
void handle_operations(operation_type *op_list) {
operation_type *tmp;
while (op_list) {
tmp = op_list;
op_list = op_list->next;
switch (tmp->type) {
case reg_pred:
my_predecessors.add(*(tmp->r));
__TBB_store_with_release(tmp->status, SUCCEEDED);
if (!forwarder_busy) {
forwarder_busy = true;
spawn_forward_task();
}
break;
case rem_pred:
my_predecessors.remove(*(tmp->r));
__TBB_store_with_release(tmp->status, SUCCEEDED);
break;
case app_body_bypass: {
tmp->bypass_t = NULL;
__TBB_ASSERT(my_max_concurrency != 0, NULL);
--my_concurrency;
if(my_concurrency<my_max_concurrency)
tmp->bypass_t = perform_queued_requests();
__TBB_store_with_release(tmp->status, SUCCEEDED);
}
break;
case tryput_bypass: internal_try_put_task(tmp); break;
case try_fwd: internal_forward(tmp); break;
case occupy_concurrency:
if (my_concurrency < my_max_concurrency) {
++my_concurrency;
__TBB_store_with_release(tmp->status, SUCCEEDED);
} else {
__TBB_store_with_release(tmp->status, FAILED);
}
break;
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
case add_blt_pred: {
my_predecessors.internal_add_built_predecessor(*(tmp->r));
__TBB_store_with_release(tmp->status, SUCCEEDED);
}
break;
case del_blt_pred:
my_predecessors.internal_delete_built_predecessor(*(tmp->r));
__TBB_store_with_release(tmp->status, SUCCEEDED);
break;
case blt_pred_cnt:
tmp->cnt_val = my_predecessors.predecessor_count();
__TBB_store_with_release(tmp->status, SUCCEEDED);
break;
case blt_pred_cpy:
my_predecessors.copy_predecessors( *(tmp->predv) );
__TBB_store_with_release(tmp->status, SUCCEEDED);
break;
#endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
}
}
}
void internal_try_put_task(operation_type *op) {
__TBB_ASSERT(my_max_concurrency != 0, NULL);
if (my_concurrency < my_max_concurrency) {
++my_concurrency;
task * new_task = create_body_task(*(op->elem));
op->bypass_t = new_task;
__TBB_store_with_release(op->status, SUCCEEDED);
} else if ( my_queue && my_queue->push(*(op->elem)) ) {
op->bypass_t = SUCCESSFULLY_ENQUEUED;
__TBB_store_with_release(op->status, SUCCEEDED);
} else {
op->bypass_t = NULL;
__TBB_store_with_release(op->status, FAILED);
}
}
void internal_forward(operation_type *op) {
op->bypass_t = NULL;
if (my_concurrency < my_max_concurrency || !my_max_concurrency)
op->bypass_t = perform_queued_requests();
if(op->bypass_t)
__TBB_store_with_release(op->status, SUCCEEDED);
else {
forwarder_busy = false;
__TBB_store_with_release(op->status, FAILED);
}
}
task* internal_try_put_bypass( const input_type& t ) {
operation_type op_data(t, tryput_bypass);
my_aggregator.execute(&op_data);
if( op_data.status == internal::SUCCEEDED ) {
return op_data.bypass_t;
}
return NULL;
}
task* try_put_task_impl( const input_type& t, /*lightweight=*/tbb::internal::true_type ) {
if( my_max_concurrency == 0 ) {
return apply_body_bypass(t);
} else {
operation_type check_op(t, occupy_concurrency);
my_aggregator.execute(&check_op);
if( check_op.status == internal::SUCCEEDED ) {
return apply_body_bypass(t);
}
return internal_try_put_bypass(t);
}
}
task* try_put_task_impl( const input_type& t, /*lightweight=*/tbb::internal::false_type ) {
if( my_max_concurrency == 0 ) {
return create_body_task(t);
} else {
return internal_try_put_bypass(t);
}
}
// then decides if more work is available
task * apply_body_bypass( const input_type &i ) {
return static_cast<ImplType *>(this)->apply_body_impl_bypass(i);
}
inline task * create_body_task( const input_type &input ) {
return (internal::is_graph_active(my_graph_ref)) ?
new( task::allocate_additional_child_of(*(my_graph_ref.root_task())) )
apply_body_task_bypass < class_type, input_type >(
*this, __TBB_FLOW_GRAPH_PRIORITY_ARG1(input, my_priority))
: NULL;
}
task* forward_task() {
operation_type op_data(try_fwd);
task* rval = NULL;
do {
op_data.status = WAIT;
my_aggregator.execute(&op_data);
if(op_data.status == SUCCEEDED) {
task* ttask = op_data.bypass_t;
__TBB_ASSERT( ttask && ttask != SUCCESSFULLY_ENQUEUED, NULL );
rval = combine_tasks(my_graph_ref, rval, ttask);
}
} while (op_data.status == SUCCEEDED);
return rval;
}
inline task *create_forward_task() {
return (internal::is_graph_active(my_graph_ref)) ?
new( task::allocate_additional_child_of(*(my_graph_ref.root_task())) )
forward_task_bypass< class_type >( __TBB_FLOW_GRAPH_PRIORITY_ARG1(*this, my_priority) )
: NULL;
}
inline void spawn_forward_task() {
task* tp = create_forward_task();
if(tp) {
internal::spawn_in_graph_arena(graph_reference(), *tp);
}
}
}; // function_input_base
// a type Output to its successors.
template< typename Input, typename Output, typename Policy, typename A>
class function_input : public function_input_base<Input, Policy, A, function_input<Input,Output,Policy,A> > {
public:
typedef Input input_type;
typedef Output output_type;
typedef function_body<input_type, output_type> function_body_type;
typedef function_input<Input, Output, Policy,A> my_class;
typedef function_input_base<Input, Policy, A, my_class> base_type;
typedef function_input_queue<input_type, A> input_queue_type;
// constructor
template<typename Body>
function_input(
graph &g, size_t max_concurrency,
__TBB_FLOW_GRAPH_PRIORITY_ARG1(Body& body, node_priority_t priority)
) : base_type(g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(max_concurrency, priority))
, my_body( new internal::function_body_leaf< input_type, output_type, Body>(body) )
, my_init_body( new internal::function_body_leaf< input_type, output_type, Body>(body) ) {
}
function_input( const function_input& src ) :
base_type(src),
my_body( src.my_init_body->clone() ),
my_init_body(src.my_init_body->clone() ) {
}
~function_input() {
delete my_body;
delete my_init_body;
}
template< typename Body >
Body copy_function_object() {
function_body_type &body_ref = *this->my_body;
return dynamic_cast< internal::function_body_leaf<input_type, output_type, Body> & >(body_ref).get_body();
}
output_type apply_body_impl( const input_type& i) {
// There is an extra copied needed to capture the
// body execution without the try_put
tbb::internal::fgt_begin_body( my_body );
output_type v = (*my_body)(i);
tbb::internal::fgt_end_body( my_body );
return v;
}
//TODO: consider moving into the base class
task * apply_body_impl_bypass( const input_type &i) {
output_type v = apply_body_impl(i);
#if TBB_DEPRECATED_MESSAGE_FLOW_ORDER
task* successor_task = successors().try_put_task(v);
#endif
task* postponed_task = NULL;
if( base_type::my_max_concurrency != 0 ) {
postponed_task = base_type::try_get_postponed_task(i);
__TBB_ASSERT( !postponed_task || postponed_task != SUCCESSFULLY_ENQUEUED, NULL );
}
#if TBB_DEPRECATED_MESSAGE_FLOW_ORDER
graph& g = base_type::my_graph_ref;
return combine_tasks(g, successor_task, postponed_task);
#else
if( postponed_task ) {
// make the task available for other workers since we do not know successors'
// execution policy
internal::spawn_in_graph_arena(base_type::graph_reference(), *postponed_task);
}
task* successor_task = successors().try_put_task(v);
#if _MSC_VER && !__INTEL_COMPILER
#pragma warning (push)
#pragma warning (disable: 4127) /* suppress conditional expression is constant */
#endif
if(internal::has_policy<lightweight, Policy>::value) {
#if _MSC_VER && !__INTEL_COMPILER
#pragma warning (pop)
#endif
if(!successor_task) {
// Return confirmative status since current
// node's body has been executed anyway
successor_task = SUCCESSFULLY_ENQUEUED;
}
}
return successor_task;
#endif /* TBB_DEPRECATED_MESSAGE_FLOW_ORDER */
}
protected:
void reset_function_input(reset_flags f) {
base_type::reset_function_input_base(f);
if(f & rf_reset_bodies) {
function_body_type *tmp = my_init_body->clone();
delete my_body;
my_body = tmp;
}
}
function_body_type *my_body;
function_body_type *my_init_body;
virtual broadcast_cache<output_type > &successors() = 0;
}; // function_input
// helper templates to clear the successor edges of the output ports of an multifunction_node
template<int N> struct clear_element {
template<typename P> static void clear_this(P &p) {
(void)tbb::flow::get<N-1>(p).successors().clear();
clear_element<N-1>::clear_this(p);
}
template<typename P> static bool this_empty(P &p) {
if(tbb::flow::get<N-1>(p).successors().empty())
return clear_element<N-1>::this_empty(p);
return false;
}
};
template<> struct clear_element<1> {
template<typename P> static void clear_this(P &p) {
(void)tbb::flow::get<0>(p).successors().clear();
}
template<typename P> static bool this_empty(P &p) {
return tbb::flow::get<0>(p).successors().empty();
}
};
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
// helper templates to extract the output ports of an multifunction_node from graph
template<int N> struct extract_element {
template<typename P> static void extract_this(P &p) {
(void)tbb::flow::get<N-1>(p).successors().built_successors().sender_extract(tbb::flow::get<N-1>(p));
extract_element<N-1>::extract_this(p);
}
};
template<> struct extract_element<1> {
template<typename P> static void extract_this(P &p) {
(void)tbb::flow::get<0>(p).successors().built_successors().sender_extract(tbb::flow::get<0>(p));
}
};
#endif
template <typename OutputTuple>
struct init_output_ports {
#if __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT
template <typename... Args>
static OutputTuple call(graph& g, const tbb::flow::tuple<Args...>&) {
return OutputTuple(Args(g)...);
}
#else // __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT
template <typename T1>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1>&) {
return OutputTuple(T1(g));
}
template <typename T1, typename T2>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1, T2>&) {
return OutputTuple(T1(g), T2(g));
}
template <typename T1, typename T2, typename T3>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1, T2, T3>&) {
return OutputTuple(T1(g), T2(g), T3(g));
}
template <typename T1, typename T2, typename T3, typename T4>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1, T2, T3, T4>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g));
}
template <typename T1, typename T2, typename T3, typename T4, typename T5>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1, T2, T3, T4, T5>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g));
}
#if __TBB_VARIADIC_MAX >= 6
template <typename T1, typename T2, typename T3, typename T4, typename T5, typename T6>
static OutputTuple call(graph& g, const tbb::flow::tuple<T1, T2, T3, T4, T5, T6>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g));
}
#endif
#if __TBB_VARIADIC_MAX >= 7
template <typename T1, typename T2, typename T3, typename T4,
typename T5, typename T6, typename T7>
static OutputTuple call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g));
}
#endif
#if __TBB_VARIADIC_MAX >= 8
template <typename T1, typename T2, typename T3, typename T4,
typename T5, typename T6, typename T7, typename T8>
static OutputTuple call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g));
}
#endif
#if __TBB_VARIADIC_MAX >= 9
template <typename T1, typename T2, typename T3, typename T4,
typename T5, typename T6, typename T7, typename T8, typename T9>
static OutputTuple call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8, T9>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g), T9(g));
}
#endif
#if __TBB_VARIADIC_MAX >= 9
template <typename T1, typename T2, typename T3, typename T4, typename T5,
typename T6, typename T7, typename T8, typename T9, typename T10>
static OutputTuple call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>&) {
return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g), T9(g), T10(g));
}
#endif
#endif // __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT
}; // struct init_output_ports
// and has a tuple of output ports specified.
template< typename Input, typename OutputPortSet, typename Policy, typename A>
class multifunction_input : public function_input_base<Input, Policy, A, multifunction_input<Input,OutputPortSet,Policy,A> > {
public:
static const int N = tbb::flow::tuple_size<OutputPortSet>::value;
typedef Input input_type;
typedef OutputPortSet output_ports_type;
typedef multifunction_body<input_type, output_ports_type> multifunction_body_type;
typedef multifunction_input<Input, OutputPortSet, Policy, A> my_class;
typedef function_input_base<Input, Policy, A, my_class> base_type;
typedef function_input_queue<input_type, A> input_queue_type;
// constructor
template<typename Body>
multifunction_input(graph &g, size_t max_concurrency,
__TBB_FLOW_GRAPH_PRIORITY_ARG1(Body& body, node_priority_t priority)
) : base_type(g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(max_concurrency, priority))
, my_body( new internal::multifunction_body_leaf<input_type, output_ports_type, Body>(body) )
, my_init_body( new internal::multifunction_body_leaf<input_type, output_ports_type, Body>(body) )
, my_output_ports(init_output_ports<output_ports_type>::call(g, my_output_ports)){
}
multifunction_input( const multifunction_input& src ) :
base_type(src),
my_body( src.my_init_body->clone() ),
my_init_body(src.my_init_body->clone() ),
my_output_ports( init_output_ports<output_ports_type>::call(src.my_graph_ref, my_output_ports) ) {
}
~multifunction_input() {
delete my_body;
delete my_init_body;
}
template< typename Body >
Body copy_function_object() {
multifunction_body_type &body_ref = *this->my_body;
return *static_cast<Body*>(dynamic_cast< internal::multifunction_body_leaf<input_type, output_ports_type, Body> & >(body_ref).get_body_ptr());
}
// for multifunction nodes we do not have a single successor as such. So we just tell
// the task we were successful.
//TODO: consider moving common parts with implementation in function_input into separate function
task * apply_body_impl_bypass( const input_type &i) {
tbb::internal::fgt_begin_body( my_body );
(*my_body)(i, my_output_ports);
tbb::internal::fgt_end_body( my_body );
task* ttask = NULL;
if(base_type::my_max_concurrency != 0) {
ttask = base_type::try_get_postponed_task(i);
}
return ttask ? ttask : SUCCESSFULLY_ENQUEUED;
}
output_ports_type &output_ports(){ return my_output_ports; }
protected:
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
void extract() {
extract_element<N>::extract_this(my_output_ports);
}
#endif
void reset(reset_flags f) {
base_type::reset_function_input_base(f);
if(f & rf_clear_edges)clear_element<N>::clear_this(my_output_ports);
if(f & rf_reset_bodies) {
multifunction_body_type *tmp = my_init_body->clone();
delete my_body;
my_body = tmp;
}
__TBB_ASSERT(!(f & rf_clear_edges) || clear_element<N>::this_empty(my_output_ports), "multifunction_node reset failed");
}
multifunction_body_type *my_body;
multifunction_body_type *my_init_body;
output_ports_type my_output_ports;
}; // multifunction_input
// template to refer to an output port of a multifunction_node
template<size_t N, typename MOP>
typename tbb::flow::tuple_element<N, typename MOP::output_ports_type>::type &output_port(MOP &op) {
return tbb::flow::get<N>(op.output_ports());
}
inline void check_task_and_spawn(graph& g, task* t) {
if (t && t != SUCCESSFULLY_ENQUEUED) {
internal::spawn_in_graph_arena(g, *t);
}
}
// helper structs for split_node
template<int N>
struct emit_element {
template<typename T, typename P>
static task* emit_this(graph& g, const T &t, P &p) {
// TODO: consider to collect all the tasks in task_list and spawn them all at once
task* last_task = tbb::flow::get<N-1>(p).try_put_task(tbb::flow::get<N-1>(t));
check_task_and_spawn(g, last_task);
return emit_element<N-1>::emit_this(g,t,p);
}
};
template<>
struct emit_element<1> {
template<typename T, typename P>
static task* emit_this(graph& g, const T &t, P &p) {
task* last_task = tbb::flow::get<0>(p).try_put_task(tbb::flow::get<0>(t));
check_task_and_spawn(g, last_task);
return SUCCESSFULLY_ENQUEUED;
}
};
template< typename Output, typename Policy>
class continue_input : public continue_receiver {
public:
typedef continue_msg input_type;
typedef Output output_type;
typedef function_body<input_type, output_type> function_body_type;
typedef continue_input<output_type, Policy> class_type;
template< typename Body >
continue_input( graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(Body& body, node_priority_t priority) )
: continue_receiver(__TBB_FLOW_GRAPH_PRIORITY_ARG1(/*number_of_predecessors=*/0, priority))
, my_graph_ref(g)
, my_body( new internal::function_body_leaf< input_type, output_type, Body>(body) )
, my_init_body( new internal::function_body_leaf< input_type, output_type, Body>(body) )
{ }
template< typename Body >
continue_input( graph &g, int number_of_predecessors,
__TBB_FLOW_GRAPH_PRIORITY_ARG1(Body& body, node_priority_t priority)
) : continue_receiver( __TBB_FLOW_GRAPH_PRIORITY_ARG1(number_of_predecessors, priority) )
, my_graph_ref(g)
, my_body( new internal::function_body_leaf< input_type, output_type, Body>(body) )
, my_init_body( new internal::function_body_leaf< input_type, output_type, Body>(body) )
{ }
continue_input( const continue_input& src ) : continue_receiver(src),
my_graph_ref(src.my_graph_ref),
my_body( src.my_init_body->clone() ),
my_init_body( src.my_init_body->clone() ) {}
~continue_input() {
delete my_body;
delete my_init_body;
}
template< typename Body >
Body copy_function_object() {
function_body_type &body_ref = *my_body;
return dynamic_cast< internal::function_body_leaf<input_type, output_type, Body> & >(body_ref).get_body();
}
void reset_receiver( reset_flags f) __TBB_override {
continue_receiver::reset_receiver(f);
if(f & rf_reset_bodies) {
function_body_type *tmp = my_init_body->clone();
delete my_body;
my_body = tmp;
}
}
protected:
graph& my_graph_ref;
function_body_type *my_body;
function_body_type *my_init_body;
virtual broadcast_cache<output_type > &successors() = 0;
friend class apply_body_task_bypass< class_type, continue_msg >;
task *apply_body_bypass( input_type ) {
// There is an extra copied needed to capture the
// body execution without the try_put
tbb::internal::fgt_begin_body( my_body );
output_type v = (*my_body)( continue_msg() );
tbb::internal::fgt_end_body( my_body );
return successors().try_put_task( v );
}
task* execute() __TBB_override {
if(!internal::is_graph_active(my_graph_ref)) {
return NULL;
}
#if _MSC_VER && !__INTEL_COMPILER
#pragma warning (push)
#pragma warning (disable: 4127) /* suppress conditional expression is constant */
#endif
if(internal::has_policy<lightweight, Policy>::value) {
#if _MSC_VER && !__INTEL_COMPILER
#pragma warning (pop)
#endif
return apply_body_bypass( continue_msg() );
}
else {
return new ( task::allocate_additional_child_of( *(my_graph_ref.root_task()) ) )
apply_body_task_bypass< class_type, continue_msg >(
*this, __TBB_FLOW_GRAPH_PRIORITY_ARG1(continue_msg(), my_priority) );
}
}
graph& graph_reference() const __TBB_override {
return my_graph_ref;
}
}; // continue_input
template< typename Output >
class function_output : public sender<Output> {
public:
template<int N> friend struct clear_element;
typedef Output output_type;
typedef typename sender<output_type>::successor_type successor_type;
typedef broadcast_cache<output_type> broadcast_cache_type;
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
typedef typename sender<output_type>::built_successors_type built_successors_type;
typedef typename sender<output_type>::successor_list_type successor_list_type;
#endif
function_output( graph& g) : my_graph_ref(g) { my_successors.set_owner(this); }
function_output(const function_output & other) : sender<output_type>(), my_graph_ref(other.my_graph_ref) {
my_successors.set_owner(this);
}
bool register_successor( successor_type &r ) __TBB_override {
successors().register_successor( r );
return true;
}
bool remove_successor( successor_type &r ) __TBB_override {
successors().remove_successor( r );
return true;
}
#if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
built_successors_type &built_successors() __TBB_override { return successors().built_successors(); }
void internal_add_built_successor( successor_type &r) __TBB_override {
successors().internal_add_built_successor( r );
}
void internal_delete_built_successor( successor_type &r) __TBB_override {
successors().internal_delete_built_successor( r );
}
size_t successor_count() __TBB_override {
return successors().successor_count();
}
void copy_successors( successor_list_type &v) __TBB_override {
successors().copy_successors(v);
}
#endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
// for multifunction_node. The function_body that implements
// the node will have an input and an output tuple of ports. To put
// an item to a successor, the body should
//
// get<I>(output_ports).try_put(output_value);
//
// if task pointer is returned will always spawn and return true, else
// return value will be bool returned from successors.try_put.
task *try_put_task(const output_type &i) { // not a virtual method in this class
return my_successors.try_put_task(i);
}
broadcast_cache_type &successors() { return my_successors; }
graph& graph_reference() const { return my_graph_ref; }
protected:
broadcast_cache_type my_successors;
graph& my_graph_ref;
}; // function_output
template< typename Output >
class multifunction_output : public function_output<Output> {
public:
typedef Output output_type;
typedef function_output<output_type> base_type;
using base_type::my_successors;
multifunction_output(graph& g) : base_type(g) {my_successors.set_owner(this);}
multifunction_output( const multifunction_output& other) : base_type(other.my_graph_ref) { my_successors.set_owner(this); }
bool try_put(const output_type &i) {
task *res = try_put_task(i);
if(!res) return false;
if(res != SUCCESSFULLY_ENQUEUED) {
FLOW_SPAWN(*res); // TODO: Spawn task inside arena
}
return true;
}
using base_type::graph_reference;
protected:
task* try_put_task(const output_type &i) {
return my_successors.try_put_task(i);
}
template <int N> friend struct emit_element;
}; // multifunction_output
//composite_node
#if __TBB_FLOW_GRAPH_CPP11_FEATURES
template<typename CompositeType>
void add_nodes_impl(CompositeType*, bool) {}
template< typename CompositeType, typename NodeType1, typename... NodeTypes >
void add_nodes_impl(CompositeType *c_node, bool visible, const NodeType1& n1, const NodeTypes&... n) {
void *addr = const_cast<NodeType1 *>(&n1);
fgt_alias_port(c_node, addr, visible);
add_nodes_impl(c_node, visible, n...);
}
#endif
} // internal
#endif // __TBB__flow_graph_node_impl_H
FLOW_SPAWN
#define FLOW_SPAWN(a)
Definition: flow_graph.h:65
__TBB_FLOW_GRAPH_PRIORITY_ARG1
#define __TBB_FLOW_GRAPH_PRIORITY_ARG1(arg1, priority)
Definition: _flow_graph_impl.h:40
__TBB_FLOW_GRAPH_PRIORITY_EXPR
#define __TBB_FLOW_GRAPH_PRIORITY_EXPR(expr)
Definition: _flow_graph_impl.h:38
_flow_graph_item_buffer_impl.h
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
__TBB_override
#define __TBB_override
Definition: tbb_stddef.h:240
addr
void * addr
Definition: ittnotify_static.h:91
task
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task * task
Definition: ittnotify_static.h:119
void
void
Definition: ittnotify_static.h:91
value
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task __itt_model_task_instance ITT_FORMAT p void ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long value
Definition: ittnotify_static.h:192
p
void const char const char int ITT_FORMAT __itt_group_sync p
Definition: ittnotify_static.h:91
type
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task __itt_model_task_instance ITT_FORMAT p void ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type type
Definition: ittnotify_static.h:198
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::fgt_begin_body
static void fgt_begin_body(void *)
Definition: _flow_graph_trace_impl.h:336
tbb::internal::call
void call(F &&f, Pack &&p)
Calls the given function with arguments taken from a stored_pack.
Definition: _template_helpers.h:175
tbb::internal::fgt_alias_port
static void fgt_alias_port(void *, void *, bool)
Definition: _flow_graph_trace_impl.h:299
tbb::internal::fgt_end_body
static void fgt_end_body(void *)
Definition: _flow_graph_trace_impl.h:337
tbb::internal::__TBB_store_with_release
void __TBB_store_with_release(volatile T &location, V value)
Definition: tbb_machine.h:716
tbb::flow::interface11::reset_flags
reset_flags
Definition: _flow_graph_impl.h:158
tbb::flow::interface11::rf_clear_edges
@ rf_clear_edges
Definition: _flow_graph_impl.h:161
tbb::flow::interface11::rf_reset_bodies
@ rf_reset_bodies
Definition: _flow_graph_impl.h:160
tbb::flow::interface11::combine_tasks
static tbb::task * combine_tasks(graph &g, tbb::task *left, tbb::task *right)
Definition: flow_graph.h:199
tbb::flow::interface11::internal::spawn_in_graph_arena
void spawn_in_graph_arena(tbb::flow::interface10::graph &g, tbb::task &arena_task)
Spawns a task inside graph arena.
Definition: _flow_graph_impl.h:521
tbb::flow::interface11::internal::is_graph_active
bool is_graph_active(tbb::flow::interface10::graph &g)
Definition: _flow_graph_impl.h:494
internal
Definition: _flow_graph_async_msg_impl.h:24
internal::output_port
tbb::flow::tuple_element< N, typename MOP::output_ports_type >::type & output_port(MOP &op)
Definition: _flow_graph_node_impl.h:719
internal::WAIT
@ WAIT
Definition: _flow_graph_types_impl.h:719
internal::FAILED
@ FAILED
Definition: _flow_graph_types_impl.h:719
internal::SUCCEEDED
@ SUCCEEDED
Definition: _flow_graph_types_impl.h:719
internal::add_nodes_impl
void add_nodes_impl(CompositeType *, bool)
Definition: _flow_graph_node_impl.h:958
internal::check_task_and_spawn
void check_task_and_spawn(graph &g, task *t)
Definition: _flow_graph_node_impl.h:723
tbb::flow::internal::node_priority_t
unsigned int node_priority_t
Definition: _flow_graph_impl.h:63
tbb::flow::internal::SUCCESSFULLY_ENQUEUED
static tbb::task *const SUCCESSFULLY_ENQUEUED
Definition: _flow_graph_impl.h:61
tbb::this_task_arena::max_concurrency
int max_concurrency()
Returns the maximal number of threads that can work inside the arena.
Definition: task_arena.h:490
tbb::flow::interface11::internal::untyped_receiver
Definition: flow_graph.h:358
tbb::flow::interface11::internal::untyped_receiver::predecessor_type
untyped_sender predecessor_type
The predecessor type for this node.
Definition: flow_graph.h:370
tbb::interface6::internal::aggregated_operation
aggregated_operation base class
Definition: _aggregator_impl.h:33
tbb::interface6::internal::aggregated_operation::next
Derived * next
Definition: _aggregator_impl.h:38
tbb::interface6::internal::aggregated_operation::status
uintptr_t status
Zero value means "wait" status, all other values are "user" specified values and are defined into the...
Definition: _aggregator_impl.h:36
tbb::interface6::internal::aggregator
Definition: _aggregator_impl.h:144
tbb::interface6::internal::aggregating_functor
Definition: _aggregator_impl.h:160
tbb::internal::allocator_rebind::type
allocator_traits< Alloc >::template rebind_alloc< T >::other type
Definition: _allocator_traits.h:149
internal::Policy
Definition: _flow_graph_body_impl.h:36
internal::has_policy
Definition: _flow_graph_body_impl.h:38
internal::graph_policy_namespace::rejecting
Definition: _flow_graph_body_impl.h:70
internal::function_body
A functor that takes an Input and generates an Output.
Definition: _flow_graph_body_impl.h:177
internal::function_body::clone
virtual function_body * clone()=0
internal::function_body_leaf
the leaf for function_body
Definition: _flow_graph_body_impl.h:186
internal::multifunction_body
function_body that takes an Input and a set of output ports
Definition: _flow_graph_body_impl.h:250
internal::multifunction_body::clone
virtual multifunction_body * clone()=0
internal::multifunction_body_leaf
leaf for multifunction. OutputSet can be a std::tuple or a vector.
Definition: _flow_graph_body_impl.h:260
internal::forward_task_bypass
A task that calls a node's forward_task function.
Definition: _flow_graph_body_impl.h:328
internal::apply_body_task_bypass
A task that calls a node's apply_body_bypass function, passing in an input of type Input.
Definition: _flow_graph_body_impl.h:353
internal::node_cache::clear
void clear()
Definition: _flow_graph_cache_impl.h:54
internal::node_cache::empty
bool empty()
Definition: _flow_graph_cache_impl.h:35
internal::node_cache::add
void add(T &n)
Definition: _flow_graph_cache_impl.h:40
internal::node_cache::remove
void remove(T &n)
Definition: _flow_graph_cache_impl.h:45
internal::predecessor_cache< input_type, null_mutex >
internal::predecessor_cache::get_item
bool get_item(output_type &v)
Definition: _flow_graph_cache_impl.h:143
internal::predecessor_cache::reset
void reset()
Definition: _flow_graph_cache_impl.h:173
internal::predecessor_cache::set_owner
void set_owner(successor_type *owner)
Definition: _flow_graph_cache_impl.h:141
internal::successor_cache::register_successor
void register_successor(successor_type &r)
Definition: _flow_graph_cache_impl.h:330
internal::successor_cache::set_owner
void set_owner(owner_type *owner)
Definition: _flow_graph_cache_impl.h:326
internal::successor_cache::remove_successor
void remove_successor(successor_type &r)
Definition: _flow_graph_cache_impl.h:335
internal::broadcast_cache< output_type >
internal::broadcast_cache::try_put_task
task * try_put_task(const T &t) __TBB_override
Definition: _flow_graph_cache_impl.h:478
internal::item_buffer
Definition: _flow_graph_item_buffer_impl.h:40
internal::item_buffer::front
const item_type & front() const
Definition: _flow_graph_item_buffer_impl.h:129
internal::item_buffer< T, A >::buffer_empty
bool buffer_empty() const
Definition: _flow_graph_item_buffer_impl.h:54
internal::item_buffer::reset
void reset()
Definition: _flow_graph_item_buffer_impl.h:239
internal::item_buffer< T, A >::pop_front
bool pop_front(item_type &v)
Definition: _flow_graph_item_buffer_impl.h:203
internal::item_buffer< T, A >::destroy_front
void destroy_front()
Definition: _flow_graph_item_buffer_impl.h:146
internal::item_buffer< T, A >::push_back
bool push_back(item_type &v)
Definition: _flow_graph_item_buffer_impl.h:185
internal::function_input_queue
Definition: _flow_graph_node_impl.h:34
internal::function_input_queue::empty
bool empty() const
Definition: _flow_graph_node_impl.h:36
internal::function_input_queue::pop
bool pop(T &t)
Definition: _flow_graph_node_impl.h:44
internal::function_input_queue::front
const T & front() const
Definition: _flow_graph_node_impl.h:40
internal::function_input_queue::pop
void pop()
Definition: _flow_graph_node_impl.h:48
internal::function_input_queue::push
bool push(T &t)
Definition: _flow_graph_node_impl.h:52
internal::function_input_base
Input and scheduling for a function node that takes a type Input as input.
Definition: _flow_graph_node_impl.h:61
internal::function_input_base::apply_body_bypass
task * apply_body_bypass(const input_type &i)
Applies the body to the provided input.
Definition: _flow_graph_node_impl.h:373
internal::function_input_base::forwarder_busy
bool forwarder_busy
Definition: _flow_graph_node_impl.h:226
internal::function_input_base::try_put_task_impl
task * try_put_task_impl(const input_type &t, tbb::internal::true_type)
Definition: _flow_graph_node_impl.h:350
internal::function_input_base::queue_allocator_type
tbb::internal::allocator_rebind< A, input_queue_type >::type queue_allocator_type
Definition: _flow_graph_node_impl.h:77
internal::function_input_base::op_type
op_type
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::reg_pred
@ reg_pred
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::try_fwd
@ try_fwd
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::app_body_bypass
@ app_body_bypass
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::occupy_concurrency
@ occupy_concurrency
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::tryput_bypass
@ tryput_bypass
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::rem_pred
@ rem_pred
Definition: _flow_graph_node_impl.h:62
internal::function_input_base::reset_receiver
void reset_receiver(reset_flags f) __TBB_override
Definition: _flow_graph_node_impl.h:187
internal::function_input_base::my_aggregator
aggregator< handler_type, operation_type > my_aggregator
Definition: _flow_graph_node_impl.h:229
internal::function_input_base::create_forward_task
task * create_forward_task()
Definition: _flow_graph_node_impl.h:402
internal::function_input_base::handler_type
internal::aggregating_functor< class_type, operation_type > handler_type
Definition: _flow_graph_node_impl.h:227
internal::function_input_base::my_max_concurrency
const size_t my_max_concurrency
Definition: _flow_graph_node_impl.h:181
internal::function_input_base::spawn_forward_task
void spawn_forward_task()
Spawns a task that calls forward()
Definition: _flow_graph_node_impl.h:410
internal::function_input_base::try_put_task
task * try_put_task(const input_type &t) __TBB_override
Definition: _flow_graph_node_impl.h:117
internal::function_input_base::input_queue_type
function_input_queue< input_type, A > input_queue_type
Definition: _flow_graph_node_impl.h:76
internal::function_input_base::internal_try_put_bypass
task * internal_try_put_bypass(const input_type &t)
Definition: _flow_graph_node_impl.h:341
internal::function_input_base::~function_input_base
virtual ~function_input_base()
Destructor.
Definition: _flow_graph_node_impl.h:113
internal::function_input_base::internal_try_put_task
void internal_try_put_task(operation_type *op)
Put to the node, but return the task instead of enqueueing it.
Definition: _flow_graph_node_impl.h:312
internal::function_input_base::input_type
Input input_type
The input type of this receiver.
Definition: _flow_graph_node_impl.h:73
internal::function_input_base::predecessor_cache_type
predecessor_cache< input_type, null_mutex > predecessor_cache_type
Definition: _flow_graph_node_impl.h:75
internal::function_input_base::forward_task
task * forward_task()
This is executed by an enqueued task, the "forwarder".
Definition: _flow_graph_node_impl.h:387
internal::function_input_base::create_body_task
task * create_body_task(const input_type &input)
allocates a task to apply a body
Definition: _flow_graph_node_impl.h:378
internal::function_input_base::perform_queued_requests
task * perform_queued_requests()
Definition: _flow_graph_node_impl.h:231
internal::function_input_base::my_graph_ref
graph & my_graph_ref
Definition: _flow_graph_node_impl.h:180
internal::function_input_base::my_predecessors
predecessor_cache< input_type, null_mutex > my_predecessors
Definition: _flow_graph_node_impl.h:185
internal::function_input_base::reset_function_input_base
void reset_function_input_base(reset_flags f)
Definition: _flow_graph_node_impl.h:171
internal::function_input_base::remove_predecessor
bool remove_predecessor(predecessor_type &src) __TBB_override
Removes src from the list of cached predecessors.
Definition: _flow_graph_node_impl.h:130
internal::function_input_base::my_concurrency
size_t my_concurrency
Definition: _flow_graph_node_impl.h:182
internal::function_input_base::handle_operations
void handle_operations(operation_type *op_list)
Definition: _flow_graph_node_impl.h:250
internal::function_input_base::function_input_base
function_input_base(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(size_t max_concurrency, node_priority_t priority))
Constructor for function_input_base.
Definition: _flow_graph_node_impl.h:87
internal::function_input_base::try_put_task_impl
task * try_put_task_impl(const input_type &t, tbb::internal::false_type)
Definition: _flow_graph_node_impl.h:363
internal::function_input_base::__TBB_STATIC_ASSERT
__TBB_STATIC_ASSERT(!((internal::has_policy< queueing, Policy >::value) &&(internal::has_policy< rejecting, Policy >::value)), "queueing and rejecting policies can't be specified simultaneously")
internal::function_input_base::graph_reference
graph & graph_reference() const __TBB_override
Definition: _flow_graph_node_impl.h:194
internal::function_input_base::function_input_base
function_input_base(const function_input_base &src)
Copy constructor.
Definition: _flow_graph_node_impl.h:99
internal::function_input_base::predecessor_type
receiver< input_type >::predecessor_type predecessor_type
Definition: _flow_graph_node_impl.h:74
internal::function_input_base::class_type
function_input_base< Input, Policy, A, ImplType > class_type
Definition: _flow_graph_node_impl.h:68
internal::function_input_base::register_predecessor
bool register_predecessor(predecessor_type &src) __TBB_override
Adds src to the list of cached predecessors.
Definition: _flow_graph_node_impl.h:122
internal::function_input_base::my_queue
input_queue_type * my_queue
Definition: _flow_graph_node_impl.h:184
internal::function_input_base::try_get_postponed_task
task * try_get_postponed_task(const input_type &i)
Definition: _flow_graph_node_impl.h:198
internal::function_input_base::internal_forward
void internal_forward(operation_type *op)
Creates tasks for postponed messages if available and if concurrency allows.
Definition: _flow_graph_node_impl.h:329
internal::function_input_base::operation_type
Definition: _flow_graph_node_impl.h:209
internal::function_input_base::operation_type::type
char type
Definition: _flow_graph_node_impl.h:211
internal::function_input_base::operation_type::elem
input_type * elem
Definition: _flow_graph_node_impl.h:213
internal::function_input_base::operation_type::r
predecessor_type * r
Definition: _flow_graph_node_impl.h:214
internal::function_input_base::operation_type::bypass_t
tbb::task * bypass_t
Definition: _flow_graph_node_impl.h:220
internal::function_input_base::operation_type::operation_type
operation_type(op_type t)
Definition: _flow_graph_node_impl.h:223
internal::function_input_base::operation_type::operation_type
operation_type(const input_type &e, op_type t)
Definition: _flow_graph_node_impl.h:221
internal::function_input
Implements methods for a function node that takes a type Input as input and sends.
Definition: _flow_graph_node_impl.h:421
internal::function_input::reset_function_input
void reset_function_input(reset_flags f)
Definition: _flow_graph_node_impl.h:508
internal::function_input::function_input
function_input(const function_input &src)
Copy constructor.
Definition: _flow_graph_node_impl.h:441
internal::function_input::my_class
function_input< Input, Output, Policy, A > my_class
Definition: _flow_graph_node_impl.h:426
internal::function_input::function_input
function_input(graph &g, size_t max_concurrency,)
Definition: _flow_graph_node_impl.h:432
internal::function_input::input_type
Input input_type
Definition: _flow_graph_node_impl.h:423
internal::function_input::copy_function_object
Body copy_function_object()
Definition: _flow_graph_node_impl.h:453
internal::function_input::base_type
function_input_base< Input, Policy, A, my_class > base_type
Definition: _flow_graph_node_impl.h:427
internal::function_input::function_body_type
function_body< input_type, output_type > function_body_type
Definition: _flow_graph_node_impl.h:425
internal::function_input::output_type
Output output_type
Definition: _flow_graph_node_impl.h:424
internal::function_input::successors
virtual broadcast_cache< output_type > & successors()=0
internal::function_input::input_queue_type
function_input_queue< input_type, A > input_queue_type
Definition: _flow_graph_node_impl.h:428
internal::function_input::my_body
function_body_type * my_body
Definition: _flow_graph_node_impl.h:517
internal::function_input::my_init_body
function_body_type * my_init_body
Definition: _flow_graph_node_impl.h:518
internal::function_input::~function_input
~function_input()
Definition: _flow_graph_node_impl.h:447
internal::function_input::apply_body_impl
output_type apply_body_impl(const input_type &i)
Definition: _flow_graph_node_impl.h:458
internal::function_input::apply_body_impl_bypass
task * apply_body_impl_bypass(const input_type &i)
Definition: _flow_graph_node_impl.h:468
internal::clear_element
Definition: _flow_graph_node_impl.h:525
internal::clear_element::clear_this
static void clear_this(P &p)
Definition: _flow_graph_node_impl.h:526
internal::clear_element::this_empty
static bool this_empty(P &p)
Definition: _flow_graph_node_impl.h:530
internal::clear_element< 1 >::this_empty
static bool this_empty(P &p)
Definition: _flow_graph_node_impl.h:541
internal::clear_element< 1 >::clear_this
static void clear_this(P &p)
Definition: _flow_graph_node_impl.h:538
internal::init_output_ports
Definition: _flow_graph_node_impl.h:563
internal::init_output_ports::call
static OutputTuple call(graph &g, const tbb::flow::tuple< Args... > &)
Definition: _flow_graph_node_impl.h:566
internal::multifunction_input
Implements methods for a function node that takes a type Input as input.
Definition: _flow_graph_node_impl.h:638
internal::multifunction_input::apply_body_impl_bypass
task * apply_body_impl_bypass(const input_type &i)
Definition: _flow_graph_node_impl.h:680
internal::multifunction_input::input_queue_type
function_input_queue< input_type, A > input_queue_type
Definition: _flow_graph_node_impl.h:646
internal::multifunction_input::output_ports_type
OutputPortSet output_ports_type
Definition: _flow_graph_node_impl.h:642
internal::multifunction_input::base_type
function_input_base< Input, Policy, A, my_class > base_type
Definition: _flow_graph_node_impl.h:645
internal::multifunction_input::multifunction_body_type
multifunction_body< input_type, output_ports_type > multifunction_body_type
Definition: _flow_graph_node_impl.h:643
internal::multifunction_input::my_init_body
multifunction_body_type * my_init_body
Definition: _flow_graph_node_impl.h:712
internal::multifunction_input::reset
void reset(reset_flags f)
Definition: _flow_graph_node_impl.h:700
internal::multifunction_input::N
static const int N
Definition: _flow_graph_node_impl.h:640
internal::multifunction_input::multifunction_input
multifunction_input(graph &g, size_t max_concurrency,)
Definition: _flow_graph_node_impl.h:650
internal::multifunction_input::~multifunction_input
~multifunction_input()
Definition: _flow_graph_node_impl.h:666
internal::multifunction_input::my_body
multifunction_body_type * my_body
Definition: _flow_graph_node_impl.h:711
internal::multifunction_input::my_class
multifunction_input< Input, OutputPortSet, Policy, A > my_class
Definition: _flow_graph_node_impl.h:644
internal::multifunction_input::input_type
Input input_type
Definition: _flow_graph_node_impl.h:641
internal::multifunction_input::my_output_ports
output_ports_type my_output_ports
Definition: _flow_graph_node_impl.h:713
internal::multifunction_input::copy_function_object
Body copy_function_object()
Definition: _flow_graph_node_impl.h:672
internal::multifunction_input::output_ports
output_ports_type & output_ports()
Definition: _flow_graph_node_impl.h:691
internal::multifunction_input::multifunction_input
multifunction_input(const multifunction_input &src)
Copy constructor.
Definition: _flow_graph_node_impl.h:659
internal::emit_element
Definition: _flow_graph_node_impl.h:731
internal::emit_element::emit_this
static task * emit_this(graph &g, const T &t, P &p)
Definition: _flow_graph_node_impl.h:733
internal::emit_element< 1 >::emit_this
static task * emit_this(graph &g, const T &t, P &p)
Definition: _flow_graph_node_impl.h:744
internal::continue_input
Implements methods for an executable node that takes continue_msg as input.
Definition: _flow_graph_node_impl.h:753
internal::continue_input::output_type
Output output_type
The output type of this receiver.
Definition: _flow_graph_node_impl.h:760
internal::continue_input::apply_body_bypass
task * apply_body_bypass(input_type)
Applies the body to the provided input.
Definition: _flow_graph_node_impl.h:817
internal::continue_input::reset_receiver
void reset_receiver(reset_flags f) __TBB_override
Definition: _flow_graph_node_impl.h:797
internal::continue_input::execute
task * execute() __TBB_override
Definition: _flow_graph_node_impl.h:826
internal::continue_input::my_graph_ref
graph & my_graph_ref
Definition: _flow_graph_node_impl.h:808
internal::continue_input::my_init_body
function_body_type * my_init_body
Definition: _flow_graph_node_impl.h:810
internal::continue_input::input_type
continue_msg input_type
The input type of this receiver.
Definition: _flow_graph_node_impl.h:757
internal::continue_input::successors
virtual broadcast_cache< output_type > & successors()=0
internal::continue_input::class_type
continue_input< output_type, Policy > class_type
Definition: _flow_graph_node_impl.h:762
internal::continue_input::copy_function_object
Body copy_function_object()
Definition: _flow_graph_node_impl.h:792
internal::continue_input::graph_reference
graph & graph_reference() const __TBB_override
Definition: _flow_graph_node_impl.h:847
internal::continue_input::function_body_type
function_body< input_type, output_type > function_body_type
Definition: _flow_graph_node_impl.h:761
internal::continue_input::continue_input
continue_input(const continue_input &src)
Definition: _flow_graph_node_impl.h:781
internal::continue_input::my_body
function_body_type * my_body
Definition: _flow_graph_node_impl.h:809
internal::continue_input::continue_input
continue_input(graph &g, int number_of_predecessors,)
Definition: _flow_graph_node_impl.h:773
internal::continue_input::continue_input
continue_input(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(Body &body, node_priority_t priority))
Definition: _flow_graph_node_impl.h:765
internal::continue_input::~continue_input
~continue_input()
Definition: _flow_graph_node_impl.h:786
internal::function_output
Implements methods for both executable and function nodes that puts Output to its successors.
Definition: _flow_graph_node_impl.h:854
internal::function_output::function_output
function_output(const function_output &other)
Definition: _flow_graph_node_impl.h:867
internal::function_output::output_type
Output output_type
Definition: _flow_graph_node_impl.h:858
internal::function_output::broadcast_cache_type
broadcast_cache< output_type > broadcast_cache_type
Definition: _flow_graph_node_impl.h:860
internal::function_output::my_graph_ref
graph & my_graph_ref
Definition: _flow_graph_node_impl.h:921
internal::function_output::remove_successor
bool remove_successor(successor_type &r) __TBB_override
Removes a successor from this node.
Definition: _flow_graph_node_impl.h:878
internal::function_output::graph_reference
graph & graph_reference() const
Definition: _flow_graph_node_impl.h:918
internal::function_output::my_successors
broadcast_cache_type my_successors
Definition: _flow_graph_node_impl.h:920
internal::function_output::successors
broadcast_cache_type & successors()
Definition: _flow_graph_node_impl.h:916
internal::function_output::try_put_task
task * try_put_task(const output_type &i)
Definition: _flow_graph_node_impl.h:912
internal::function_output::successor_type
sender< output_type >::successor_type successor_type
Definition: _flow_graph_node_impl.h:859
internal::function_output::function_output
function_output(graph &g)
Definition: _flow_graph_node_impl.h:866
internal::function_output::register_successor
bool register_successor(successor_type &r) __TBB_override
Adds a new successor to this node.
Definition: _flow_graph_node_impl.h:872
internal::multifunction_output
Definition: _flow_graph_node_impl.h:925
internal::multifunction_output::multifunction_output
multifunction_output(const multifunction_output &other)
Definition: _flow_graph_node_impl.h:932
internal::multifunction_output::multifunction_output
multifunction_output(graph &g)
Definition: _flow_graph_node_impl.h:931
internal::multifunction_output::try_put_task
task * try_put_task(const output_type &i)
Definition: _flow_graph_node_impl.h:947
internal::multifunction_output::my_successors
broadcast_cache_type my_successors
Definition: _flow_graph_node_impl.h:920
internal::multifunction_output::base_type
function_output< output_type > base_type
Definition: _flow_graph_node_impl.h:928
internal::multifunction_output::output_type
Output output_type
Definition: _flow_graph_node_impl.h:927
internal::multifunction_output::try_put
bool try_put(const output_type &i)
Definition: _flow_graph_node_impl.h:934
tbb::task
Base class for user-defined tasks.
Definition: task.h:615
tbb::internal::no_assign
Base class for types that should not be assigned.
Definition: tbb_stddef.h:322
tbb::internal::no_assign::no_assign
no_assign()=default
tbb::internal::bool_constant
Definition: tbb_stddef.h:486

Copyright © 2005-2020 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.