17 #ifndef __TBB__flow_graph_node_impl_H 18 #define __TBB__flow_graph_node_impl_H 20 #ifndef __TBB_flow_graph_H 21 #error Do not #include this internal file directly; use public TBB headers instead. 33 template<
typename T,
typename A >
60 template<
typename Input,
typename Policy,
typename A,
typename ImplType >
63 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 64 , add_blt_pred, del_blt_pred,
65 blt_pred_cnt, blt_pred_cpy
79 "queueing and rejecting policies can't be specified simultaneously");
81 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 82 typedef typename predecessor_cache_type::built_predecessors_type built_predecessors_type;
83 typedef typename receiver<input_type>::predecessor_list_type predecessor_list_type;
89 ) : my_graph_ref(g), my_max_concurrency(max_concurrency)
92 , forwarder_busy(false)
94 my_predecessors.set_owner(
this);
101 , my_graph_ref(src.my_graph_ref), my_max_concurrency(src.my_max_concurrency)
103 , my_queue(src.my_queue ? new input_queue_type() : NULL), forwarder_busy(false)
105 my_predecessors.set_owner(
this);
114 if ( my_queue )
delete my_queue;
123 operation_type op_data(reg_pred);
125 my_aggregator.execute(&op_data);
131 operation_type op_data(rem_pred);
133 my_aggregator.execute(&op_data);
137 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 138 void internal_add_built_predecessor( predecessor_type &src)
__TBB_override {
140 operation_type op_data(add_blt_pred);
142 my_aggregator.execute(&op_data);
146 void internal_delete_built_predecessor( predecessor_type &src)
__TBB_override {
147 operation_type op_data(del_blt_pred);
149 my_aggregator.execute(&op_data);
153 operation_type op_data(blt_pred_cnt);
154 my_aggregator.execute(&op_data);
155 return op_data.cnt_val;
159 operation_type op_data(blt_pred_cpy);
161 my_aggregator.execute(&op_data);
165 return my_predecessors.built_predecessors();
177 forwarder_busy =
false;
184 input_queue_type *my_queue;
190 my_predecessors.
reset();
191 __TBB_ASSERT(!(f & rf_clear_edges) || my_predecessors.
empty(),
"function_input_base reset failed");
199 operation_type op_data(i, app_body_bypass);
200 my_aggregator.execute(&op_data);
201 return op_data.bypass_t;
215 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 217 predecessor_list_type *predv;
222 type(char(t)), elem(const_cast<input_type*>(&e)) {}
232 task* new_task = NULL;
234 if(!my_queue->empty()) {
236 new_task = create_body_task(my_queue->front());
245 new_task = create_body_task(i);
254 op_list = op_list->next;
257 my_predecessors.
add(*(tmp->r));
259 if (!forwarder_busy) {
260 forwarder_busy =
true;
261 spawn_forward_task();
265 my_predecessors.
remove(*(tmp->r));
268 case app_body_bypass: {
269 tmp->bypass_t = NULL;
272 if(my_concurrency<my_max_concurrency)
273 tmp->bypass_t = perform_queued_requests();
278 case tryput_bypass: internal_try_put_task(tmp);
break;
279 case try_fwd: internal_forward(tmp);
break;
280 case occupy_concurrency:
281 if (my_concurrency < my_max_concurrency) {
288 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 290 my_predecessors.internal_add_built_predecessor(*(tmp->r));
295 my_predecessors.internal_delete_built_predecessor(*(tmp->r));
299 tmp->cnt_val = my_predecessors.predecessor_count();
303 my_predecessors.copy_predecessors( *(tmp->predv) );
314 if (my_concurrency < my_max_concurrency) {
316 task * new_task = create_body_task(*(op->elem));
317 op->bypass_t = new_task;
319 }
else if ( my_queue && my_queue->push(*(op->elem)) ) {
331 if (my_concurrency < my_max_concurrency || !my_max_concurrency)
332 op->bypass_t = perform_queued_requests();
336 forwarder_busy =
false;
342 operation_type op_data(t, tryput_bypass);
343 my_aggregator.
execute(&op_data);
345 return op_data.bypass_t;
351 if( my_max_concurrency == 0 ) {
352 return apply_body_bypass(t);
354 operation_type check_op(t, occupy_concurrency);
355 my_aggregator.
execute(&check_op);
357 return apply_body_bypass(t);
359 return internal_try_put_bypass(t);
364 if( my_max_concurrency == 0 ) {
365 return create_body_task(t);
367 return internal_try_put_bypass(t);
374 return static_cast<ImplType *
>(
this)->apply_body_impl_bypass(i);
380 new( task::allocate_additional_child_of(*(my_graph_ref.root_task())) )
388 operation_type op_data(try_fwd);
391 op_data.status =
WAIT;
392 my_aggregator.
execute(&op_data);
394 task* ttask = op_data.bypass_t;
404 new( task::allocate_additional_child_of(*(my_graph_ref.root_task())) )
411 task* tp = create_forward_task();
420 template<
typename Input,
typename Output,
typename Policy,
typename A>
431 template<
typename Body>
443 my_body( src.my_init_body->clone() ),
444 my_init_body(src.my_init_body->clone() ) {
452 template<
typename Body >
454 function_body_type &body_ref = *this->my_body;
462 output_type v = (*my_body)(i);
469 output_type v = apply_body_impl(i);
470 #if TBB_DEPRECATED_MESSAGE_FLOW_ORDER 471 task* successor_task = successors().try_put_task(v);
473 task* postponed_task = NULL;
474 if( base_type::my_max_concurrency != 0 ) {
475 postponed_task = base_type::try_get_postponed_task(i);
478 #if TBB_DEPRECATED_MESSAGE_FLOW_ORDER 479 graph& g = base_type::my_graph_ref;
482 if( postponed_task ) {
487 task* successor_task = successors().try_put_task(v);
488 #if _MSC_VER && !__INTEL_COMPILER 489 #pragma warning (push) 490 #pragma warning (disable: 4127) 493 #if _MSC_VER && !__INTEL_COMPILER 494 #pragma warning (pop) 496 if(!successor_task) {
502 return successor_task;
509 base_type::reset_function_input_base(f);
511 function_body_type *tmp = my_init_body->clone();
527 (
void)tbb::flow::get<N-1>(p).successors().clear();
531 if(tbb::flow::get<N-1>(p).successors().
empty())
539 (
void)tbb::flow::get<0>(p).successors().clear();
542 return tbb::flow::get<0>(
p).successors().empty();
546 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 548 template<
int N>
struct extract_element {
549 template<
typename P>
static void extract_this(P &
p) {
550 (
void)tbb::flow::get<N-1>(p).successors().built_successors().sender_extract(tbb::flow::get<N-1>(p));
551 extract_element<N-1>::extract_this(p);
555 template<>
struct extract_element<1> {
556 template<
typename P>
static void extract_this(P &
p) {
557 (
void)tbb::flow::get<0>(p).successors().built_successors().sender_extract(tbb::flow::get<0>(p));
562 template <
typename OutputTuple>
564 #if __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT 565 template <
typename... Args>
566 static OutputTuple
call(graph& g,
const tbb::flow::tuple<Args...>&) {
567 return OutputTuple(Args(g)...);
569 #else // __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT 570 template <
typename T1>
571 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1>&) {
572 return OutputTuple(T1(g));
575 template <
typename T1,
typename T2>
576 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1, T2>&) {
577 return OutputTuple(T1(g), T2(g));
580 template <
typename T1,
typename T2,
typename T3>
581 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1, T2, T3>&) {
582 return OutputTuple(T1(g), T2(g), T3(g));
585 template <
typename T1,
typename T2,
typename T3,
typename T4>
586 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4>&) {
587 return OutputTuple(T1(g), T2(g), T3(g), T4(g));
590 template <
typename T1,
typename T2,
typename T3,
typename T4,
typename T5>
591 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5>&) {
592 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g));
594 #if __TBB_VARIADIC_MAX >= 6 595 template <
typename T1,
typename T2,
typename T3,
typename T4,
typename T5,
typename T6>
596 static OutputTuple
call(graph& g,
const tbb::flow::tuple<T1, T2, T3, T4, T5, T6>&) {
597 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g));
600 #if __TBB_VARIADIC_MAX >= 7 601 template <
typename T1,
typename T2,
typename T3,
typename T4,
602 typename T5,
typename T6,
typename T7>
603 static OutputTuple
call(graph& g,
604 const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7>&) {
605 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g));
608 #if __TBB_VARIADIC_MAX >= 8 609 template <
typename T1,
typename T2,
typename T3,
typename T4,
610 typename T5,
typename T6,
typename T7,
typename T8>
611 static OutputTuple
call(graph& g,
612 const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8>&) {
613 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g));
616 #if __TBB_VARIADIC_MAX >= 9 617 template <
typename T1,
typename T2,
typename T3,
typename T4,
618 typename T5,
typename T6,
typename T7,
typename T8,
typename T9>
619 static OutputTuple
call(graph& g,
620 const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8, T9>&) {
621 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g), T9(g));
624 #if __TBB_VARIADIC_MAX >= 9 625 template <
typename T1,
typename T2,
typename T3,
typename T4,
typename T5,
626 typename T6,
typename T7,
typename T8,
typename T9,
typename T10>
627 static OutputTuple
call(graph& g,
628 const tbb::flow::tuple<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>&) {
629 return OutputTuple(T1(g), T2(g), T3(g), T4(g), T5(g), T6(g), T7(g), T8(g), T9(g), T10(g));
632 #endif // __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT 637 template<
typename Input,
typename OutputPortSet,
typename Policy,
typename A>
649 template<
typename Body>
661 my_body( src.my_init_body->clone() ),
662 my_init_body(src.my_init_body->clone() ),
671 template<
typename Body >
673 multifunction_body_type &body_ref = *this->my_body;
682 (*my_body)(i, my_output_ports);
685 if(base_type::my_max_concurrency != 0) {
686 ttask = base_type::try_get_postponed_task(i);
694 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 696 extract_element<N>::extract_this(my_output_ports);
701 base_type::reset_function_input_base(f);
704 multifunction_body_type *tmp = my_init_body->clone();
718 template<
size_t N,
typename MOP>
720 return tbb::flow::get<N>(op.output_ports());
732 template<
typename T,
typename P>
735 task* last_task = tbb::flow::get<N-1>(
p).try_put_task(tbb::flow::get<N-1>(t));
743 template<
typename T,
typename P>
745 task* last_task = tbb::flow::get<0>(
p).try_put_task(tbb::flow::get<0>(t));
752 template<
typename Output,
typename Policy>
764 template<
typename Body >
772 template<
typename Body >
782 my_graph_ref(src.my_graph_ref),
783 my_body( src.my_init_body->clone() ),
784 my_init_body( src.my_init_body->clone() ) {}
791 template<
typename Body >
793 function_body_type &body_ref = *my_body;
798 continue_receiver::reset_receiver(f);
800 function_body_type *tmp = my_init_body->clone();
821 output_type v = (*my_body)( continue_msg() );
823 return successors().try_put_task( v );
830 #if _MSC_VER && !__INTEL_COMPILER 831 #pragma warning (push) 832 #pragma warning (disable: 4127) 835 #if _MSC_VER && !__INTEL_COMPILER 836 #pragma warning (pop) 838 return apply_body_bypass( continue_msg() );
841 return new ( task::allocate_additional_child_of( *(my_graph_ref.root_task()) ) )
853 template<
typename Output >
861 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 862 typedef typename sender<output_type>::built_successors_type built_successors_type;
863 typedef typename sender<output_type>::successor_list_type successor_list_type;
868 my_successors.set_owner(
this);
873 successors().register_successor( r );
879 successors().remove_successor( r );
883 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 884 built_successors_type &built_successors()
__TBB_override {
return successors().built_successors(); }
887 void internal_add_built_successor( successor_type &r)
__TBB_override {
888 successors().internal_add_built_successor( r );
891 void internal_delete_built_successor( successor_type &r)
__TBB_override {
892 successors().internal_delete_built_successor( r );
896 return successors().successor_count();
900 successors().copy_successors(v);
913 return my_successors.try_put_task(i);
924 template<
typename Output >
929 using base_type::my_successors;
935 task *res = try_put_task(i);
936 if(!res)
return false;
943 using base_type::graph_reference;
948 return my_successors.try_put_task(i);
956 #if __TBB_FLOW_GRAPH_CPP11_FEATURES 957 template<
typename CompositeType>
960 template<
typename CompositeType,
typename NodeType1,
typename... NodeTypes >
961 void add_nodes_impl(CompositeType *c_node,
bool visible,
const NodeType1& n1,
const NodeTypes&... n) {
962 void *
addr =
const_cast<NodeType1 *
>(&n1);
971 #endif // __TBB__flow_graph_node_impl_H static void fgt_alias_port(void *, void *, bool)
const size_t my_max_concurrency
void internal_try_put_task(operation_type *op)
Put to the node, but return the task instead of enqueueing it.
void reset_receiver(reset_flags f) __TBB_override
task * try_put_task_impl(const input_type &t, tbb::internal::false_type)
Base class for types that should not be assigned.
output_ports_type my_output_ports
continue_input(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(Body &body, node_priority_t priority))
Input input_type
The input type of this receiver.
task * try_put_task_impl(const input_type &t, tbb::internal::true_type)
graph & graph_reference() const __TBB_override
task * apply_body_bypass(const input_type &i)
Applies the body to the provided input.
bool remove_successor(successor_type &r) __TBB_override
Removes a successor from this node.
function_input_queue< input_type, A > input_queue_type
function_body_type * my_body
task * create_forward_task()
the leaf for function_body
task * try_get_postponed_task(const input_type &i)
task * apply_body_impl_bypass(const input_type &i)
bool register_successor(successor_type &r) __TBB_override
Adds a new successor to this node.
bool push_back(item_type &v)
static tbb::task * combine_tasks(graph &g, tbb::task *left, tbb::task *right)
void __TBB_store_with_release(volatile T &location, V value)
virtual ~function_input_base()
Destructor.
task * try_put_task(const output_type &i)
void execute(operation_type *op)
bool is_graph_active(tbb::flow::interface10::graph &g)
task * apply_body_bypass(input_type)
Applies the body to the provided input.
graph & graph_reference() const __TBB_override
broadcast_cache_type & successors()
multifunction_input< Input, OutputPortSet, Policy, A > my_class
Body copy_function_object()
task * try_put_task(const input_type &t) __TBB_override
Put item to successor; return task to run the successor if possible.
multifunction_body< input_type, output_ports_type > multifunction_body_type
internal::aggregating_functor< class_type, operation_type > handler_type
OutputPortSet output_ports_type
bool buffer_empty() const
function_input(graph &g, size_t max_concurrency,)
static bool this_empty(P &p)
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_id __itt_string_handle __itt_metadata_type type
static task * emit_this(graph &g, const T &t, P &p)
function_input_base< Input, Policy, A, my_class > base_type
void reset(reset_flags f)
operation_type(const input_type &e, op_type t)
broadcast_cache_type my_successors
function_input_queue< input_type, A > input_queue_type
Body copy_function_object()
output_ports_type & output_ports()
function_body that takes an Input and a set of output ports
continue_msg input_type
The input type of this receiver.
const item_type & front() const
graph & graph_reference() const
function_input_queue< input_type, A > input_queue_type
predecessor_cache< input_type, null_mutex > my_predecessors
function_body_type * my_init_body
task * try_put_task(const output_type &i)
function_output(const function_output &other)
static tbb::task *const SUCCESSFULLY_ENQUEUED
void check_task_and_spawn(graph &g, task *t)
function_body< input_type, output_type > function_body_type
void call(F &&f, Pack &&p)
Calls the given function with arguments taken from a stored_pack.
tbb::flow::tuple_element< N, typename MOP::output_ports_type >::type & output_port(MOP &op)
#define __TBB_STATIC_ASSERT(condition, msg)
static void fgt_begin_body(void *)
void reset_receiver(reset_flags f) __TBB_override
function_output(graph &g)
function_input_base< Input, Policy, A, my_class > base_type
bool try_put(const output_type &i)
task * internal_try_put_bypass(const input_type &t)
A::template rebind< input_queue_type >::other queue_allocator_type
bool register_predecessor(predecessor_type &src) __TBB_override
Adds src to the list of cached predecessors.
#define __TBB_FLOW_GRAPH_PRIORITY_ARG1(arg1, priority)
void reset_function_input(reset_flags f)
void handle_operations(operation_type *op_list)
multifunction_body_type * my_body
static void clear_this(P &p)
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
aggregator< handler_type, operation_type > my_aggregator
function_body< input_type, output_type > function_body_type
multifunction_input(const multifunction_input &src)
Copy constructor.
static task * emit_this(graph &g, const T &t, P &p)
bool get_item(output_type &v)
untyped_sender predecessor_type
The predecessor type for this node.
void spawn_forward_task()
Spawns a task that calls forward()
multifunction_body_type * my_init_body
void internal_forward(operation_type *op)
Creates tasks for postponed messages if available and if concurrency allows.
continue_input< output_type, Policy > class_type
function_input< Input, Output, Policy, A > my_class
task * apply_body_impl_bypass(const input_type &i)
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
#define __TBB_FLOW_GRAPH_PRIORITY_EXPR(expr)
function_body_type * my_init_body
multifunction_input(graph &g, size_t max_concurrency,)
task * execute() __TBB_override
A task that calls a node's apply_body_bypass function, passing in an input of type Input...
A functor that takes an Input and generates an Output.
multifunction_output(graph &g)
task * perform_queued_requests()
void add_nodes_impl(CompositeType *, bool)
static OutputTuple call(graph &g, const tbb::flow::tuple< Args... > &)
operation_type(op_type t)
Implements methods for a function node that takes a type Input as input and sends.
leaf for multifunction. OutputSet can be a std::tuple or a vector.
Body copy_function_object()
unsigned int node_priority_t
function_input_base(const function_input_base &src)
Copy constructor.
Input and scheduling for a function node that takes a type Input as input.
receiver< input_type >::predecessor_type predecessor_type
int max_concurrency()
Returns the maximal number of threads that can work inside the arena.
sender< output_type >::successor_type successor_type
static void clear_this(P &p)
aggregated_operation base class
Implements methods for both executable and function nodes that puts Output to its successors...
function_input(const function_input &src)
Copy constructor.
task * create_body_task(const input_type &input)
allocates a task to apply a body
static bool this_empty(P &p)
continue_input(const continue_input &src)
function_input_base< Input, Policy, A, ImplType > class_type
A task that calls a node's forward_task function.
function_input_base(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(size_t max_concurrency, node_priority_t priority))
Constructor for function_input_base.
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
function_body_type * my_body
bool remove_predecessor(predecessor_type &src) __TBB_override
Removes src from the list of cached predecessors.
Output output_type
The output type of this receiver.
output_type apply_body_impl(const input_type &i)
multifunction_output(const multifunction_output &other)
void const char const char int ITT_FORMAT __itt_group_sync p
broadcast_cache< output_type > broadcast_cache_type
task * forward_task()
This is executed by an enqueued task, the "forwarder".
Implements methods for an executable node that takes continue_msg as input.
Base class for user-defined tasks.
static void fgt_end_body(void *)
void reset_function_input_base(reset_flags f)
continue_input(graph &g, int number_of_predecessors,)
bool pop_front(item_type &v)
predecessor_cache< input_type, null_mutex > predecessor_cache_type
Implements methods for a function node that takes a type Input as input.
function_output< output_type > base_type
void spawn_in_graph_arena(tbb::flow::interface10::graph &g, tbb::task &arena_task)
Spawns a task inside graph arena.