tbb::pipeline Class Reference

A processing pipeline that applies filters to items. More...

#include <pipeline.h>

Collaboration diagram for tbb::pipeline:

Public Member Functions
__TBB_EXPORTED_METHOD	pipeline ()
	Construct empty pipeline. More...
virtual __TBB_EXPORTED_METHOD	~pipeline ()
void __TBB_EXPORTED_METHOD	add_filter (filter &filter_)
	Add filter to end of pipeline. More...
void __TBB_EXPORTED_METHOD	run (size_t max_number_of_live_tokens)
	Run the pipeline to completion. More...
void __TBB_EXPORTED_METHOD	run (size_t max_number_of_live_tokens, tbb::task_group_context &context)
	Run the pipeline to completion with user-supplied context. More...
void __TBB_EXPORTED_METHOD	clear ()
	Remove all filters from the pipeline. More...

Private Member Functions
void	remove_filter (filter &filter_)
	Remove filter from pipeline. More...
void __TBB_EXPORTED_METHOD	inject_token (task &self)
	Not used, but retained to satisfy old export files. More...
void	clear_filters ()
	Does clean up if pipeline is cancelled or exception occurred. More...

Private Attributes
filter *	filter_list
	Pointer to first filter in the pipeline. More...
filter *	filter_end
	Pointer to location where address of next filter to be added should be stored. More...
task *	end_counter
	task who's reference count is used to determine when all stages are done. More...
atomic< internal::Token >	input_tokens
	Number of idle tokens waiting for input stage. More...
atomic< internal::Token >	token_counter
	Global counter of tokens. More...
bool	end_of_input
	False until fetch_input returns NULL. More...
bool	has_thread_bound_filters
	True if the pipeline contains a thread-bound filter; false otherwise. More...

Friends
class	internal::stage_task
class	internal::pipeline_root_task
class	filter
class	thread_bound_filter
class	internal::pipeline_cleaner
class	tbb::interface6::internal::pipeline_proxy

Detailed Description

A processing pipeline that applies filters to items.

Definition at line 236 of file pipeline.h.

Constructor & Destructor Documentation

pipeline()

tbb::pipeline::pipeline

(

)

Construct empty pipeline.

Definition at line 518 of file pipeline.cpp.

References input_tokens, and token_counter.

                   :
    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()

tbb::pipeline::~pipeline ( )

virtual

Though the current implementation declares the destructor virtual, do not rely on this detail. The virtualness is deprecated and may disappear in future versions of TBB.

Definition at line 529 of file pipeline.cpp.

References clear().

                    {
    clear();
}

Here is the call graph for this function:

Member Function Documentation

add_filter()

void tbb::pipeline::add_filter

(

filter &

filter_

)

Add filter to end of pipeline.

Definition at line 552 of file pipeline.cpp.

References __TBB_ASSERT, __TBB_PIPELINE_VERSION, end_counter, filter_end, filter_list, has_thread_bound_filters, tbb::filter::is_bound(), tbb::filter::is_ordered(), tbb::filter::is_serial(), tbb::filter::my_filter_mode, tbb::filter::my_input_buffer, tbb::filter::my_pipeline, tbb::filter::next_filter_in_pipeline, tbb::filter::not_in_pipeline(), tbb::filter::object_may_be_null(), tbb::filter::prev_filter_in_pipeline, and tbb::filter::version_mask.

                                           {
#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 );
        }
    }

}

Here is the call graph for this function:

clear()

void tbb::pipeline::clear

(

)

Remove all filters from the pipeline.

Definition at line 533 of file pipeline.cpp.

References __TBB_PIPELINE_VERSION, filter_end, filter_list, tbb::filter::my_input_buffer, tbb::filter::next_filter_in_pipeline, tbb::filter::not_in_pipeline(), and tbb::filter::version_mask.

Referenced by tbb::internal::input_buffer::note_done(), and ~pipeline().

                     {
    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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

clear_filters()

void tbb::pipeline::clear_filters ( )

private

Does clean up if pipeline is cancelled or exception occurred.

Referenced by inject_token(), and tbb::internal::pipeline_cleaner::~pipeline_cleaner().

Here is the caller graph for this function:

inject_token()

void tbb::pipeline::inject_token ( task &  self )

private

Not used, but retained to satisfy old export files.

Definition at line 504 of file pipeline.cpp.

References __TBB_ASSERT, __TBB_PIPELINE_VERSION, clear_filters(), tbb::filter::next_filter_in_pipeline, and tbb::filter::version_mask.

                                   {
    __TBB_ASSERT(false,"illegal call to inject_token");
}

Here is the call graph for this function:

remove_filter()

void tbb::pipeline::remove_filter ( filter &  filter_ )

private

Remove filter from pipeline.

Definition at line 607 of file pipeline.cpp.

References __TBB_ASSERT, __TBB_PIPELINE_VERSION, end_counter, filter_end, filter_list, tbb::filter::my_filter_mode, tbb::filter::my_input_buffer, tbb::filter::my_pipeline, tbb::filter::next_filter_in_pipeline, tbb::filter::next_segment, tbb::filter::not_in_pipeline(), tbb::filter::prev_filter_in_pipeline, and tbb::filter::version_mask.

                                              {
    __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;
}

Here is the call graph for this function:

run() [1/2]

void tbb::pipeline::run

(

size_t

max_number_of_live_tokens

)

Run the pipeline to completion.

Definition at line 633 of file pipeline.cpp.

References __TBB_ASSERT, tbb::task::allocate_root(), tbb::task_group_context::bound, tbb::task_group_context::default_traits, end_counter, end_of_input, tbb::filter::exact_exception_propagation, filter_list, has_thread_bound_filters, input_tokens, internal::pipeline_cleaner, internal::pipeline_root_task, tbb::filter::is_bound(), tbb::filter::my_filter_mode, tbb::filter::my_input_buffer, tbb::filter::next_filter_in_pipeline, and tbb::task::spawn_root_and_wait().

      {
    __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
                }
            }
        }
    }
}

Here is the call graph for this function:

run() [2/2]

void __TBB_EXPORTED_METHOD tbb::pipeline::run	(	size_t	max_number_of_live_tokens,
		tbb::task_group_context &	context
	)

Run the pipeline to completion with user-supplied context.

Friends And Related Function Documentation

filter

friend class filter

friend

Definition at line 262 of file pipeline.h.

internal::pipeline_cleaner

friend class internal::pipeline_cleaner

friend

Definition at line 264 of file pipeline.h.

Referenced by run().

internal::pipeline_root_task

friend class internal::pipeline_root_task

friend

Definition at line 261 of file pipeline.h.

Referenced by run().

internal::stage_task

friend class internal::stage_task

friend

Definition at line 260 of file pipeline.h.

tbb::interface6::internal::pipeline_proxy

friend class tbb::interface6::internal::pipeline_proxy

friend

Definition at line 265 of file pipeline.h.

thread_bound_filter

friend class thread_bound_filter

friend

Definition at line 263 of file pipeline.h.

Member Data Documentation

end_counter

task* tbb::pipeline::end_counter

private

task who's reference count is used to determine when all stages are done.

Definition at line 274 of file pipeline.h.

Referenced by add_filter(), remove_filter(), run(), and tbb::internal::pipeline_cleaner::~pipeline_cleaner().

end_of_input

bool tbb::pipeline::end_of_input

private

False until fetch_input returns NULL.

Definition at line 283 of file pipeline.h.

Referenced by tbb::internal::pipeline_root_task::execute(), and run().

filter_end

filter* tbb::pipeline::filter_end

private

Pointer to location where address of next filter to be added should be stored.

Definition at line 271 of file pipeline.h.

Referenced by add_filter(), clear(), and remove_filter().

filter_list

filter* tbb::pipeline::filter_list

private

Pointer to first filter in the pipeline.

Definition at line 268 of file pipeline.h.

Referenced by add_filter(), clear(), tbb::internal::pipeline_root_task::execute(), tbb::internal::pipeline_root_task::pipeline_root_task(), remove_filter(), tbb::internal::stage_task::reset(), run(), and tbb::interface6::internal::pipeline_proxy::~pipeline_proxy().

has_thread_bound_filters

bool tbb::pipeline::has_thread_bound_filters

private

True if the pipeline contains a thread-bound filter; false otherwise.

Definition at line 286 of file pipeline.h.

Referenced by add_filter(), and run().

input_tokens

atomic<internal::Token> tbb::pipeline::input_tokens

private

Number of idle tokens waiting for input stage.

Definition at line 277 of file pipeline.h.

Referenced by tbb::internal::pipeline_root_task::execute(), pipeline(), and run().

token_counter

atomic<internal::Token> tbb::pipeline::token_counter

private

Global counter of tokens.

Definition at line 280 of file pipeline.h.

Referenced by pipeline().

---

The documentation for this class was generated from the following files:

• pipeline.h
• pipeline.cpp