tbb::internal::input_buffer Class Reference

A buffer of input items for a filter. More...

Inheritance diagram for tbb::internal::input_buffer:

Collaboration diagram for tbb::internal::input_buffer:

Public Member Functions
	input_buffer (bool is_ordered_, bool is_bound_)
	Construct empty buffer. More...
	~input_buffer ()
	Destroy the buffer. More...
bool	put_token (task_info &info_, bool force_put=false)
	Put a token into the buffer. More...
template<typename StageTask >
void	note_done (Token token, StageTask &spawner)
	Note that processing of a token is finished. More...
bool	return_item (task_info &info, bool advance)
	return an item, invalidate the queued item, but only advance if advance More...
bool	has_item ()
	true if the current low_token is valid. More...
void	create_my_tls ()
void	destroy_my_tls ()
bool	my_tls_end_of_input ()
void	set_my_tls_end_of_input ()

Private Types
typedef Token	size_type
typedef basic_tls< intptr_t >	end_of_input_tls_t
	for parallel filters that accepts NULLs, thread-local flag for reaching end_of_input More...

Private Member Functions
void	grow (size_type minimum_size)
	Resize "array". More...
void	create_sema (size_t initial_tokens)
void	free_sema ()
void	sema_P ()
void	sema_V ()
Private Member Functions inherited from tbb::internal::no_copy
	no_copy ()
	Allow default construction. More...

Private Attributes
task_info *	array
	Array of deferred tasks that cannot yet start executing. More...
semaphore *	my_sem
	for thread-bound filter, semaphore for waiting, NULL otherwise. More...
size_type	array_size
	Size of array. More...
Token	low_token
	Lowest token that can start executing. More...
spin_mutex	array_mutex
	Serializes updates. More...
Token	high_token
	Used for out of order buffer, and for assigning my_token if is_ordered and my_token not already assigned. More...
bool	is_ordered
	True for ordered filter, false otherwise. More...
bool	is_bound
	True for thread-bound filter, false otherwise. More...
end_of_input_tls_t	end_of_input_tls
bool	end_of_input_tls_allocated

Static Private Attributes
static const size_type	initial_buffer_size = 4
	Initial size for "array". More...

Friends
class	tbb::internal::pipeline_root_task
class	tbb::filter
class	tbb::thread_bound_filter
class	tbb::internal::stage_task
class	tbb::pipeline

Detailed Description

A buffer of input items for a filter.

Each item is a task_info, inserted into a position in the buffer corresponding to a Token.

Definition at line 52 of file pipeline.cpp.

Member Typedef Documentation

end_of_input_tls_t

typedef basic_tls<intptr_t> tbb::internal::input_buffer::end_of_input_tls_t

private

for parallel filters that accepts NULLs, thread-local flag for reaching end_of_input

Definition at line 96 of file pipeline.cpp.

size_type

typedef Token tbb::internal::input_buffer::size_type

private

Definition at line 59 of file pipeline.cpp.

Constructor & Destructor Documentation

input_buffer()

tbb::internal::input_buffer::input_buffer ( bool  is_ordered_, bool  is_bound_  )

inline

Construct empty buffer.

Definition at line 107 of file pipeline.cpp.

References __TBB_ASSERT.

                                                     :
            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::internal::input_buffer::~input_buffer ( )

inline

Destroy the buffer.

Definition at line 118 of file pipeline.cpp.

References __TBB_ASSERT, and tbb::internal::poison_pointer().

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

Here is the call graph for this function:

Member Function Documentation

create_my_tls()

void tbb::internal::input_buffer::create_my_tls ( )

inline

Definition at line 229 of file pipeline.cpp.

References tbb::internal::basic_tls< T >::create(), and tbb::internal::handle_perror().

{ int status = end_of_input_tls.create(); if(status) handle_perror(status, "TLS not allocated for filter"); end_of_input_tls_allocated = true; }

Here is the call graph for this function:

create_sema()

void tbb::internal::input_buffer::create_sema ( size_t  initial_tokens )

inlineprivate

Definition at line 100 of file pipeline.cpp.

References __TBB_ASSERT.

{ __TBB_ASSERT(!my_sem,NULL); my_sem = new internal::semaphore(initial_tokens); }

destroy_my_tls()

void tbb::internal::input_buffer::destroy_my_tls ( )

inline

Definition at line 230 of file pipeline.cpp.

References tbb::internal::basic_tls< T >::destroy(), and tbb::internal::handle_perror().

{ int status = end_of_input_tls.destroy(); if(status) handle_perror(status, "Failed to destroy filter TLS"); }

Here is the call graph for this function:

free_sema()

void tbb::internal::input_buffer::free_sema ( )

inlineprivate

Definition at line 101 of file pipeline.cpp.

References __TBB_ASSERT.

{ __TBB_ASSERT(my_sem,NULL); delete my_sem; }

grow()

void tbb::internal::input_buffer::grow ( size_type  minimum_size )

private

Resize "array".

Caller is responsible to acquiring a lock on "array_mutex".

Definition at line 235 of file pipeline.cpp.

References tbb::internal::task_info::is_valid, and new_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);
}

has_item()

bool tbb::internal::input_buffer::has_item ( )

inline

true if the current low_token is valid.

Definition at line 226 of file pipeline.cpp.

References tbb::internal::task_info::is_valid, and lock.

{ spin_mutex::scoped_lock lock(array_mutex); return array[low_token&(array_size -1)].is_valid; }

my_tls_end_of_input()

bool tbb::internal::input_buffer::my_tls_end_of_input ( )

inline

Definition at line 231 of file pipeline.cpp.

References tbb::internal::basic_tls< T >::get().

{ return end_of_input_tls.get() != 0; }

Here is the call graph for this function:

note_done()

template<typename StageTask >

void tbb::internal::input_buffer::note_done ( Token  token, StageTask &  spawner  )

inline

Note that processing of a token is finished.

Fires up processing of the next token, if processing was deferred.

Definition at line 178 of file pipeline.cpp.

References tbb::pipeline::clear(), tbb::filter::finalize(), tbb::internal::task_info::is_valid, ITT_NOTIFY, lock, tbb::internal::task_info::my_object, and tbb::internal::task_info::reset().

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

Here is the call graph for this function:

put_token()

bool tbb::internal::input_buffer::put_token ( task_info &  info_, bool  force_put = false  )

inline

Put a token into the buffer.

If task information was placed into buffer, returns true; otherwise returns false, informing the caller to create and spawn a task. If input buffer owned by thread-bound filter and the item at low_token was not valid, issue a V() If the input_buffer is owned by a successor to a thread-bound filter, the force_put parameter should be true to ensure the token is inserted in the buffer.

Definition at line 139 of file pipeline.cpp.

References __TBB_ASSERT, tbb::internal::task_info::is_valid, ITT_NOTIFY, lock, tbb::internal::task_info::my_token, tbb::internal::task_info::my_token_ready, and sync_releasing.

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

return_item()

bool tbb::internal::input_buffer::return_item ( task_info &  info, bool  advance  )

inline

return an item, invalidate the queued item, but only advance if advance

Definition at line 212 of file pipeline.cpp.

References tbb::internal::task_info::is_valid, ITT_NOTIFY, and lock.

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

sema_P()

void tbb::internal::input_buffer::sema_P ( )

inlineprivate

Definition at line 102 of file pipeline.cpp.

References __TBB_ASSERT, and tbb::internal::semaphore::P().

{ __TBB_ASSERT(my_sem,NULL); my_sem->P(); }

Here is the call graph for this function:

sema_V()

void tbb::internal::input_buffer::sema_V ( )

inlineprivate

Definition at line 103 of file pipeline.cpp.

References __TBB_ASSERT, and tbb::internal::semaphore::V().

{ __TBB_ASSERT(my_sem,NULL); my_sem->V(); }

Here is the call graph for this function:

set_my_tls_end_of_input()

void tbb::internal::input_buffer::set_my_tls_end_of_input ( )

inline

Definition at line 232 of file pipeline.cpp.

References tbb::internal::basic_tls< T >::set().

{ end_of_input_tls.set(1); }

Here is the call graph for this function:

Friends And Related Function Documentation

tbb::filter

friend class tbb::filter

friend

Definition at line 54 of file pipeline.cpp.

tbb::internal::pipeline_root_task

friend class tbb::internal::pipeline_root_task

friend

Definition at line 53 of file pipeline.cpp.

tbb::internal::stage_task

friend class tbb::internal::stage_task

friend

Definition at line 56 of file pipeline.cpp.

tbb::pipeline

friend class tbb::pipeline

friend

Definition at line 57 of file pipeline.cpp.

tbb::thread_bound_filter

friend class tbb::thread_bound_filter

friend

Definition at line 55 of file pipeline.cpp.

Member Data Documentation

array

task_info* tbb::internal::input_buffer::array

private

Array of deferred tasks that cannot yet start executing.

Definition at line 62 of file pipeline.cpp.

array_mutex

spin_mutex tbb::internal::input_buffer::array_mutex

private

Serializes updates.

Definition at line 76 of file pipeline.cpp.

array_size

size_type tbb::internal::input_buffer::array_size

private

Size of array.

Always 0 or a power of 2

Definition at line 69 of file pipeline.cpp.

end_of_input_tls

end_of_input_tls_t tbb::internal::input_buffer::end_of_input_tls

private

Definition at line 97 of file pipeline.cpp.

end_of_input_tls_allocated

bool tbb::internal::input_buffer::end_of_input_tls_allocated

private

Definition at line 98 of file pipeline.cpp.

high_token

Token tbb::internal::input_buffer::high_token

private

Used for out of order buffer, and for assigning my_token if is_ordered and my_token not already assigned.

Definition at line 87 of file pipeline.cpp.

initial_buffer_size

const size_type tbb::internal::input_buffer::initial_buffer_size = 4

staticprivate

Initial size for "array".

Must be a power of 2

Definition at line 84 of file pipeline.cpp.

is_bound

bool tbb::internal::input_buffer::is_bound

private

True for thread-bound filter, false otherwise.

Definition at line 93 of file pipeline.cpp.

is_ordered

bool tbb::internal::input_buffer::is_ordered

private

True for ordered filter, false otherwise.

Definition at line 90 of file pipeline.cpp.

low_token

Token tbb::internal::input_buffer::low_token

private

Lowest token that can start executing.

All prior Token have already been seen.

Definition at line 73 of file pipeline.cpp.

my_sem

semaphore* tbb::internal::input_buffer::my_sem

private

for thread-bound filter, semaphore for waiting, NULL otherwise.

Definition at line 65 of file pipeline.cpp.

---

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

• pipeline.cpp