task_stream.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_H
#define _TBB_task_stream_H
#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;
}
template<int Levels>
class task_stream : no_copy {
typedef queue_and_mutex <task*, spin_mutex> 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];
}
void push( task* source, int level, FastRandom& random ) {
// 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;
}
}
}
task* pop( int level, unsigned& last_used_lane ) {
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;
}
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(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_H */
atomic.h
spin_mutex.h
tbb_allocator.h
__TBB_Log2
intptr_t __TBB_Log2(uintptr_t x)
Definition: tbb_machine.h:863
__TBB_AtomicOR
void __TBB_AtomicOR(volatile void *operand, uintptr_t addend)
Definition: tbb_machine.h:881
__TBB_AtomicAND
void __TBB_AtomicAND(volatile void *operand, uintptr_t addend)
Definition: tbb_machine.h:891
tbb_stddef.h
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
scheduler_common.h
tbb_misc.h
mutex_t
CRITICAL_SECTION mutex_t
Definition: ittnotify_config.h:209
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
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
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::internal::population_t
uintptr_t population_t
Definition: task_stream.h:46
tbb::internal::set_one_bit
void set_one_bit(population_t &dest, int pos)
Definition: task_stream.h:49
tbb::internal::one
const population_t one
Definition: task_stream.h:47
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
tbb::task
Base class for user-defined tasks.
Definition: task.h:615
tbb::internal::padded
Pads type T to fill out to a multiple of cache line size.
Definition: tbb_stddef.h:261
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
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::queue_and_mutex::my_queue
queue_base_t my_queue
Definition: task_stream.h:39
tbb::internal::queue_and_mutex::queue_base_t
std::deque< T, tbb_allocator< T > > queue_base_t
Definition: task_stream.h:37
tbb::internal::queue_and_mutex::my_mutex
mutex_t my_mutex
Definition: task_stream.h:40
tbb::internal::queue_and_mutex::queue_and_mutex
queue_and_mutex()
Definition: task_stream.h:42
tbb::internal::queue_and_mutex::~queue_and_mutex
~queue_and_mutex()
Definition: task_stream.h:43
tbb::internal::task_stream
The container for "fairness-oriented" aka "enqueued" tasks.
Definition: task_stream.h:69
tbb::internal::task_stream::push
void push(task *source, int level, FastRandom &random)
Push a task into a lane.
Definition: task_stream.h:101
tbb::internal::task_stream::lanes
padded< lane_t > * lanes[Levels]
Definition: task_stream.h:72
tbb::internal::task_stream::initialize
void initialize(unsigned n_lanes)
Definition: task_stream.h:83
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
tbb::internal::task_stream::task_stream
task_stream()
Definition: task_stream.h:76
tbb::internal::task_stream::pop
task * pop(int level, unsigned &last_used_lane)
Try finding and popping a task.
Definition: task_stream.h:116
tbb::internal::task_stream::~task_stream
~task_stream()
Definition: task_stream.h:95
tbb::internal::task_stream::lane_t
queue_and_mutex< task *, spin_mutex > lane_t
Definition: task_stream.h:70
tbb::internal::task_stream::drain
intptr_t drain()
Destroys all remaining tasks in every lane. Returns the number of destroyed tasks.
Definition: task_stream.h:145
tbb::internal::task_stream::empty
bool empty(int level)
Checks existence of a task.
Definition: task_stream.h:138
tbb::internal::FastRandom
A fast random number generator.
Definition: tbb_misc.h:135
tbb::internal::FastRandom::get
unsigned short get()
Get a random number.
Definition: tbb_misc.h:146

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.