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);
100 : receiver<Input>(),
tbb::internal::no_assign()
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)) {}
227 typedef internal::aggregating_functor<class_type, operation_type>
handler_type;
228 friend class internal::aggregating_functor<class_type, operation_type>;
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>
437 , my_init_body( new internal::
function_body_leaf< input_type, output_type, Body>(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));
564 template<
typename Input,
typename OutputPortSet,
typename Policy,
typename A>
576 template<
typename Body>
587 my_body( src.my_init_body->clone() ),
588 my_init_body(src.my_init_body->clone() ) {
596 template<
typename Body >
598 multifunction_body_type &body_ref = *this->my_body;
607 (*my_body)(i, my_output_ports);
610 if(base_type::my_max_concurrency != 0) {
611 ttask = base_type::try_get_postponed_task(i);
619 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 621 extract_element<N>::extract_this(my_output_ports);
626 base_type::reset_function_input_base(f);
629 multifunction_body_type *tmp = my_init_body->clone();
643 template<
size_t N,
typename MOP>
645 return tbb::flow::get<N>(op.output_ports());
657 template<
typename T,
typename P>
660 task* last_task = tbb::flow::get<N-1>(
p).try_put_task(tbb::flow::get<N-1>(t));
668 template<
typename T,
typename P>
670 task* last_task = tbb::flow::get<0>(
p).try_put_task(tbb::flow::get<0>(t));
677 template<
typename Output,
typename Policy>
689 template<
typename Body >
694 , my_init_body( new internal::
function_body_leaf< input_type, output_type, Body>(body) )
697 template<
typename Body >
703 , my_init_body( new internal::
function_body_leaf< input_type, output_type, Body>(body) )
707 my_graph_ref(src.my_graph_ref),
708 my_body( src.my_init_body->clone() ),
709 my_init_body( src.my_init_body->clone() ) {}
716 template<
typename Body >
718 function_body_type &body_ref = *my_body;
723 continue_receiver::reset_receiver(f);
725 function_body_type *tmp = my_init_body->clone();
746 output_type v = (*my_body)( continue_msg() );
748 return successors().try_put_task( v );
755 #if _MSC_VER && !__INTEL_COMPILER 756 #pragma warning (push) 757 #pragma warning (disable: 4127) 760 #if _MSC_VER && !__INTEL_COMPILER 761 #pragma warning (pop) 763 return apply_body_bypass( continue_msg() );
766 return new ( task::allocate_additional_child_of( *(my_graph_ref.root_task()) ) )
778 template<
typename Output >
786 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 787 typedef typename sender<output_type>::built_successors_type built_successors_type;
788 typedef typename sender<output_type>::successor_list_type successor_list_type;
793 my_successors.set_owner(
this);
798 successors().register_successor( r );
804 successors().remove_successor( r );
808 #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION 809 built_successors_type &built_successors()
__TBB_override {
return successors().built_successors(); }
812 void internal_add_built_successor( successor_type &r)
__TBB_override {
813 successors().internal_add_built_successor( r );
816 void internal_delete_built_successor( successor_type &r)
__TBB_override {
817 successors().internal_delete_built_successor( r );
821 return successors().successor_count();
825 successors().copy_successors(v);
838 return my_successors.try_put_task(i);
847 template<
typename Output >
852 using base_type::my_successors;
858 task *res = try_put_task(i);
859 if(!res)
return false;
869 return my_successors.try_put_task(i);
877 #if __TBB_FLOW_GRAPH_CPP11_FEATURES 878 template<
typename CompositeType>
881 template<
typename CompositeType,
typename NodeType1,
typename... NodeTypes >
882 void add_nodes_impl(CompositeType *c_node,
bool visible,
const NodeType1& n1,
const NodeTypes&... n) {
883 void *
addr =
const_cast<NodeType1 *
>(&n1);
892 #endif // __TBB__flow_graph_node_impl_H function_input(const function_input &src)
Copy constructor.
task * execute() __TBB_override
leaf for multifunction. OutputSet can be a std::tuple or a vector.
continue_msg input_type
The input type of this receiver.
Body copy_function_object()
function_input_base< Input, Policy, A, ImplType > class_type
Body copy_function_object()
static task * emit_this(graph &g, const T &t, P &p)
function_input_base(const function_input_base &src)
Copy constructor.
task * try_put_task(const output_type &i)
output_type apply_body_impl(const input_type &i)
void spawn_forward_task()
Spawns a task that calls forward()
operation_type(op_type t)
unsigned int node_priority_t
static void fgt_end_body(void *)
broadcast_cache_type my_successors
task * create_forward_task()
task * internal_try_put_bypass(const input_type &t)
task * forward_task()
This is executed by an enqueued task, the "forwarder".
void add_nodes_impl(CompositeType *, bool)
predecessor_cache< input_type, null_mutex > predecessor_cache_type
const item_type & front() const
task * perform_queued_requests()
multifunction_body_type * my_body
continue_input(const continue_input &src)
const size_t my_max_concurrency
aggregator< handler_type, operation_type > my_aggregator
bool register_successor(successor_type &r) __TBB_override
Adds a new successor to this node.
receiver< input_type >::predecessor_type predecessor_type
static task * emit_this(graph &g, const T &t, P &p)
void handle_operations(operation_type *op_list)
predecessor_cache< input_type, null_mutex > my_predecessors
function_body_type * my_body
function_input(graph &g, size_t max_concurrency,)
static bool this_empty(P &p)
bool try_put(const output_type &i)
Implements methods for an executable node that takes continue_msg as input.
function_input< Input, Output, Policy, A > my_class
task * try_get_postponed_task(const input_type &i)
int max_concurrency()
Returns the maximal number of threads that can work inside the arena.
multifunction_body< input_type, output_ports_type > multifunction_body_type
#define __TBB_STATIC_ASSERT(condition, msg)
void spawn_in_graph_arena(graph &g, tbb::task &arena_task)
Spawns a task inside graph arena.
static tbb::task *const SUCCESSFULLY_ENQUEUED
task * apply_body_bypass(const input_type &i)
Applies the body to the provided input.
task * apply_body_impl_bypass(const input_type &i)
Body copy_function_object()
static void clear_this(P &p)
multifunction_input(const multifunction_input &src)
Copy constructor.
void reset_function_input_base(reset_flags f)
bool is_graph_active(graph &g)
function_output(const function_output &)
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
task * try_put_task(const output_type &i)
A task that calls a node's apply_body_bypass function, passing in an input of type Input...
bool get_item(output_type &v)
multifunction_body_type * my_init_body
void reset_function_input(reset_flags f)
function_input_base(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(size_t max_concurrency, node_priority_t priority))
Constructor for function_input_base.
A functor that takes an Input and generates an Output.
Implements methods for a function node that takes a type Input as input.
#define __TBB_FLOW_GRAPH_PRIORITY_ARG1(arg1, priority)
void internal_try_put_task(operation_type *op)
Put to the node, but return the task instead of enqueueing it.
task * apply_body_bypass(input_type)
Applies the body to the provided input.
Implements methods for a function node that takes a type Input as input and sends.
bool register_predecessor(predecessor_type &src) __TBB_override
Adds src to the list of cached predecessors.
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Input and scheduling for a function node that takes a type Input as input.
static void clear_this(P &p)
OutputPortSet output_ports_type
bool push_back(item_type &v)
task * try_put_task_impl(const input_type &t, tbb::internal::true_type)
task * create_body_task(const input_type &input)
allocates a task to apply a body
aggregated_operation base class
function_body that takes an Input and a set of output ports
function_input_queue< input_type, A > input_queue_type
broadcast_cache< output_type > broadcast_cache_type
A task that calls a node's forward_task function.
output_ports_type & output_ports()
void __TBB_store_with_release(volatile T &location, V value)
Output output_type
The output type of this receiver.
static void fgt_begin_body(void *)
void check_task_and_spawn(graph &g, task *t)
graph & graph_reference() __TBB_override
function_input_queue< input_type, A > input_queue_type
operation_type(const input_type &e, op_type t)
#define __TBB_FLOW_GRAPH_PRIORITY_EXPR(expr)
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
continue_input(graph &g, __TBB_FLOW_GRAPH_PRIORITY_ARG1(Body &body, node_priority_t priority))
multifunction_output(const multifunction_output &)
output_ports_type my_output_ports
task * try_put_task(const input_type &t) __TBB_override
Put item to successor; return task to run the successor if possible.
Input input_type
The input type of this receiver.
static tbb::task * combine_tasks(graph &g, tbb::task *left, tbb::task *right)
internal::aggregating_functor< class_type, operation_type > handler_type
Implements methods for both executable and function nodes that puts Output to its successors...
function_body_type * my_init_body
bool pop_front(item_type &v)
static bool this_empty(P &p)
Base class for user-defined tasks.
function_input_base< Input, Policy, A, my_class > base_type
graph & graph_reference() __TBB_override
function_input_base< Input, Policy, A, my_class > base_type
bool remove_predecessor(predecessor_type &src) __TBB_override
Removes src from the list of cached predecessors.
void reset_receiver(reset_flags f) __TBB_override
function_body_type * my_init_body
void reset_receiver(reset_flags f) __TBB_override
static void fgt_alias_port(void *, void *, bool)
task * apply_body_impl_bypass(const input_type &i)
void execute(operation_type *op)
Base class for types that should not be assigned.
function_output< output_type > base_type
void const char const char int ITT_FORMAT __itt_group_sync p
broadcast_cache_type & successors()
function_body< input_type, output_type > function_body_type
tbb::flow::tuple_element< N, typename MOP::output_ports_type >::type & output_port(MOP &op)
multifunction_input< Input, OutputPortSet, Policy, A > my_class
sender< output_type >::successor_type successor_type
the leaf for function_body
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
function_body< input_type, output_type > function_body_type
virtual ~function_input_base()
Destructor.
void reset(reset_flags f)
bool remove_successor(successor_type &r) __TBB_override
Removes a successor from this node.
void internal_forward(operation_type *op)
Creates tasks for postponed messages if available and if concurrency allows.
bool buffer_empty() const
continue_input< output_type, Policy > class_type
A::template rebind< input_queue_type >::other queue_allocator_type
continue_input(graph &g, int number_of_predecessors,)
task * try_put_task_impl(const input_type &t, tbb::internal::false_type)
function_input_queue< input_type, A > input_queue_type
function_body_type * my_body
multifunction_input(graph &g, size_t max_concurrency,)