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...

Private Types
typedef queue_and_mutex< task *, spin_mutex >	lane_t

Private Attributes
population_t	population [Levels]

padded< lane_t > *	lanes [Levels]

unsigned	N

Additional Inherited Members
Private Member Functions inherited from tbb::internal::no_copy
	no_copy ()
	Allow default construction. More...

Detailed Description

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

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

Definition at line 73 of file task_stream.h.

Member Typedef Documentation

◆ lane_t

template<int Levels>

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

private

Definition at line 74 of file task_stream.h.

Constructor & Destructor Documentation

◆ task_stream()

template<int Levels>

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

inline

Definition at line 80 of file task_stream.h.

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

◆ ~task_stream()

template<int Levels>

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

inline

Definition at line 99 of file task_stream.h.

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

Member Function Documentation

◆ drain()

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 149 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:

◆ empty()

template<int Levels>

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

inline

Checks existence of a task.

Definition at line 142 of file task_stream.h.

                           {
         return !population[level];
     }

Referenced by tbb::internal::arena::has_enqueued_tasks(), tbb::internal::arena::is_out_of_work(), and tbb::internal::arena::restore_priority_if_need().

Here is the caller graph for this function:

◆ initialize()

template<int Levels>

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

inline

Definition at line 87 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:

◆ pop()

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

◆ push()

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