task_stream_extended.h
Go to the documentation of this file.
/*
Copyright (c) 2005-2020 Intel Corporation
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef _TBB_task_stream_extended_H
#define _TBB_task_stream_extended_H
#if _TBB_task_stream_H
#error Either task_stream.h or this file can be included at the same time.
#endif
#if !__TBB_CPF_BUILD
#error This code bears a preview status until it proves its usefulness/performance suitability.
#endif
#include "tbb/tbb_stddef.h"
#include <deque>
#include <climits>
#include "tbb/atomic.h" // for __TBB_Atomic*
#include "tbb/spin_mutex.h"
#include "tbb/tbb_allocator.h"
#include "scheduler_common.h"
#include "tbb_misc.h" // for FastRandom
namespace tbb {
namespace internal {
template< typename T, typename mutex_t >
struct queue_and_mutex {
typedef std::deque< T, tbb_allocator<T> > queue_base_t;
queue_base_t my_queue;
mutex_t my_mutex;
queue_and_mutex () : my_queue(), my_mutex() {}
~queue_and_mutex () {}
};
typedef uintptr_t population_t;
const population_t one = 1;
inline void set_one_bit( population_t& dest, int pos ) {
__TBB_ASSERT( pos>=0, NULL );
__TBB_ASSERT( pos<int(sizeof(population_t)*CHAR_BIT), NULL );
__TBB_AtomicOR( &dest, one<<pos );
}
inline void clear_one_bit( population_t& dest, int pos ) {
__TBB_ASSERT( pos>=0, NULL );
__TBB_ASSERT( pos<int(sizeof(population_t)*CHAR_BIT), NULL );
__TBB_AtomicAND( &dest, ~(one<<pos) );
}
inline bool is_bit_set( population_t val, int pos ) {
__TBB_ASSERT( pos>=0, NULL );
__TBB_ASSERT( pos<int(sizeof(population_t)*CHAR_BIT), NULL );
return (val & (one<<pos)) != 0;
}
struct random_lane_selector :
#if __INTEL_COMPILER == 1110 || __INTEL_COMPILER == 1500
no_assign
#else
no_copy
#endif
{
random_lane_selector( FastRandom& random ) : my_random( random ) {}
unsigned operator()( unsigned out_of ) const {
__TBB_ASSERT( ((out_of-1) & out_of) == 0, "number of lanes is not power of two." );
return my_random.get() & (out_of-1);
}
private:
FastRandom& my_random;
};
struct lane_selector_base :
#if __INTEL_COMPILER == 1110 || __INTEL_COMPILER == 1500
no_assign
#else
no_copy
#endif
{
unsigned& my_previous;
lane_selector_base( unsigned& previous ) : my_previous( previous ) {}
};
struct subsequent_lane_selector : lane_selector_base {
subsequent_lane_selector( unsigned& previous ) : lane_selector_base( previous ) {}
unsigned operator()( unsigned out_of ) const {
__TBB_ASSERT( ((out_of-1) & out_of) == 0, "number of lanes is not power of two." );
return (++my_previous &= out_of-1);
}
};
struct preceding_lane_selector : lane_selector_base {
preceding_lane_selector( unsigned& previous ) : lane_selector_base( previous ) {}
unsigned operator()( unsigned out_of ) const {
__TBB_ASSERT( ((out_of-1) & out_of) == 0, "number of lanes is not power of two." );
return (--my_previous &= (out_of-1));
}
};
class task_stream_base : no_copy {
protected:
typedef queue_and_mutex <task*, spin_mutex> lane_t;
};
enum task_stream_accessor_type { front_accessor = 0, back_nonnull_accessor };
template<task_stream_accessor_type accessor>
class task_stream_accessor : public task_stream_base {
protected:
using task_stream_base::lane_t;
task* get_item( lane_t::queue_base_t& queue ) {
task* result = queue.front();
queue.pop_front();
return result;
}
};
template<>
class task_stream_accessor< back_nonnull_accessor > : public task_stream_base {
protected:
task* get_item( lane_t::queue_base_t& queue ) {
task* result = NULL;
do {
result = queue.back();
queue.pop_back();
} while( !result && !queue.empty() );
return result;
}
};
template<int Levels, task_stream_accessor_type accessor>
class task_stream : public task_stream_accessor< accessor > {
typedef typename task_stream_accessor<accessor>::lane_t lane_t;
population_t population[Levels];
padded<lane_t>* lanes[Levels];
unsigned N;
public:
task_stream() : N() {
for(int level = 0; level < Levels; level++) {
population[level] = 0;
lanes[level] = NULL;
}
}
void initialize( unsigned n_lanes ) {
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 );
}
}
~task_stream() {
for(int level = 0; level < Levels; level++)
if (lanes[level]) delete[] lanes[level];
}
bool try_push( task* source, int level, unsigned lane_idx ) {
__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;
}
template<typename lane_selector_t>
void push( task* source, int level, const lane_selector_t& next_lane ) {
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 )) );
}
task* try_pop( int level, unsigned lane_idx ) {
__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;
}
template<typename lane_selector_t>
task* pop( int level, const lane_selector_t& next_lane ) {
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;
}
// TODO: unify '*_specific' logic with 'pop' methods above
task* look_specific( __TBB_ISOLATION_ARG(task_stream_base::lane_t::queue_base_t& queue, isolation_tag isolation) ) {
__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;
}
task* pop_specific( int level, __TBB_ISOLATION_ARG(unsigned& last_used_lane, isolation_tag isolation) ) {
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;
}
bool empty(int level) {
return !population[level];
}
intptr_t drain() {
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;
}
}; // task_stream
} // namespace internal
} // namespace tbb
#endif /* _TBB_task_stream_extended_H */
tbb::task::prefix
internal::task_prefix & prefix(internal::version_tag *=NULL) const
Get reference to corresponding task_prefix.
Definition: task.h:991
tbb::internal::padded
Pads type T to fill out to a multiple of cache line size.
Definition: tbb_stddef.h:261
tbb::internal::population_t
uintptr_t population_t
Definition: task_stream.h:46
scheduler_common.h
spin_mutex.h
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::internal::task_stream::N
unsigned N
Definition: task_stream.h:73
tbb::internal::task_stream::initialize
void initialize(unsigned n_lanes)
Definition: task_stream_extended.h:175
tbb::internal::queue_and_mutex::~queue_and_mutex
~queue_and_mutex()
Definition: task_stream_extended.h:59
__TBB_ISOLATION_EXPR
#define __TBB_ISOLATION_EXPR(isolation)
Definition: scheduler_common.h:67
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
tbb::internal::queue_and_mutex::queue_base_t
std::deque< T, tbb_allocator< T > > queue_base_t
Definition: task_stream_extended.h:53
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
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::task_stream::push
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.
Definition: task_stream_extended.h:206
tbb::internal::subsequent_lane_selector::operator()
unsigned operator()(unsigned out_of) const
Definition: task_stream_extended.h:112
tbb::internal::clear_one_bit
void clear_one_bit(population_t &dest, int pos)
Definition: task_stream.h:55
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::task_stream_base
Definition: task_stream_extended.h:126
tbb::internal::lane_selector_base
Definition: task_stream_extended.h:99
tbb::internal::task_stream::empty
bool empty(int level)
Checks existence of a task.
Definition: task_stream.h:138
tbb::internal::task_stream_base::lane_t
queue_and_mutex< task *, spin_mutex > lane_t
Definition: task_stream_extended.h:128
tbb::internal::task_stream
The container for "fairness-oriented" aka "enqueued" tasks.
Definition: task_stream.h:69
mutex_t
CRITICAL_SECTION mutex_t
Definition: ittnotify_config.h:209
tbb::internal::random_lane_selector::operator()
unsigned operator()(unsigned out_of) const
Definition: task_stream_extended.h:91
tbb::internal::preceding_lane_selector::preceding_lane_selector
preceding_lane_selector(unsigned &previous)
Definition: task_stream_extended.h:119
tbb::internal::task_stream::task_stream
task_stream()
Definition: task_stream_extended.h:168
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
tbb::internal::task_stream::population
population_t population[Levels]
Definition: task_stream.h:71
tbb::internal::queue_and_mutex::my_mutex
mutex_t my_mutex
Definition: task_stream.h:40
tbb::internal::task_stream::~task_stream
~task_stream()
Definition: task_stream_extended.h:187
tbb::internal::is_bit_set
bool is_bit_set(population_t val, int pos)
Definition: task_stream.h:61
tbb::internal::preceding_lane_selector
Definition: task_stream_extended.h:118
tbb::internal::queue_and_mutex::queue_and_mutex
queue_and_mutex()
Definition: task_stream_extended.h:58
__TBB_ISOLATION_ARG
#define __TBB_ISOLATION_ARG(arg1, isolation)
Definition: scheduler_common.h:68
__TBB_AtomicAND
void __TBB_AtomicAND(volatile void *operand, uintptr_t addend)
Definition: tbb_machine.h:891
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
atomic.h
tbb::internal::task_stream_accessor< back_nonnull_accessor >::get_item
task * get_item(lane_t::queue_base_t &queue)
Definition: task_stream_extended.h:149
tbb::internal::lane_selector_base::lane_selector_base
lane_selector_base(unsigned &previous)
Definition: task_stream_extended.h:107
tbb::internal::isolation_tag
intptr_t isolation_tag
A tag for task isolation.
Definition: task.h:132
tbb::internal::task_stream_accessor
Definition: task_stream_extended.h:136
tbb_misc.h
tbb::internal::task_stream_accessor_type
task_stream_accessor_type
Definition: task_stream_extended.h:131
tbb::internal::task_stream::lanes
padded< lane_t > * lanes[Levels]
Definition: task_stream.h:72
tbb::internal::one
const population_t one
Definition: task_stream.h:47
tbb::internal::subsequent_lane_selector
Definition: task_stream_extended.h:110
__TBB_AtomicOR
void __TBB_AtomicOR(volatile void *operand, uintptr_t addend)
Definition: tbb_machine.h:881
__TBB_Log2
intptr_t __TBB_Log2(uintptr_t x)
Definition: tbb_machine.h:863
tbb::internal::task_stream::lane_t
task_stream_accessor< accessor >::lane_t lane_t
Definition: task_stream_extended.h:162
tbb::internal::no_assign
Base class for types that should not be assigned.
Definition: tbb_stddef.h:322
tbb::internal::front_accessor
Definition: task_stream_extended.h:131
tbb::internal::queue_and_mutex::my_queue
queue_base_t my_queue
Definition: task_stream.h:39
tbb::internal::FastRandom::get
unsigned short get()
Get a random number.
Definition: tbb_misc.h:146
tbb::internal::random_lane_selector
Definition: task_stream_extended.h:83
tbb::internal::back_nonnull_accessor
Definition: task_stream_extended.h:131
tbb::internal::task_prefix::isolation
isolation_tag isolation
The tag used for task isolation.
Definition: task.h:209
tbb::task
Base class for user-defined tasks.
Definition: task.h:604
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::FastRandom
A fast random number generator.
Definition: tbb_misc.h:135
tbb_stddef.h
tbb::internal::task_stream::pop
task * pop(int level, const lane_selector_t &next_lane)
Definition: task_stream_extended.h:234
tbb::internal::lane_selector_base::my_previous
unsigned & my_previous
Definition: task_stream_extended.h:106
tbb::internal::subsequent_lane_selector::subsequent_lane_selector
subsequent_lane_selector(unsigned &previous)
Definition: task_stream_extended.h:111
tbb::internal::preceding_lane_selector::operator()
unsigned operator()(unsigned out_of) const
Definition: task_stream_extended.h:120
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
tbb_allocator.h
tbb::internal::queue_and_mutex
Essentially, this is just a pair of a queue and a mutex to protect the queue.
Definition: task_stream.h:36
tbb::internal::task_stream_accessor::get_item
task * get_item(lane_t::queue_base_t &queue)
Definition: task_stream_extended.h:139
tbb::internal::random_lane_selector::random_lane_selector
random_lane_selector(FastRandom &random)
Definition: task_stream_extended.h:90
tbb::internal::set_one_bit
void set_one_bit(population_t &dest, int pos)
Definition: task_stream.h:49
tbb::internal::task_stream::pop_specific
task * pop_specific(int level, __TBB_ISOLATION_ARG(unsigned &last_used_lane, isolation_tag isolation))
Try finding and popping a related task.
Definition: task_stream_extended.h:266
tbb::internal::task_stream::drain
intptr_t drain()
Destroys all remaining tasks in every lane. Returns the number of destroyed tasks.
Definition: task_stream_extended.h:296
tbb::internal::random_lane_selector::my_random
FastRandom & my_random
Definition: task_stream_extended.h:96

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.