tbb::internal::task_stream< Levels > Class Template Reference

The container for "fairness-oriented" aka "enqueued" tasks. More...

#include <task_stream.h>

Inheritance diagram for tbb::internal::task_stream< Levels >:

Collaboration diagram for tbb::internal::task_stream< Levels >:

Public Member Functions
	task_stream ()
void	initialize (unsigned n_lanes)
	~task_stream ()
void	push (task *source, int level, FastRandom &random)
	Push a task into a lane. More...
task *	pop (int level, unsigned &last_used_lane)
	Try finding and popping a task. More...
bool	empty (int level)
	Checks existence of a task. More...
intptr_t	drain ()
	Destroys all remaining tasks in every lane. Returns the number of destroyed tasks. More...
	task_stream ()
void	initialize (unsigned n_lanes)
	~task_stream ()
bool	try_push (task *source, int level, unsigned lane_idx)
	Returns true on successful push, otherwise - false. More...
template<typename lane_selector_t >
void	push (task *source, int level, const lane_selector_t &next_lane)
	Push a task into a lane. Lane selection is performed by passed functor. More...
task *	try_pop (int level, unsigned lane_idx)
	Returns pointer to task on successful pop, otherwise - NULL. More...
template<typename lane_selector_t >
task *	pop (int level, const lane_selector_t &next_lane)
task *	look_specific (__TBB_ISOLATION_ARG(task_stream_base::lane_t::queue_base_t &queue, isolation_tag isolation))
task *	pop_specific (int level, __TBB_ISOLATION_ARG(unsigned &last_used_lane, isolation_tag isolation))
	Try finding and popping a related task. More...
bool	empty (int level)
	Checks existence of a task. More...
intptr_t	drain ()
	Destroys all remaining tasks in every lane. Returns the number of destroyed tasks. More...

Private Types
typedef queue_and_mutex< task *, spin_mutex >	lane_t
typedef task_stream_accessor< accessor >::lane_t	lane_t

Private Attributes
population_t	population [Levels]
padded< lane_t > *	lanes [Levels]
unsigned	N

Additional Inherited Members
Protected Types inherited from tbb::internal::task_stream_accessor< accessor >
typedef queue_and_mutex< task *, spin_mutex >	lane_t
Protected Types inherited from tbb::internal::task_stream_base
typedef queue_and_mutex< task *, spin_mutex >	lane_t
Protected Member Functions inherited from tbb::internal::task_stream_accessor< accessor >
task *	get_item (lane_t::queue_base_t &queue)
Private Member Functions inherited from tbb::internal::no_copy
	no_copy (const no_copy &)=delete
	no_copy ()=default

Detailed Description

template<int Levels>
class tbb::internal::task_stream< Levels >

The container for "fairness-oriented" aka "enqueued" tasks.

Definition at line 69 of file task_stream.h.

Member Typedef Documentation

lane_t [1/2]

template<int Levels>

typedef queue_and_mutex<task*, spin_mutex> tbb::internal::task_stream< Levels >::lane_t

private

Definition at line 70 of file task_stream.h.

lane_t [2/2]

template<int Levels>

typedef task_stream_accessor<accessor>::lane_t tbb::internal::task_stream< Levels >::lane_t

private

Definition at line 162 of file task_stream_extended.h.

Constructor & Destructor Documentation

task_stream() [1/2]

template<int Levels>

tbb::internal::task_stream< Levels >::task_stream ( )

inline

Definition at line 76 of file task_stream.h.

                  : N() {
        for(int level = 0; level < Levels; level++) {
            population[level] = 0;
            lanes[level] = NULL;
        }
    }

~task_stream() [1/2]

template<int Levels>

tbb::internal::task_stream< Levels >::~task_stream ( )

inline

Definition at line 95 of file task_stream.h.

                   {
        for(int level = 0; level < Levels; level++)
            if (lanes[level]) delete[] lanes[level];
    }

task_stream() [2/2]

template<int Levels>

tbb::internal::task_stream< Levels >::task_stream ( )

inline

Definition at line 168 of file task_stream_extended.h.

                  : N() {
        for(int level = 0; level < Levels; level++) {
            population[level] = 0;
            lanes[level] = NULL;
        }
    }

~task_stream() [2/2]

template<int Levels>

tbb::internal::task_stream< Levels >::~task_stream ( )

inline

Definition at line 187 of file task_stream_extended.h.

                   {
        for(int level = 0; level < Levels; level++)
            if (lanes[level]) delete[] lanes[level];
    }

Member Function Documentation

drain() [1/2]

template<int Levels>

intptr_t tbb::internal::task_stream< Levels >::drain ( )

inline

Destroys all remaining tasks in every lane. Returns the number of destroyed tasks.

Tasks are not executed, because it would potentially create more tasks at a late stage. The scheduler is really expected to execute all tasks before task_stream destruction.

Definition at line 145 of file task_stream.h.

                     {
        intptr_t result = 0;
        for(int level = 0; level < Levels; level++)
            for(unsigned i=0; i<N; ++i) {
                lane_t& lane = lanes[level][i];
                spin_mutex::scoped_lock lock(lane.my_mutex);
                for(lane_t::queue_base_t::iterator it=lane.my_queue.begin();
                    it!=lane.my_queue.end(); ++it, ++result)
                {
                    __TBB_ASSERT( is_bit_set( population[level], i ), NULL );
                    task* t = *it;
                    tbb::task::destroy(*t);
                }
                lane.my_queue.clear();
                clear_one_bit( population[level], i );
            }
        return result;
    }

Referenced by tbb::internal::arena::free_arena().

Here is the caller graph for this function:

drain() [2/2]

template<int Levels>

intptr_t tbb::internal::task_stream< Levels >::drain ( )

inline

Destroys all remaining tasks in every lane. Returns the number of destroyed tasks.

Tasks are not executed, because it would potentially create more tasks at a late stage. The scheduler is really expected to execute all tasks before task_stream destruction.

Definition at line 296 of file task_stream_extended.h.

                     {
        intptr_t result = 0;
        for(int level = 0; level < Levels; level++)
            for(unsigned i=0; i<N; ++i) {
                lane_t& lane = lanes[level][i];
                spin_mutex::scoped_lock lock(lane.my_mutex);
                for(typename lane_t::queue_base_t::iterator it=lane.my_queue.begin();
                    it!=lane.my_queue.end(); ++it, ++result)
                {
                    __TBB_ASSERT( is_bit_set( population[level], i ), NULL );
                    task* t = *it;
                    tbb::task::destroy(*t);
                }
                lane.my_queue.clear();
                clear_one_bit( population[level], i );
            }
        return result;
    }

empty() [1/2]

template<int Levels>

bool tbb::internal::task_stream< Levels >::empty ( int  level )

inline

Checks existence of a task.

Definition at line 138 of file task_stream.h.

                          {
        return !population[level];
    }

Referenced by tbb::internal::arena::has_enqueued_tasks(), tbb::internal::arena::is_out_of_work(), tbb::internal::task_stream< num_priority_levels >::pop(), tbb::internal::task_stream< num_priority_levels >::pop_specific(), and tbb::internal::arena::restore_priority_if_need().

Here is the caller graph for this function:

empty() [2/2]

template<int Levels>

bool tbb::internal::task_stream< Levels >::empty ( int  level )

inline

Checks existence of a task.

Definition at line 289 of file task_stream_extended.h.

                          {
        return !population[level];
    }

initialize() [1/2]

template<int Levels>

void tbb::internal::task_stream< Levels >::initialize ( unsigned  n_lanes )

inline

Definition at line 83 of file task_stream.h.

                                        {
        const unsigned max_lanes = sizeof(population_t) * CHAR_BIT;

        N = n_lanes>=max_lanes ? max_lanes : n_lanes>2 ? 1<<(__TBB_Log2(n_lanes-1)+1) : 2;
        __TBB_ASSERT( N==max_lanes || N>=n_lanes && ((N-1)&N)==0, "number of lanes miscalculated");
        __TBB_ASSERT( N <= sizeof(population_t) * CHAR_BIT, NULL );
        for(int level = 0; level < Levels; level++) {
            lanes[level] = new padded<lane_t>[N];
            __TBB_ASSERT( !population[level], NULL );
        }
    }

Referenced by tbb::internal::arena::arena().

Here is the caller graph for this function:

initialize() [2/2]

template<int Levels>

void tbb::internal::task_stream< Levels >::initialize ( unsigned  n_lanes )

inline

Definition at line 175 of file task_stream_extended.h.

                                        {
        const unsigned max_lanes = sizeof(population_t) * CHAR_BIT;

        N = n_lanes>=max_lanes ? max_lanes : n_lanes>2 ? 1<<(__TBB_Log2(n_lanes-1)+1) : 2;
        __TBB_ASSERT( N==max_lanes || N>=n_lanes && ((N-1)&N)==0, "number of lanes miscalculated");
        __TBB_ASSERT( N <= sizeof(population_t) * CHAR_BIT, NULL );
        for(int level = 0; level < Levels; level++) {
            lanes[level] = new padded<lane_t>[N];
            __TBB_ASSERT( !population[level], NULL );
        }
    }

look_specific()

template<int Levels>

task* tbb::internal::task_stream< Levels >::look_specific ( __TBB_ISOLATION_ARG(task_stream_base::lane_t::queue_base_t &queue, isolation_tag isolation)  )

inline

Definition at line 245 of file task_stream_extended.h.

                                                                                                                   {
        __TBB_ASSERT( !queue.empty(), NULL );
        // TODO: add a worst-case performance test and consider an alternative container with better
        // performance for isolation search.
        typename lane_t::queue_base_t::iterator curr = queue.end();
        do {
            // TODO: consider logic from get_task to simplify the code.
            task* result = *--curr;
            if( result __TBB_ISOLATION_EXPR( && result->prefix().isolation == isolation ) ) {
                if( queue.end() - curr == 1 )
                    queue.pop_back(); // a little of housekeeping along the way
                else
                    *curr = 0;      // grabbing task with the same isolation
                // TODO: move one of the container's ends instead if the task has been found there
                return result;
            }
        } while( curr != queue.begin() );
        return NULL;
    }

Referenced by tbb::internal::task_stream< num_priority_levels >::pop_specific().

Here is the caller graph for this function:

pop() [1/2]

template<int Levels>

task* tbb::internal::task_stream< Levels >::pop ( int  level, unsigned &  last_used_lane  )

inline

Try finding and popping a task.

Definition at line 116 of file task_stream.h.

                                                     {
        task* result = NULL;
        // Lane selection is round-robin. Each thread should keep its last used lane.
        unsigned idx = (last_used_lane+1)&(N-1);
        for( ; population[level]; idx=(idx+1)&(N-1) ) {
            if( is_bit_set( population[level], idx ) ) {
                lane_t& lane = lanes[level][idx];
                spin_mutex::scoped_lock lock;
                if( lock.try_acquire(lane.my_mutex) && !lane.my_queue.empty() ) {
                    result = lane.my_queue.front();
                    lane.my_queue.pop_front();
                    if( lane.my_queue.empty() )
                        clear_one_bit( population[level], idx );
                    break;
                }
            }
        }
        last_used_lane = idx;
        return result;
    }

pop() [2/2]

template<int Levels>

template<typename lane_selector_t >

task* tbb::internal::task_stream< Levels >::pop ( int  level, const lane_selector_t &  next_lane  )

inline

Try finding and popping a task using passed functor for lane selection. Last used lane is updated inside lane selector.

Definition at line 234 of file task_stream_extended.h.

                                                             {
        task* popped = NULL;
        unsigned lane = 0;
        do {
            lane = next_lane( /*out_of=*/N );
            __TBB_ASSERT( lane < N, "Incorrect lane index." );
        } while( !empty( level ) && !(popped = try_pop( level, lane )) );
        return popped;
    }

pop_specific()

template<int Levels>

task* tbb::internal::task_stream< Levels >::pop_specific ( int  level, __TBB_ISOLATION_ARG(unsigned &last_used_lane, isolation_tag isolation)    )

inline

Try finding and popping a related task.

Definition at line 266 of file task_stream_extended.h.

                                                                                                            {
        task* result = NULL;
        // Lane selection is round-robin in backward direction.
        unsigned idx = last_used_lane & (N-1);
        do {
            if( is_bit_set( population[level], idx ) ) {
                lane_t& lane = lanes[level][idx];
                spin_mutex::scoped_lock lock;
                if( lock.try_acquire(lane.my_mutex) && !lane.my_queue.empty() ) {
                    result = look_specific( __TBB_ISOLATION_ARG(lane.my_queue, isolation) );
                    if( lane.my_queue.empty() )
                        clear_one_bit( population[level], idx );
                    if( result )
                        break;
                }
            }
            idx=(idx-1)&(N-1);
        } while( !empty(level) && idx != last_used_lane );
        last_used_lane = idx;
        return result;
    }

push() [1/2]

template<int Levels>

void tbb::internal::task_stream< Levels >::push ( task *  source, int  level, FastRandom &  random  )

inline

Push a task into a lane.

Definition at line 101 of file task_stream.h.

                                                             {
        // Lane selection is random. Each thread should keep a separate seed value.
        unsigned idx;
        for( ; ; ) {
            idx = random.get() & (N-1);
            spin_mutex::scoped_lock lock;
            if( lock.try_acquire(lanes[level][idx].my_mutex) ) {
                lanes[level][idx].my_queue.push_back(source);
                set_one_bit( population[level], idx ); //TODO: avoid atomic op if the bit is already set
                break;
            }
        }
    }

push() [2/2]

template<int Levels>

template<typename lane_selector_t >

void tbb::internal::task_stream< Levels >::push ( task *  source, int  level, const lane_selector_t &  next_lane  )

inline

Push a task into a lane. Lane selection is performed by passed functor.

Definition at line 206 of file task_stream_extended.h.

                                                                           {
        bool succeed = false;
        unsigned lane = 0;
        do {
            lane = next_lane( /*out_of=*/N );
            __TBB_ASSERT( lane < N, "Incorrect lane index." );
        } while( ! (succeed = try_push( source, level, lane )) );
    }

try_pop()

template<int Levels>

task* tbb::internal::task_stream< Levels >::try_pop ( int  level, unsigned  lane_idx  )

inline

Returns pointer to task on successful pop, otherwise - NULL.

Definition at line 216 of file task_stream_extended.h.

                                                  {
        __TBB_ASSERT( 0 <= level && level < Levels, "Incorrect lane level specified." );
        if( !is_bit_set( population[level], lane_idx ) )
            return NULL;
        task* result = NULL;
        lane_t& lane = lanes[level][lane_idx];
        spin_mutex::scoped_lock lock;
        if( lock.try_acquire( lane.my_mutex ) && !lane.my_queue.empty() ) {
            result = this->get_item( lane.my_queue );
            if( lane.my_queue.empty() )
                clear_one_bit( population[level], lane_idx );
        }
        return result;
    }

Referenced by tbb::internal::task_stream< num_priority_levels >::pop().

Here is the caller graph for this function:

try_push()

template<int Levels>

bool tbb::internal::task_stream< Levels >::try_push ( task *  source, int  level, unsigned  lane_idx  )

inline

Returns true on successful push, otherwise - false.

Definition at line 193 of file task_stream_extended.h.

                                                                {
        __TBB_ASSERT( 0 <= level && level < Levels, "Incorrect lane level specified." );
        spin_mutex::scoped_lock lock;
        if( lock.try_acquire( lanes[level][lane_idx].my_mutex ) ) {
            lanes[level][lane_idx].my_queue.push_back( source );
            set_one_bit( population[level], lane_idx ); // TODO: avoid atomic op if the bit is already set
            return true;
        }
        return false;
    }

Referenced by tbb::internal::task_stream< num_priority_levels >::push().

Here is the caller graph for this function:

Member Data Documentation

lanes

template<int Levels>

padded< lane_t > * tbb::internal::task_stream< Levels >::lanes

private

Definition at line 72 of file task_stream.h.

Referenced by tbb::internal::task_stream< num_priority_levels >::drain(), tbb::internal::task_stream< num_priority_levels >::initialize(), tbb::internal::task_stream< num_priority_levels >::pop(), tbb::internal::task_stream< num_priority_levels >::pop_specific(), tbb::internal::task_stream< num_priority_levels >::push(), tbb::internal::task_stream< num_priority_levels >::task_stream(), tbb::internal::task_stream< num_priority_levels >::try_pop(), tbb::internal::task_stream< num_priority_levels >::try_push(), and tbb::internal::task_stream< num_priority_levels >::~task_stream().

N

template<int Levels>

unsigned tbb::internal::task_stream< Levels >::N

private

Definition at line 73 of file task_stream.h.

Referenced by tbb::internal::task_stream< num_priority_levels >::drain(), tbb::internal::task_stream< num_priority_levels >::initialize(), tbb::internal::task_stream< num_priority_levels >::pop(), tbb::internal::task_stream< num_priority_levels >::pop_specific(), and tbb::internal::task_stream< num_priority_levels >::push().

population

template<int Levels>

population_t tbb::internal::task_stream< Levels >::population

private

Definition at line 71 of file task_stream.h.

Referenced by tbb::internal::task_stream< num_priority_levels >::drain(), tbb::internal::task_stream< num_priority_levels >::empty(), tbb::internal::task_stream< num_priority_levels >::initialize(), tbb::internal::task_stream< num_priority_levels >::pop(), tbb::internal::task_stream< num_priority_levels >::pop_specific(), tbb::internal::task_stream< num_priority_levels >::push(), tbb::internal::task_stream< num_priority_levels >::task_stream(), tbb::internal::task_stream< num_priority_levels >::try_pop(), and tbb::internal::task_stream< num_priority_levels >::try_push().

---

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

• task_stream.h
• task_stream_extended.h