173 template<
typename StageTask>
188 spawner.spawn_stage_task(wakee);
191 #if __TBB_TASK_GROUP_CONTEXT 239 new_array[i].is_valid =
false;
241 for(
size_type i=0; i<old_size; ++i, ++t )
242 new_array[t&(
new_size-1)] = old_array[t&(old_size-1)];
282 #if __TBB_TASK_GROUP_CONTEXT 291 #endif // __TBB_TASK_GROUP_CONTEXT 318 goto process_another_stage;
369 goto process_another_stage;
379 if(ntokens_avail == 1) {
392 process_another_stage:
416 filter* first_suitable_filter = current_filter;
417 while( current_filter ) {
447 if( !current_filter ) {
452 current_filter = first_suitable_filter;
458 first_suitable_filter = first_suitable_filter->
next_segment;
459 current_filter = first_suitable_filter;
479 for(
filter* subfilter=
first->next_filter_in_pipeline;
481 subfilter=subfilter->next_filter_in_pipeline )
483 if( subfilter->prev_filter_in_pipeline->is_bound() && !subfilter->is_bound() ) {
486 head_of_previous_segment = subfilter;
493 #if _MSC_VER && !defined(__INTEL_COMPILER) 496 #pragma warning (disable: 4127) 507 #if __TBB_TASK_GROUP_CONTEXT 521 #if __TBB_TASK_GROUP_CONTEXT 525 if( internal::input_buffer* b = f->my_input_buffer )
536 has_thread_bound_filters(false)
549 if( internal::input_buffer* b = f->my_input_buffer ) {
557 f->my_pipeline = NULL;
560 f->next_segment = NULL;
585 *reinterpret_cast<filter**>(
filter_end) = &filter_;
587 *reinterpret_cast<filter**>(
filter_end) = NULL;
647 __TBB_ASSERT( max_number_of_live_tokens>0,
"pipeline::run must have at least one token" );
659 #if __TBB_TASK_GROUP_CONTEXT 670 f->my_input_buffer->sema_V();
677 #if __TBB_TASK_GROUP_CONTEXT 686 run(max_number_of_live_tokens, context);
689 #endif // __TBB_TASK_GROUP_CONTEXT 728 __TBB_ASSERT(
my_pipeline != NULL,
"It's not supposed that process_item is called for a filter that is not in a pipeline.");
729 internal::task_info info;
743 info.my_object = (*this)(info.my_object);
744 if( info.my_object ) {
749 info.my_token_ready =
true;
769 info.my_object = (*this)(info.my_object);
773 __TBB_ASSERT(
false,
"Couldn't put token after thread-bound buffer");
778 if( ntokens_avail == 1 ) {
bool is_ordered() const
True if filter must receive stream in order.
pipeline * my_pipeline
Pointer to the pipeline.
void __TBB_EXPORTED_METHOD inject_token(task &self)
Not used, but retained to satisfy old export files.
virtual void finalize(void *)
Destroys item if pipeline was cancelled.
#define __TBB_PIPELINE_VERSION(x)
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 sync_releasing
stage_task(pipeline &pipeline)
Construct stage_task for first stage in a pipeline.
static const size_type initial_buffer_size
Initial size for "array".
Token low_token
Lowest token that can start executing.
A processing pipeline that applies filters to items.
Represents acquisition of a mutex.
Token my_token
Invalid unless a task went through an ordered stage.
void recycle_as_continuation()
Change this to be a continuation of its former self.
Token high_token
Used for out of order buffer, and for assigning my_token if is_ordered and my_token not already assig...
void create_sema(size_t initial_tokens)
bool return_item(task_info &info, bool advance)
return an item, invalidate the queued item, but only advance if the filter
task * execute() __TBB_override
The virtual task execution method.
void set_my_tls_end_of_input()
Used to form groups of tasks.
This structure is used to store task information in a input buffer.
spin_mutex array_mutex
Serializes updates.
A stage in a pipeline served by a user thread.
void reset()
Roughly equivalent to the constructor of input stage task.
pipeline_root_task(pipeline &pipeline)
void reset()
Set to initial state (no object, no token)
bool object_may_be_null()
true if an input filter can emit null
bool my_token_ready
False until my_token is set.
Base class for types that should not be copied or assigned.
task * parent() const
task on whose behalf this task is working, or NULL if this is a root.
bool put_token(task_info &info_, bool force_put=false)
Put a token into the buffer.
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
filter * prev_filter_in_pipeline
Pointer to previous filter in the pipeline.
friend class internal::pipeline_root_task
Base class for user-defined tasks.
void __TBB_EXPORTED_METHOD set_end_of_input()
virtual __TBB_EXPORTED_METHOD ~pipeline()
bool is_bound() const
True if filter is thread-bound.
A lock that occupies a single byte.
#define ITT_NOTIFY(name, obj)
bool is_ordered
True for ordered filter, false otherwise.
static const unsigned char version_mask
Meets "allocator" requirements of ISO C++ Standard, Section 20.1.5.
bool my_at_start
True if this task has not yet read the input.
filter * next_segment
Pointer to the next "segment" of filters, or NULL if not required.
task * execute() __TBB_override
Should be overridden by derived classes.
task_info * array
Array of deferred tasks that cannot yet start executing.
void grow(size_type minimum_size)
Resize "array".
void note_done(Token token, StageTask &spawner)
Note that processing of a token is finished.
~input_buffer()
Destroy the buffer.
Edsger Dijkstra's counting semaphore.
stage_task(pipeline &pipeline, filter *filter_, const task_info &info)
Construct stage_task for a subsequent stage in a pipeline.
auto first(Container &c) -> decltype(begin(c))
pointer allocate(size_type n, const void *hint=0)
Allocate space for n objects, starting on a cache/sector line.
bool is_serial() const
True if filter is serial.
void deallocate(pointer p, size_type)
Free block of memory that starts on a cache line.
input_buffer(bool is_ordered_, bool is_bound_)
Construct empty buffer.
A buffer of input items for a filter.
bool is_bound
True for thread-bound filter, false otherwise.
internal::input_buffer * my_input_buffer
Buffer for incoming tokens, or NULL if not required.
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 new_size
void __TBB_EXPORTED_METHOD clear()
Remove all filters from the pipeline.
static filter * not_in_pipeline()
Value used to mark "not in pipeline".
void __TBB_EXPORTED_METHOD add_filter(filter &filter_)
Add filter to end of pipeline.
task * end_counter
task who's reference count is used to determine when all stages are done.
bool end_of_input
False until fetch_input returns NULL.
virtual __TBB_EXPORTED_METHOD ~filter()
Destroy filter.
void push_back(task &task)
Push task onto back of list.
const unsigned char my_filter_mode
Storage for filter mode and dynamically checked implementation version.
bool end_of_input_tls_allocated
bool has_more_work()
has the filter not yet processed all the tokens it will ever see?
static void spawn_root_and_wait(task &root)
Spawn task allocated by allocate_root, wait for it to complete, and deallocate it.
result_type __TBB_EXPORTED_METHOD try_process_item()
If a data item is available, invoke operator() on that item.
void __TBB_EXPORTED_METHOD run(size_t max_number_of_live_tokens)
Run the pipeline to completion.
end_of_input_tls_t end_of_input_tls
__TBB_EXPORTED_METHOD pipeline()
Construct empty pipeline.
bool my_tls_end_of_input()
bool is_valid
True if my_object is valid.
result_type internal_process_item(bool is_blocking)
Internal routine for item processing.
#define __TBB_TASK_GROUP_CONTEXT
void __TBB_EXPORTED_FUNC handle_perror(int error_code, const char *aux_info)
Throws std::runtime_error with what() returning error_code description prefixed with aux_info.
void poison_pointer(T *__TBB_atomic &)
bool has_item()
true if the current low_token is valid.
result_type __TBB_EXPORTED_METHOD process_item()
Wait until a data item becomes available, and invoke operator() on that item.
filter * filter_end
Pointer to location where address of next filter to be added should be stored.
bool has_thread_bound_filters
True if the pipeline contains a thread-bound filter; false otherwise.
void spawn_stage_task(const task_info &info)
Creates and spawns stage_task from task_info.
filter * next_filter_in_pipeline
Pointer to next filter in the pipeline.
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 * lock
internal::allocate_child_proxy & allocate_child()
Returns proxy for overloaded new that allocates a child task of *this.
semaphore * my_sem
for thread-bound filter, semaphore for waiting, NULL otherwise.
void set_ref_count(int count)
Set reference count.
basic_tls< intptr_t > end_of_input_tls_t
for parallel filters that accepts NULLs, thread-local flag for reaching end_of_input
bool is_cancelled() const
Returns true if the context has received cancellation request.
filter * filter_list
Pointer to first filter in the pipeline.
size_type array_size
Size of array.
void remove_filter(filter &filter_)
Remove filter from pipeline.
friend class internal::pipeline_cleaner
static const unsigned char exact_exception_propagation
7th bit defines exception propagation mode expected by the application.
void clear_filters()
Does clean up if pipeline is cancelled or exception occurred.
static internal::allocate_root_proxy allocate_root()
Returns proxy for overloaded new that allocates a root task.
atomic< internal::Token > token_counter
Global counter of tokens.
pipeline_cleaner(pipeline &_pipeline)
atomic< internal::Token > input_tokens
Number of idle tokens waiting for input stage.