pipeline.cpp
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.
*/
#include "tbb/pipeline.h"
#include "tbb/spin_mutex.h"
#include "tbb/cache_aligned_allocator.h"
#include "itt_notify.h"
#include "semaphore.h"
#include "tls.h" // for parallel filters that do not use NULL as end_of_input
namespace tbb {
namespace internal {
struct task_info {
void* my_object;
Token my_token;
bool my_token_ready;
bool is_valid;
void reset() {
my_object = NULL;
my_token = 0;
my_token_ready = false;
is_valid = false;
}
};
class input_buffer : no_copy {
friend class tbb::internal::pipeline_root_task;
friend class tbb::filter;
friend class tbb::thread_bound_filter;
friend class tbb::internal::stage_task;
friend class tbb::pipeline;
typedef Token size_type;
task_info* array;
semaphore* my_sem;
size_type array_size;
Token low_token;
spin_mutex array_mutex;
void grow( size_type minimum_size );
static const size_type initial_buffer_size = 4;
Token high_token;
bool is_ordered;
bool is_bound;
typedef basic_tls<intptr_t> end_of_input_tls_t;
end_of_input_tls_t end_of_input_tls;
bool end_of_input_tls_allocated; // no way to test pthread creation of TLS
void create_sema(size_t initial_tokens) { __TBB_ASSERT(!my_sem,NULL); my_sem = new internal::semaphore(initial_tokens); }
void free_sema() { __TBB_ASSERT(my_sem,NULL); delete my_sem; }
void sema_P() { __TBB_ASSERT(my_sem,NULL); my_sem->P(); }
void sema_V() { __TBB_ASSERT(my_sem,NULL); my_sem->V(); }
public:
input_buffer( bool is_ordered_, bool is_bound_ ) :
array(NULL), my_sem(NULL), array_size(0),
low_token(0), high_token(0),
is_ordered(is_ordered_), is_bound(is_bound_),
end_of_input_tls_allocated(false) {
grow(initial_buffer_size);
__TBB_ASSERT( array, NULL );
if(is_bound) create_sema(0);
}
~input_buffer() {
__TBB_ASSERT( array, NULL );
cache_aligned_allocator<task_info>().deallocate(array,array_size);
poison_pointer( array );
if(my_sem) {
free_sema();
}
if(end_of_input_tls_allocated) {
destroy_my_tls();
}
}
bool put_token( task_info& info_, bool force_put = false ) {
{
info_.is_valid = true;
spin_mutex::scoped_lock lock( array_mutex );
Token token;
bool was_empty = !array[low_token&(array_size-1)].is_valid;
if( is_ordered ) {
if( !info_.my_token_ready ) {
info_.my_token = high_token++;
info_.my_token_ready = true;
}
token = info_.my_token;
} else
token = high_token++;
__TBB_ASSERT( (tokendiff_t)(token-low_token)>=0, NULL );
if( token!=low_token || is_bound || force_put ) {
// Trying to put token that is beyond low_token.
// Need to wait until low_token catches up before dispatching.
if( token-low_token>=array_size )
grow( token-low_token+1 );
ITT_NOTIFY( sync_releasing, this );
array[token&(array_size-1)] = info_;
if(was_empty && is_bound) {
sema_V();
}
return true;
}
}
return false;
}
// Uses template to avoid explicit dependency on stage_task.
// This is only called for serial filters, and is the reason for the
// advance parameter in return_item (we're incrementing low_token here.)
// Non-TBF serial stages don't advance the token at the start because the presence
// of the current token in the buffer keeps another stage from being spawned.
template<typename StageTask>
void note_done( Token token, StageTask& spawner ) {
task_info wakee;
wakee.reset();
{
spin_mutex::scoped_lock lock( array_mutex );
if( !is_ordered || token==low_token ) {
// Wake the next task
task_info& item = array[++low_token & (array_size-1)];
ITT_NOTIFY( sync_acquired, this );
wakee = item;
item.is_valid = false;
}
}
if( wakee.is_valid )
spawner.spawn_stage_task(wakee);
}
#if __TBB_TASK_GROUP_CONTEXT
void clear( filter* my_filter ) {
long t=low_token;
for( size_type i=0; i<array_size; ++i, ++t ){
task_info& temp = array[t&(array_size-1)];
if (temp.is_valid ) {
my_filter->finalize(temp.my_object);
temp.is_valid = false;
}
}
}
#endif
// is parallel (as indicated by advance == true). If the filter is serial, leave the
// item in the buffer to keep another stage from being spawned.
bool return_item(task_info& info, bool advance) {
spin_mutex::scoped_lock lock( array_mutex );
task_info& item = array[low_token&(array_size-1)];
ITT_NOTIFY( sync_acquired, this );
if( item.is_valid ) {
info = item;
item.is_valid = false;
if (advance) low_token++;
return true;
}
return false;
}
bool has_item() { spin_mutex::scoped_lock lock(array_mutex); return array[low_token&(array_size -1)].is_valid; }
// end_of_input signal for parallel_pipeline, parallel input filters with 0 tokens allowed.
void create_my_tls() { int status = end_of_input_tls.create(); if(status) handle_perror(status, "TLS not allocated for filter"); end_of_input_tls_allocated = true; }
void destroy_my_tls() { int status = end_of_input_tls.destroy(); if(status) handle_perror(status, "Failed to destroy filter TLS"); }
bool my_tls_end_of_input() { return end_of_input_tls.get() != 0; }
void set_my_tls_end_of_input() { end_of_input_tls.set(1); }
};
void input_buffer::grow( size_type minimum_size ) {
size_type old_size = array_size;
size_type new_size = old_size ? 2*old_size : initial_buffer_size;
while( new_size<minimum_size )
new_size*=2;
task_info* new_array = cache_aligned_allocator<task_info>().allocate(new_size);
task_info* old_array = array;
for( size_type i=0; i<new_size; ++i )
new_array[i].is_valid = false;
long t=low_token;
for( size_type i=0; i<old_size; ++i, ++t )
new_array[t&(new_size-1)] = old_array[t&(old_size-1)];
array = new_array;
array_size = new_size;
if( old_array )
cache_aligned_allocator<task_info>().deallocate(old_array,old_size);
}
class stage_task: public task, public task_info {
private:
friend class tbb::pipeline;
pipeline& my_pipeline;
filter* my_filter;
bool my_at_start;
public:
stage_task( pipeline& pipeline ) :
my_pipeline(pipeline),
my_filter(pipeline.filter_list),
my_at_start(true)
{
task_info::reset();
}
stage_task( pipeline& pipeline, filter* filter_, const task_info& info ) :
task_info(info),
my_pipeline(pipeline),
my_filter(filter_),
my_at_start(false)
{}
void reset() {
task_info::reset();
my_filter = my_pipeline.filter_list;
my_at_start = true;
}
task* execute() __TBB_override;
#if __TBB_TASK_GROUP_CONTEXT
~stage_task()
{
if (my_filter && my_object && (my_filter->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(4)) {
__TBB_ASSERT(is_cancelled(), "Trying to finalize the task that wasn't cancelled");
my_filter->finalize(my_object);
my_object = NULL;
}
}
#endif // __TBB_TASK_GROUP_CONTEXT
void spawn_stage_task(const task_info& info)
{
stage_task* clone = new (allocate_additional_child_of(*parent()))
stage_task( my_pipeline, my_filter, info );
spawn(*clone);
}
};
task* stage_task::execute() {
__TBB_ASSERT( !my_at_start || !my_object, NULL );
__TBB_ASSERT( !my_filter->is_bound(), NULL );
if( my_at_start ) {
if( my_filter->is_serial() ) {
my_object = (*my_filter)(my_object);
if( my_object || ( my_filter->object_may_be_null() && !my_pipeline.end_of_input) )
{
if( my_filter->is_ordered() ) {
my_token = my_pipeline.token_counter++; // ideally, with relaxed semantics
my_token_ready = true;
} else if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
if( my_pipeline.has_thread_bound_filters )
my_pipeline.token_counter++; // ideally, with relaxed semantics
}
if( !my_filter->next_filter_in_pipeline ) { // we're only filter in pipeline
reset();
goto process_another_stage;
} else {
ITT_NOTIFY( sync_releasing, &my_pipeline.input_tokens );
if( --my_pipeline.input_tokens>0 )
spawn( *new( allocate_additional_child_of(*parent()) ) stage_task( my_pipeline ) );
}
} else {
my_pipeline.end_of_input = true;
return NULL;
}
} else /*not is_serial*/ {
if( my_pipeline.end_of_input )
return NULL;
if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
if( my_pipeline.has_thread_bound_filters )
my_pipeline.token_counter++;
}
ITT_NOTIFY( sync_releasing, &my_pipeline.input_tokens );
if( --my_pipeline.input_tokens>0 )
spawn( *new( allocate_additional_child_of(*parent()) ) stage_task( my_pipeline ) );
my_object = (*my_filter)(my_object);
if( !my_object && (!my_filter->object_may_be_null() || my_filter->my_input_buffer->my_tls_end_of_input()) )
{
my_pipeline.end_of_input = true;
if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
if( my_pipeline.has_thread_bound_filters )
my_pipeline.token_counter--; // fix token_counter
}
return NULL;
}
}
my_at_start = false;
} else {
my_object = (*my_filter)(my_object);
if( my_filter->is_serial() )
my_filter->my_input_buffer->note_done(my_token, *this);
}
my_filter = my_filter->next_filter_in_pipeline;
if( my_filter ) {
// There is another filter to execute.
if( my_filter->is_serial() ) {
// The next filter must execute tokens in order
if( my_filter->my_input_buffer->put_token(*this) ){
// Can't proceed with the same item
if( my_filter->is_bound() ) {
// Find the next non-thread-bound filter
do {
my_filter = my_filter->next_filter_in_pipeline;
} while( my_filter && my_filter->is_bound() );
// Check if there is an item ready to process
if( my_filter && my_filter->my_input_buffer->return_item(*this, !my_filter->is_serial()))
goto process_another_stage;
}
my_filter = NULL; // To prevent deleting my_object twice if exception occurs
return NULL;
}
}
} else {
// Reached end of the pipe.
size_t ntokens_avail = ++my_pipeline.input_tokens;
if(my_pipeline.filter_list->is_bound() ) {
if(ntokens_avail == 1) {
my_pipeline.filter_list->my_input_buffer->sema_V();
}
return NULL;
}
if( ntokens_avail>1 // Only recycle if there is one available token
|| my_pipeline.end_of_input ) {
return NULL; // No need to recycle for new input
}
ITT_NOTIFY( sync_acquired, &my_pipeline.input_tokens );
// Recycle as an input stage task.
reset();
}
process_another_stage:
/* A semi-hackish way to reexecute the same task object immediately without spawning.
recycle_as_continuation marks the task for future execution,
and then 'this' pointer is returned to bypass spawning. */
recycle_as_continuation();
return this;
}
class pipeline_root_task: public task {
pipeline& my_pipeline;
bool do_segment_scanning;
task* execute() __TBB_override {
if( !my_pipeline.end_of_input )
if( !my_pipeline.filter_list->is_bound() )
if( my_pipeline.input_tokens > 0 ) {
recycle_as_continuation();
set_ref_count(1);
return new( allocate_child() ) stage_task( my_pipeline );
}
if( do_segment_scanning ) {
filter* current_filter = my_pipeline.filter_list->next_segment;
/* first non-thread-bound filter that follows thread-bound one
and may have valid items to process */
filter* first_suitable_filter = current_filter;
while( current_filter ) {
__TBB_ASSERT( !current_filter->is_bound(), "filter is thread-bound?" );
__TBB_ASSERT( current_filter->prev_filter_in_pipeline->is_bound(), "previous filter is not thread-bound?" );
if( !my_pipeline.end_of_input || current_filter->has_more_work())
{
task_info info;
info.reset();
task* bypass = NULL;
int refcnt = 0;
task_list list;
// No new tokens are created; it's OK to process all waiting tokens.
// If the filter is serial, the second call to return_item will return false.
while( current_filter->my_input_buffer->return_item(info, !current_filter->is_serial()) ) {
task* t = new( allocate_child() ) stage_task( my_pipeline, current_filter, info );
if( ++refcnt == 1 )
bypass = t;
else // there's more than one task
list.push_back(*t);
// TODO: limit the list size (to arena size?) to spawn tasks sooner
__TBB_ASSERT( refcnt <= int(my_pipeline.token_counter), "token counting error" );
info.reset();
}
if( refcnt ) {
set_ref_count( refcnt );
if( refcnt > 1 )
spawn(list);
recycle_as_continuation();
return bypass;
}
current_filter = current_filter->next_segment;
if( !current_filter ) {
if( !my_pipeline.end_of_input ) {
recycle_as_continuation();
return this;
}
current_filter = first_suitable_filter;
__TBB_Yield();
}
} else {
/* The preceding pipeline segment is empty.
Fast-forward to the next post-TBF segment. */
first_suitable_filter = first_suitable_filter->next_segment;
current_filter = first_suitable_filter;
}
} /* while( current_filter ) */
return NULL;
} else {
if( !my_pipeline.end_of_input ) {
recycle_as_continuation();
return this;
}
return NULL;
}
}
public:
pipeline_root_task( pipeline& pipeline ): my_pipeline(pipeline), do_segment_scanning(false)
{
__TBB_ASSERT( my_pipeline.filter_list, NULL );
filter* first = my_pipeline.filter_list;
if( (first->my_filter_mode & first->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
// Scanning the pipeline for segments
filter* head_of_previous_segment = first;
for( filter* subfilter=first->next_filter_in_pipeline;
subfilter!=NULL;
subfilter=subfilter->next_filter_in_pipeline )
{
if( subfilter->prev_filter_in_pipeline->is_bound() && !subfilter->is_bound() ) {
do_segment_scanning = true;
head_of_previous_segment->next_segment = subfilter;
head_of_previous_segment = subfilter;
}
}
}
}
};
#if _MSC_VER && !defined(__INTEL_COMPILER)
// Workaround for overzealous compiler warnings
// Suppress compiler warning about constant conditional expression
#pragma warning (disable: 4127)
#endif
// The class destroys end_counter and clears all input buffers if pipeline was cancelled.
class pipeline_cleaner: internal::no_copy {
pipeline& my_pipeline;
public:
pipeline_cleaner(pipeline& _pipeline) :
my_pipeline(_pipeline)
{}
~pipeline_cleaner(){
#if __TBB_TASK_GROUP_CONTEXT
if (my_pipeline.end_counter->is_cancelled()) // Pipeline was cancelled
my_pipeline.clear_filters();
#endif
my_pipeline.end_counter = NULL;
}
};
} // namespace internal
void pipeline::inject_token( task& ) {
__TBB_ASSERT(false,"illegal call to inject_token");
}
#if __TBB_TASK_GROUP_CONTEXT
void pipeline::clear_filters() {
for( filter* f = filter_list; f; f = f->next_filter_in_pipeline ) {
if ((f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(4))
if( internal::input_buffer* b = f->my_input_buffer )
b->clear(f);
}
}
#endif
pipeline::pipeline() :
filter_list(NULL),
filter_end(NULL),
end_counter(NULL),
end_of_input(false),
has_thread_bound_filters(false)
{
token_counter = 0;
input_tokens = 0;
}
pipeline::~pipeline() {
clear();
}
void pipeline::clear() {
filter* next;
for( filter* f = filter_list; f; f=next ) {
if( internal::input_buffer* b = f->my_input_buffer ) {
delete b;
f->my_input_buffer = NULL;
}
next=f->next_filter_in_pipeline;
f->next_filter_in_pipeline = filter::not_in_pipeline();
if ( (f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
f->prev_filter_in_pipeline = filter::not_in_pipeline();
f->my_pipeline = NULL;
}
if ( (f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(5) )
f->next_segment = NULL;
}
filter_list = filter_end = NULL;
}
void pipeline::add_filter( filter& filter_ ) {
#if TBB_USE_ASSERT
if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) )
__TBB_ASSERT( filter_.prev_filter_in_pipeline==filter::not_in_pipeline(), "filter already part of pipeline?" );
__TBB_ASSERT( filter_.next_filter_in_pipeline==filter::not_in_pipeline(), "filter already part of pipeline?" );
__TBB_ASSERT( !end_counter, "invocation of add_filter on running pipeline" );
#endif
if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
filter_.my_pipeline = this;
filter_.prev_filter_in_pipeline = filter_end;
if ( filter_list == NULL)
filter_list = &filter_;
else
filter_end->next_filter_in_pipeline = &filter_;
filter_.next_filter_in_pipeline = NULL;
filter_end = &filter_;
} else {
if( !filter_end )
filter_end = reinterpret_cast<filter*>(&filter_list);
*reinterpret_cast<filter**>(filter_end) = &filter_;
filter_end = reinterpret_cast<filter*>(&filter_.next_filter_in_pipeline);
*reinterpret_cast<filter**>(filter_end) = NULL;
}
if( (filter_.my_filter_mode & filter_.version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
if( filter_.is_serial() ) {
if( filter_.is_bound() )
has_thread_bound_filters = true;
filter_.my_input_buffer = new internal::input_buffer( filter_.is_ordered(), filter_.is_bound() );
} else {
if(filter_.prev_filter_in_pipeline) {
if(filter_.prev_filter_in_pipeline->is_bound()) {
// successors to bound filters must have an input_buffer
filter_.my_input_buffer = new internal::input_buffer( /*is_ordered*/false, false );
}
} else { // input filter
if(filter_.object_may_be_null() ) {
//TODO: buffer only needed to hold TLS; could improve
filter_.my_input_buffer = new internal::input_buffer( /*is_ordered*/false, false );
filter_.my_input_buffer->create_my_tls();
}
}
}
} else {
if( filter_.is_serial() ) {
filter_.my_input_buffer = new internal::input_buffer( filter_.is_ordered(), false );
}
}
}
void pipeline::remove_filter( filter& filter_ ) {
__TBB_ASSERT( filter_.prev_filter_in_pipeline!=filter::not_in_pipeline(), "filter not part of pipeline" );
__TBB_ASSERT( filter_.next_filter_in_pipeline!=filter::not_in_pipeline(), "filter not part of pipeline" );
__TBB_ASSERT( !end_counter, "invocation of remove_filter on running pipeline" );
if (&filter_ == filter_list)
filter_list = filter_.next_filter_in_pipeline;
else {
__TBB_ASSERT( filter_.prev_filter_in_pipeline, "filter list broken?" );
filter_.prev_filter_in_pipeline->next_filter_in_pipeline = filter_.next_filter_in_pipeline;
}
if (&filter_ == filter_end)
filter_end = filter_.prev_filter_in_pipeline;
else {
__TBB_ASSERT( filter_.next_filter_in_pipeline, "filter list broken?" );
filter_.next_filter_in_pipeline->prev_filter_in_pipeline = filter_.prev_filter_in_pipeline;
}
if( internal::input_buffer* b = filter_.my_input_buffer ) {
delete b;
filter_.my_input_buffer = NULL;
}
filter_.next_filter_in_pipeline = filter_.prev_filter_in_pipeline = filter::not_in_pipeline();
if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(5) )
filter_.next_segment = NULL;
filter_.my_pipeline = NULL;
}
void pipeline::run( size_t max_number_of_live_tokens
#if __TBB_TASK_GROUP_CONTEXT
, tbb::task_group_context& context
#endif
) {
__TBB_ASSERT( max_number_of_live_tokens>0, "pipeline::run must have at least one token" );
__TBB_ASSERT( !end_counter, "pipeline already running?" );
if( filter_list ) {
internal::pipeline_cleaner my_pipeline_cleaner(*this);
end_of_input = false;
input_tokens = internal::Token(max_number_of_live_tokens);
if(has_thread_bound_filters) {
// release input filter if thread-bound
if(filter_list->is_bound()) {
filter_list->my_input_buffer->sema_V();
}
}
#if __TBB_TASK_GROUP_CONTEXT
end_counter = new( task::allocate_root(context) ) internal::pipeline_root_task( *this );
#else
end_counter = new( task::allocate_root() ) internal::pipeline_root_task( *this );
#endif
// Start execution of tasks
task::spawn_root_and_wait( *end_counter );
if(has_thread_bound_filters) {
for(filter* f = filter_list->next_filter_in_pipeline; f; f=f->next_filter_in_pipeline) {
if(f->is_bound()) {
f->my_input_buffer->sema_V(); // wake to end
}
}
}
}
}
#if __TBB_TASK_GROUP_CONTEXT
void pipeline::run( size_t max_number_of_live_tokens ) {
if( filter_list ) {
// Construct task group context with the exception propagation mode expected
// by the pipeline caller.
uintptr_t ctx_traits = filter_list->my_filter_mode & filter::exact_exception_propagation ?
task_group_context::default_traits :
task_group_context::default_traits & ~task_group_context::exact_exception;
task_group_context context(task_group_context::bound, ctx_traits);
run(max_number_of_live_tokens, context);
}
}
#endif // __TBB_TASK_GROUP_CONTEXT
bool filter::has_more_work() {
__TBB_ASSERT(my_pipeline, NULL);
__TBB_ASSERT(my_input_buffer, "has_more_work() called for filter with no input buffer");
return (internal::tokendiff_t)(my_pipeline->token_counter - my_input_buffer->low_token) != 0;
}
filter::~filter() {
if ( (my_filter_mode & version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
if ( next_filter_in_pipeline != filter::not_in_pipeline() )
my_pipeline->remove_filter(*this);
else
__TBB_ASSERT( prev_filter_in_pipeline == filter::not_in_pipeline(), "probably filter list is broken" );
} else {
__TBB_ASSERT( next_filter_in_pipeline==filter::not_in_pipeline(), "cannot destroy filter that is part of pipeline" );
}
}
void filter::set_end_of_input() {
__TBB_ASSERT(my_input_buffer, NULL);
__TBB_ASSERT(object_may_be_null(), NULL);
if(is_serial()) {
my_pipeline->end_of_input = true;
} else {
__TBB_ASSERT(my_input_buffer->end_of_input_tls_allocated, NULL);
my_input_buffer->set_my_tls_end_of_input();
}
}
thread_bound_filter::result_type thread_bound_filter::process_item() {
return internal_process_item(true);
}
thread_bound_filter::result_type thread_bound_filter::try_process_item() {
return internal_process_item(false);
}
thread_bound_filter::result_type thread_bound_filter::internal_process_item(bool is_blocking) {
__TBB_ASSERT(my_pipeline != NULL,"It's not supposed that process_item is called for a filter that is not in a pipeline.");
internal::task_info info;
info.reset();
if( my_pipeline->end_of_input && !has_more_work() )
return end_of_stream;
if( !prev_filter_in_pipeline ) {
if( my_pipeline->end_of_input )
return end_of_stream;
while( my_pipeline->input_tokens == 0 ) {
if( !is_blocking )
return item_not_available;
my_input_buffer->sema_P();
}
info.my_object = (*this)(info.my_object);
if( info.my_object ) {
__TBB_ASSERT(my_pipeline->input_tokens > 0, "Token failed in thread-bound filter");
my_pipeline->input_tokens--;
if( is_ordered() ) {
info.my_token = my_pipeline->token_counter;
info.my_token_ready = true;
}
my_pipeline->token_counter++; // ideally, with relaxed semantics
} else {
my_pipeline->end_of_input = true;
return end_of_stream;
}
} else { /* this is not an input filter */
while( !my_input_buffer->has_item() ) {
if( !is_blocking ) {
return item_not_available;
}
my_input_buffer->sema_P();
if( my_pipeline->end_of_input && !has_more_work() ) {
return end_of_stream;
}
}
if( !my_input_buffer->return_item(info, /*advance*/true) ) {
__TBB_ASSERT(false,"return_item failed");
}
info.my_object = (*this)(info.my_object);
}
if( next_filter_in_pipeline ) {
if ( !next_filter_in_pipeline->my_input_buffer->put_token(info,/*force_put=*/true) ) {
__TBB_ASSERT(false, "Couldn't put token after thread-bound buffer");
}
} else {
size_t ntokens_avail = ++(my_pipeline->input_tokens);
if( my_pipeline->filter_list->is_bound() ) {
if( ntokens_avail == 1 ) {
my_pipeline->filter_list->my_input_buffer->sema_V();
}
}
}
return success;
}
} // tbb
tbb::internal::basic_tls::get
T get()
Definition: tls.h:57
tbb::internal::basic_tls::destroy
int destroy()
Definition: tls.h:55
tbb::internal::input_buffer::input_buffer
input_buffer(bool is_ordered_, bool is_bound_)
Construct empty buffer.
Definition: pipeline.cpp:103
tbb::filter::is_ordered
bool is_ordered() const
True if filter must receive stream in order.
Definition: pipeline.h:133
tbb::internal::input_buffer::has_item
bool has_item()
true if the current low_token is valid.
Definition: pipeline.cpp:222
tbb::thread_bound_filter::result_type
result_type
Definition: pipeline.h:198
tbb::internal::input_buffer::my_sem
semaphore * my_sem
for thread-bound filter, semaphore for waiting, NULL otherwise.
Definition: pipeline.cpp:61
tbb::internal::basic_tls::create
int create()
Definition: tls.h:48
tbb::task::set_ref_count
void set_ref_count(int count)
Set reference count.
Definition: task.h:750
tbb::spin_mutex
A lock that occupies a single byte.
Definition: spin_mutex.h:39
spin_mutex.h
tbb::internal::input_buffer::size_type
Token size_type
Definition: pipeline.cpp:55
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::filter::not_in_pipeline
static filter * not_in_pipeline()
Value used to mark "not in pipeline".
Definition: pipeline.h:67
tbb::internal::pipeline_root_task::do_segment_scanning
bool do_segment_scanning
Definition: pipeline.cpp:402
tbb::filter::is_bound
bool is_bound() const
True if filter is thread-bound.
Definition: pipeline.h:138
tbb::internal::input_buffer::free_sema
void free_sema()
Definition: pipeline.cpp:97
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
tbb::task_group_context::bound
Definition: task.h:368
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::cache_aligned_allocator::allocate
pointer allocate(size_type n, const void *hint=0)
Allocate space for n objects, starting on a cache/sector line.
Definition: cache_aligned_allocator.h:80
tls.h
tbb::internal::task_info::reset
void reset()
Set to initial state (no object, no token)
Definition: pipeline.cpp:39
ITT_NOTIFY
#define ITT_NOTIFY(name, obj)
Definition: itt_notify.h:112
tbb::cache_aligned_allocator::deallocate
void deallocate(pointer p, size_type)
Free block of memory that starts on a cache line.
Definition: cache_aligned_allocator.h:86
tbb::task_list
A list of children.
Definition: task.h:1063
tbb::internal::stage_task::reset
void reset()
Roughly equivalent to the constructor of input stage task.
Definition: pipeline.cpp:275
tbb::task_list::push_back
void push_back(task &task)
Push task onto back of list.
Definition: task.h:1080
tbb::internal::pipeline_root_task
Definition: pipeline.cpp:400
tbb::internal::handle_perror
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.
Definition: tbb_misc.cpp:87
tbb::internal::stage_task::my_filter
filter * my_filter
Definition: pipeline.cpp:253
tbb::internal::poison_pointer
void poison_pointer(T *__TBB_atomic &)
Definition: tbb_stddef.h:305
__TBB_TASK_GROUP_CONTEXT
#define __TBB_TASK_GROUP_CONTEXT
Definition: tbb_config.h:542
tbb::internal::task_info::my_object
void * my_object
Definition: pipeline.cpp:31
tbb::internal::stage_task::spawn_stage_task
void spawn_stage_task(const task_info &info)
Creates and spawns stage_task from task_info.
Definition: pipeline.cpp:293
tbb::internal::input_buffer::end_of_input_tls_t
basic_tls< intptr_t > end_of_input_tls_t
for parallel filters that accepts NULLs, thread-local flag for reaching end_of_input
Definition: pipeline.cpp:92
tbb::internal::input_buffer::note_done
void note_done(Token token, StageTask &spawner)
Note that processing of a token is finished.
Definition: pipeline.cpp:174
tbb::task::allocate_child
internal::allocate_child_proxy & allocate_child()
Returns proxy for overloaded new that allocates a child task of *this.
Definition: task.h:670
cache_aligned_allocator.h
tbb::internal::stage_task::pipeline
friend class tbb::pipeline
Definition: pipeline.cpp:251
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::input_buffer::high_token
Token high_token
Used for out of order buffer, and for assigning my_token if is_ordered and my_token not already assig...
Definition: pipeline.cpp:83
tbb::filter::is_serial
bool is_serial() const
True if filter is serial.
Definition: pipeline.h:128
tbb::internal::pipeline_root_task::pipeline_root_task
pipeline_root_task(pipeline &pipeline)
Definition: pipeline.cpp:472
tbb::internal::spawner
Definition: parallel_invoke.h:57
__TBB_PIPELINE_VERSION
#define __TBB_PIPELINE_VERSION(x)
Definition: pipeline.h:41
tbb::internal::semaphore::P
void P()
wait/acquire
Definition: semaphore.h:105
tbb::task_group_context
Used to form groups of tasks.
Definition: task.h:347
tbb::internal::input_buffer
A buffer of input items for a filter.
Definition: pipeline.cpp:48
tbb::internal::Token
unsigned long Token
Definition: pipeline.h:43
tbb::cache_aligned_allocator
Meets "allocator" requirements of ISO C++ Standard, Section 20.1.5.
Definition: cache_aligned_allocator.h:60
tbb::task_group_context::default_traits
Definition: task.h:380
tbb::internal::input_buffer::pipeline
friend class tbb::pipeline
Definition: pipeline.cpp:53
tbb::internal::stage_task::my_at_start
bool my_at_start
True if this task has not yet read the input.
Definition: pipeline.cpp:255
tbb::filter::version_mask
static const unsigned char version_mask
Definition: pipeline.h:92
tbb::filter::set_end_of_input
void __TBB_EXPORTED_METHOD set_end_of_input()
Definition: pipeline.cpp:708
tbb::internal::stage_task::my_pipeline
pipeline & my_pipeline
Definition: pipeline.cpp:252
tbb::internal::input_buffer::create_sema
void create_sema(size_t initial_tokens)
Definition: pipeline.cpp:96
tbb::internal::stage_task::stage_task
stage_task(pipeline &pipeline, filter *filter_, const task_info &info)
Construct stage_task for a subsequent stage in a pipeline.
Definition: pipeline.cpp:268
tbb::filter::has_more_work
bool has_more_work()
has the filter not yet processed all the tokens it will ever see?
Definition: pipeline.cpp:691
tbb::internal::pipeline_cleaner::~pipeline_cleaner
~pipeline_cleaner()
Definition: pipeline.cpp:506
semaphore.h
itt_notify.h
tbb::internal::input_buffer::set_my_tls_end_of_input
void set_my_tls_end_of_input()
Definition: pipeline.cpp:228
tbb::internal::input_buffer::end_of_input_tls
end_of_input_tls_t end_of_input_tls
Definition: pipeline.cpp:93
tbb::internal::stage_task::execute
task * execute() __TBB_override
The virtual task execution method.
Definition: pipeline.cpp:301
tbb::internal::input_buffer::put_token
bool put_token(task_info &info_, bool force_put=false)
Put a token into the buffer.
Definition: pipeline.cpp:135
tbb::task::recycle_as_continuation
void recycle_as_continuation()
Change this to be a continuation of its former self.
Definition: task.h:700
tbb::filter::exact_exception_propagation
static const unsigned char exact_exception_propagation
7th bit defines exception propagation mode expected by the application.
Definition: pipeline.h:84
new_size
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
Definition: ittnotify_static.h:163
tbb::thread_bound_filter
A stage in a pipeline served by a user thread.
Definition: pipeline.h:196
tbb::internal::input_buffer::return_item
bool return_item(task_info &info, bool advance)
return an item, invalidate the queued item, but only advance if the filter
Definition: pipeline.cpp:208
tbb::internal::pipeline_cleaner::my_pipeline
pipeline & my_pipeline
Definition: pipeline.cpp:501
tbb::filter::finalize
virtual void finalize(void *)
Destroys item if pipeline was cancelled.
Definition: pipeline.h:159
tbb::internal::input_buffer::array_mutex
spin_mutex array_mutex
Serializes updates.
Definition: pipeline.cpp:72
tbb::internal::input_buffer::end_of_input_tls_allocated
bool end_of_input_tls_allocated
Definition: pipeline.cpp:94
tbb::internal::pipeline_root_task::my_pipeline
pipeline & my_pipeline
Definition: pipeline.cpp:401
tbb::internal::task_info
This structure is used to store task information in a input buffer.
Definition: pipeline.cpp:30
tbb::internal::task_info::is_valid
bool is_valid
True if my_object is valid.
Definition: pipeline.cpp:37
tbb::filter
A stage in a pipeline.
Definition: pipeline.h:64
lock
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
Definition: ittnotify_static.h:121
__TBB_Yield
#define __TBB_Yield()
Definition: ibm_aix51.h:44
tbb::filter::~filter
virtual __TBB_EXPORTED_METHOD ~filter()
Destroy filter.
Definition: pipeline.cpp:697
tbb::internal::pipeline_cleaner
Definition: pipeline.cpp:500
tbb::internal::input_buffer::sema_P
void sema_P()
Definition: pipeline.cpp:98
tbb::internal::input_buffer::my_tls_end_of_input
bool my_tls_end_of_input()
Definition: pipeline.cpp:227
tbb::internal::semaphore
Edsger Dijkstra's counting semaphore.
Definition: semaphore.h:94
tbb::thread_bound_filter::try_process_item
result_type __TBB_EXPORTED_METHOD try_process_item()
If a data item is available, invoke operator() on that item.
Definition: pipeline.cpp:723
tbb::internal::stage_task
Definition: pipeline.cpp:249
tbb::internal::first
auto first(Container &c) -> decltype(begin(c))
Definition: _range_iterator.h:34
__TBB_override
#define __TBB_override
Definition: tbb_stddef.h:240
tbb::internal::pipeline_root_task::execute
task * execute() __TBB_override
Should be overridden by derived classes.
Definition: pipeline.cpp:404
tbb::task::parent
task * parent() const
task on whose behalf this task is working, or NULL if this is a root.
Definition: task.h:854
tbb::internal::input_buffer::sema_V
void sema_V()
Definition: pipeline.cpp:99
tbb::internal::input_buffer::array
task_info * array
Array of deferred tasks that cannot yet start executing.
Definition: pipeline.cpp:58
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
tbb::internal::input_buffer::array_size
size_type array_size
Size of array.
Definition: pipeline.cpp:65
tbb::internal::task_info::my_token
Token my_token
Invalid unless a task went through an ordered stage.
Definition: pipeline.cpp:33
tbb::filter::my_input_buffer
internal::input_buffer * my_input_buffer
Buffer for incoming tokens, or NULL if not required.
Definition: pipeline.h:173
tbb::internal::input_buffer::is_bound
bool is_bound
True for thread-bound filter, false otherwise.
Definition: pipeline.cpp:89
tbb::internal::basic_tls< intptr_t >
tbb::task::allocate_root
static internal::allocate_root_proxy allocate_root()
Returns proxy for overloaded new that allocates a root task.
Definition: task.h:652
tbb::internal::input_buffer::grow
void grow(size_type minimum_size)
Resize "array".
Definition: pipeline.cpp:231
tbb::internal::input_buffer::is_ordered
bool is_ordered
True for ordered filter, false otherwise.
Definition: pipeline.cpp:86
tbb::internal::stage_task::stage_task
stage_task(pipeline &pipeline)
Construct stage_task for first stage in a pipeline.
Definition: pipeline.cpp:260
tbb::internal::basic_tls::set
void set(T value)
Definition: tls.h:56
tbb::thread_bound_filter::internal_process_item
result_type internal_process_item(bool is_blocking)
Internal routine for item processing.
Definition: pipeline.cpp:727
tbb::internal::input_buffer::destroy_my_tls
void destroy_my_tls()
Definition: pipeline.cpp:226
tbb::internal::input_buffer::create_my_tls
void create_my_tls()
Definition: pipeline.cpp:225
tbb::task
Base class for user-defined tasks.
Definition: task.h:604
tbb::filter::object_may_be_null
bool object_may_be_null()
true if an input filter can emit null
Definition: pipeline.h:143
sync_releasing
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
Definition: ittnotify_static.h:104
tbb::filter::next_filter_in_pipeline
filter * next_filter_in_pipeline
Pointer to next filter in the pipeline.
Definition: pipeline.h:164
tbb::internal::tokendiff_t
long tokendiff_t
Definition: pipeline.h:44
tbb::internal::pipeline_cleaner::pipeline_cleaner
pipeline_cleaner(pipeline &_pipeline)
Definition: pipeline.cpp:503
tbb::internal::input_buffer::~input_buffer
~input_buffer()
Destroy the buffer.
Definition: pipeline.cpp:114
tbb::internal::input_buffer::initial_buffer_size
static const size_type initial_buffer_size
Initial size for "array".
Definition: pipeline.cpp:80
tbb::internal::task_info::my_token_ready
bool my_token_ready
False until my_token is set.
Definition: pipeline.cpp:35
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
tbb::internal::semaphore::V
void V()
post/release
Definition: semaphore.h:110
tbb::task::is_cancelled
bool is_cancelled() const
Returns true if the context has received cancellation request.
Definition: task.h:963
pipeline.h
tbb::task::spawn_root_and_wait
static void spawn_root_and_wait(task &root)
Spawn task allocated by allocate_root, wait for it to complete, and deallocate it.
Definition: task.h:797
tbb::filter::my_filter_mode
const unsigned char my_filter_mode
Storage for filter mode and dynamically checked implementation version.
Definition: pipeline.h:181
tbb::filter::next_segment
filter * next_segment
Pointer to the next "segment" of filters, or NULL if not required.
Definition: pipeline.h:191
tbb::filter::prev_filter_in_pipeline
filter * prev_filter_in_pipeline
Pointer to previous filter in the pipeline.
Definition: pipeline.h:184
tbb::internal::input_buffer::low_token
Token low_token
Lowest token that can start executing.
Definition: pipeline.cpp:69
tbb::thread_bound_filter::process_item
result_type __TBB_EXPORTED_METHOD process_item()
Wait until a data item becomes available, and invoke operator() on that item.
Definition: pipeline.cpp:719

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.