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;
}
}
tbb::internal::task_stream::lanes
padded< lane_t > * lanes[Levels]
Definition: task_stream.h:72
tbb::internal::task_stream::population
population_t population[Levels]
Definition: task_stream.h:71
tbb::internal::task_stream::N
unsigned N
Definition: task_stream.h:73

References tbb::internal::task_stream< Levels >::lanes, and tbb::internal::task_stream< Levels >::population.

## ~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];
}

References tbb::internal::task_stream< Levels >::lanes.

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

References tbb::internal::task_stream< Levels >::lanes, and tbb::internal::task_stream< Levels >::population.

## ~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];
}

References tbb::internal::task_stream< Levels >::lanes.

## 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;
}
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
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
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::is_bit_set
bool is_bit_set(population_t val, int pos)
Definition: task_stream.h:61
tbb::internal::clear_one_bit
void clear_one_bit(population_t &dest, int pos)
Definition: task_stream.h:55
tbb::spin_mutex::scoped_lock
friend class scoped_lock
Definition: spin_mutex.h:179
tbb::internal::task_stream::lane_t
queue_and_mutex< task *, spin_mutex > lane_t
Definition: task_stream.h:70

References __TBB_ASSERT, tbb::internal::clear_one_bit(), tbb::internal::is_bit_set(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::queue_and_mutex< T, mutex_t >::my_mutex, tbb::internal::queue_and_mutex< T, mutex_t >::my_queue, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call 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;
}

References __TBB_ASSERT, tbb::internal::clear_one_bit(), tbb::internal::is_bit_set(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::queue_and_mutex< T, mutex_t >::my_mutex, tbb::internal::queue_and_mutex< T, mutex_t >::my_queue, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call graph for this function:

## 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];
}

References tbb::internal::task_stream< Levels >::population.

Referenced by tbb::internal::task_stream< Levels >::pop(), and tbb::internal::task_stream< Levels >::pop_specific().

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

References tbb::internal::task_stream< Levels >::population.

## 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 );
}
}
__TBB_Log2
intptr_t __TBB_Log2(uintptr_t x)
Definition: tbb_machine.h:863
tbb::internal::population_t
uintptr_t population_t
Definition: task_stream.h:46

References __TBB_ASSERT, __TBB_Log2(), tbb::internal::task_stream< Levels >::lanes, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call 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 );
}
}

References __TBB_ASSERT, __TBB_Log2(), tbb::internal::task_stream< Levels >::lanes, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call graph for this function:

## 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;
}
__TBB_ISOLATION_EXPR
#define __TBB_ISOLATION_EXPR(isolation)
Definition: scheduler_common.h:67

References __TBB_ASSERT, __TBB_ISOLATION_EXPR, tbb::internal::task_prefix::isolation, and tbb::task::prefix().

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

Here is the call graph for this function:
Here is the caller graph for this function:

## pop() [1/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;
}
tbb::internal::task_stream::try_pop
task * try_pop(int level, unsigned lane_idx)
Returns pointer to task on successful pop, otherwise - NULL.
Definition: task_stream_extended.h:216
tbb::internal::task_stream::empty
bool empty(int level)
Checks existence of a task.
Definition: task_stream.h:138

References __TBB_ASSERT, tbb::internal::task_stream< Levels >::empty(), tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::try_pop().

Here is the call graph for this function:

## pop() [2/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;
}

References tbb::internal::clear_one_bit(), tbb::internal::is_bit_set(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::queue_and_mutex< T, mutex_t >::my_mutex, tbb::internal::queue_and_mutex< T, mutex_t >::my_queue, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call graph for this function:

## 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;
}
__TBB_ISOLATION_ARG
#define __TBB_ISOLATION_ARG(arg1, isolation)
Definition: scheduler_common.h:68
tbb::internal::task_stream::look_specific
task * look_specific(__TBB_ISOLATION_ARG(task_stream_base::lane_t::queue_base_t &queue, isolation_tag isolation))
Definition: task_stream_extended.h:245

References __TBB_ISOLATION_ARG, tbb::internal::clear_one_bit(), tbb::internal::task_stream< Levels >::empty(), tbb::internal::is_bit_set(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::task_stream< Levels >::look_specific(), tbb::internal::queue_and_mutex< T, mutex_t >::my_mutex, tbb::internal::queue_and_mutex< T, mutex_t >::my_queue, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::population.

Here is the call graph for this function:

## push() [1/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 )) );
}
tbb::internal::task_stream::try_push
bool try_push(task *source, int level, unsigned lane_idx)
Returns true on successful push, otherwise - false.
Definition: task_stream_extended.h:193

References __TBB_ASSERT, tbb::internal::task_stream< Levels >::N, and tbb::internal::task_stream< Levels >::try_push().

Here is the call graph for this function:

## push() [2/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;
}
}
}
tbb::internal::set_one_bit
void set_one_bit(population_t &dest, int pos)
Definition: task_stream.h:49

References tbb::internal::FastRandom::get(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::task_stream< Levels >::N, tbb::internal::task_stream< Levels >::population, and tbb::internal::set_one_bit().

Here is the call graph for this function:

## 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;
}
tbb::internal::task_stream_accessor::get_item
task * get_item(lane_t::queue_base_t &queue)
Definition: task_stream_extended.h:139

References __TBB_ASSERT, tbb::internal::clear_one_bit(), tbb::internal::task_stream_accessor< accessor >::get_item(), tbb::internal::is_bit_set(), tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::queue_and_mutex< T, mutex_t >::my_mutex, tbb::internal::queue_and_mutex< T, mutex_t >::my_queue, and tbb::internal::task_stream< Levels >::population.

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

Here is the call graph for this function:
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;
}

References __TBB_ASSERT, tbb::internal::task_stream< Levels >::lanes, lock, tbb::internal::task_stream< Levels >::population, and tbb::internal::set_one_bit().

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

Here is the call graph for this function:
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< Levels >::drain(), tbb::internal::task_stream< Levels >::initialize(), tbb::internal::task_stream< Levels >::pop(), tbb::internal::task_stream< Levels >::pop_specific(), tbb::internal::task_stream< Levels >::push(), tbb::internal::task_stream< Levels >::task_stream(), tbb::internal::task_stream< Levels >::try_pop(), tbb::internal::task_stream< Levels >::try_push(), and tbb::internal::task_stream< 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< Levels >::drain(), tbb::internal::task_stream< Levels >::initialize(), tbb::internal::task_stream< Levels >::pop(), tbb::internal::task_stream< Levels >::pop_specific(), and tbb::internal::task_stream< 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< Levels >::drain(), tbb::internal::task_stream< Levels >::empty(), tbb::internal::task_stream< Levels >::initialize(), tbb::internal::task_stream< Levels >::pop(), tbb::internal::task_stream< Levels >::pop_specific(), tbb::internal::task_stream< Levels >::push(), tbb::internal::task_stream< Levels >::task_stream(), tbb::internal::task_stream< Levels >::try_pop(), and tbb::internal::task_stream< Levels >::try_push().

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

- task_stream.h
- task_stream_extended.h

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.