arena.cpp
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.
*/
#include "tbb/global_control.h" // thread_stack_size
#include "scheduler.h"
#include "governor.h"
#include "arena.h"
#include "itt_notify.h"
#include "semaphore.h"
#include "tbb/internal/_flow_graph_impl.h"
#include <functional>
#if __TBB_STATISTICS_STDOUT
#include <cstdio>
#endif
namespace tbb {
namespace internal {
#if __TBB_NUMA_SUPPORT
class numa_binding_observer : public tbb::task_scheduler_observer {
int my_numa_node_id;
binding_handler* binding_handler_ptr;
public:
numa_binding_observer( task_arena* ta, int numa_id, int num_slots )
: task_scheduler_observer(*ta)
, my_numa_node_id(numa_id)
, binding_handler_ptr(tbb::internal::construct_binding_handler(num_slots))
{}
void on_scheduler_entry( bool ) __TBB_override {
tbb::internal::bind_thread_to_node(
binding_handler_ptr, this_task_arena::current_thread_index(), my_numa_node_id);
}
void on_scheduler_exit( bool ) __TBB_override {
tbb::internal::restore_affinity_mask(binding_handler_ptr, this_task_arena::current_thread_index());
}
~numa_binding_observer(){
tbb::internal::destroy_binding_handler(binding_handler_ptr);
}
};
numa_binding_observer* construct_binding_observer( tbb::interface7::task_arena* ta,
int numa_id, int num_slots ) {
numa_binding_observer* binding_observer = NULL;
// numa_topology initialization will be lazily performed inside nodes_count() call
if (numa_id >= 0 && numa_topology::nodes_count() > 1) {
binding_observer = new numa_binding_observer(ta, numa_id, num_slots);
__TBB_ASSERT(binding_observer, "Failure during NUMA binding observer allocation and construction");
binding_observer->observe(true);
}
return binding_observer;
}
void destroy_binding_observer( numa_binding_observer* binding_observer ) {
__TBB_ASSERT(binding_observer, "Trying to deallocate NULL pointer");
binding_observer->observe(false);
delete binding_observer;
}
#endif
// put it here in order to enable compiler to inline it into arena::process and nested_arena_entry
void generic_scheduler::attach_arena( arena* a, size_t index, bool is_master ) {
__TBB_ASSERT( a->my_market == my_market, NULL );
my_arena = a;
my_arena_index = index;
my_arena_slot = a->my_slots + index;
attach_mailbox( affinity_id(index+1) );
if ( is_master && my_inbox.is_idle_state( true ) ) {
// Master enters an arena with its own task to be executed. It means that master is not
// going to enter stealing loop and take affinity tasks.
my_inbox.set_is_idle( false );
}
#if __TBB_TASK_GROUP_CONTEXT
// Context to be used by root tasks by default (if the user has not specified one).
if( !is_master )
my_dummy_task->prefix().context = a->my_default_ctx;
#endif /* __TBB_TASK_GROUP_CONTEXT */
#if __TBB_TASK_PRIORITY
// In the current implementation master threads continue processing even when
// there are other masters with higher priority. Only TBB worker threads are
// redistributed between arenas based on the latters' priority. Thus master
// threads use arena's top priority as a reference point (in contrast to workers
// that use my_market->my_global_top_priority).
if( is_master ) {
my_ref_top_priority = &a->my_top_priority;
my_ref_reload_epoch = &a->my_reload_epoch;
}
my_local_reload_epoch = *my_ref_reload_epoch;
__TBB_ASSERT( !my_offloaded_tasks, NULL );
#endif /* __TBB_TASK_PRIORITY */
}
inline static bool occupy_slot( generic_scheduler*& slot, generic_scheduler& s ) {
return !slot && as_atomic( slot ).compare_and_swap( &s, NULL ) == NULL;
}
size_t arena::occupy_free_slot_in_range( generic_scheduler& s, size_t lower, size_t upper ) {
if ( lower >= upper ) return out_of_arena;
// Start search for an empty slot from the one we occupied the last time
size_t index = s.my_arena_index;
if ( index < lower || index >= upper ) index = s.my_random.get() % (upper - lower) + lower;
__TBB_ASSERT( index >= lower && index < upper, NULL );
// Find a free slot
for ( size_t i = index; i < upper; ++i )
if ( occupy_slot(my_slots[i].my_scheduler, s) ) return i;
for ( size_t i = lower; i < index; ++i )
if ( occupy_slot(my_slots[i].my_scheduler, s) ) return i;
return out_of_arena;
}
template <bool as_worker>
size_t arena::occupy_free_slot( generic_scheduler& s ) {
// Firstly, masters try to occupy reserved slots
size_t index = as_worker ? out_of_arena : occupy_free_slot_in_range( s, 0, my_num_reserved_slots );
if ( index == out_of_arena ) {
// Secondly, all threads try to occupy all non-reserved slots
index = occupy_free_slot_in_range( s, my_num_reserved_slots, my_num_slots );
// Likely this arena is already saturated
if ( index == out_of_arena )
return out_of_arena;
}
ITT_NOTIFY(sync_acquired, my_slots + index);
atomic_update( my_limit, (unsigned)(index + 1), std::less<unsigned>() );
return index;
}
void arena::process( generic_scheduler& s ) {
__TBB_ASSERT( is_alive(my_guard), NULL );
__TBB_ASSERT( governor::is_set(&s), NULL );
__TBB_ASSERT( s.my_innermost_running_task == s.my_dummy_task, NULL );
__TBB_ASSERT( s.worker_outermost_level(), NULL );
__TBB_ASSERT( my_num_slots > 1, NULL );
size_t index = occupy_free_slot</*as_worker*/true>( s );
if ( index == out_of_arena )
goto quit;
__TBB_ASSERT( index >= my_num_reserved_slots, "Workers cannot occupy reserved slots" );
s.attach_arena( this, index, /*is_master*/false );
#if !__TBB_FP_CONTEXT
my_cpu_ctl_env.set_env();
#endif
#if __TBB_ARENA_OBSERVER
__TBB_ASSERT( !s.my_last_local_observer, "There cannot be notified local observers when entering arena" );
my_observers.notify_entry_observers( s.my_last_local_observer, /*worker=*/true );
#endif /* __TBB_ARENA_OBSERVER */
// Task pool can be marked as non-empty if the worker occupies the slot left by a master.
if ( s.my_arena_slot->task_pool != EmptyTaskPool ) {
__TBB_ASSERT( s.my_inbox.is_idle_state(false), NULL );
s.local_wait_for_all( *s.my_dummy_task, NULL );
__TBB_ASSERT( s.my_inbox.is_idle_state(true), NULL );
}
for ( ;; ) {
__TBB_ASSERT( s.my_innermost_running_task == s.my_dummy_task, NULL );
__TBB_ASSERT( s.worker_outermost_level(), NULL );
__TBB_ASSERT( is_alive(my_guard), NULL );
__TBB_ASSERT( s.is_quiescent_local_task_pool_reset(),
"Worker cannot leave arena while its task pool is not reset" );
__TBB_ASSERT( s.my_arena_slot->task_pool == EmptyTaskPool, "Empty task pool is not marked appropriately" );
// This check prevents relinquishing more than necessary workers because
// of the non-atomicity of the decision making procedure
if ( is_recall_requested() )
break;
// Try to steal a task.
// Passing reference count is technically unnecessary in this context,
// but omitting it here would add checks inside the function.
task* t = s.receive_or_steal_task( __TBB_ISOLATION_ARG( s.my_dummy_task->prefix().ref_count, no_isolation ) );
if (t) {
// A side effect of receive_or_steal_task is that my_innermost_running_task can be set.
// But for the outermost dispatch loop it has to be a dummy task.
s.my_innermost_running_task = s.my_dummy_task;
s.local_wait_for_all(*s.my_dummy_task,t);
}
}
#if __TBB_ARENA_OBSERVER
my_observers.notify_exit_observers( s.my_last_local_observer, /*worker=*/true );
s.my_last_local_observer = NULL;
#endif /* __TBB_ARENA_OBSERVER */
#if __TBB_TASK_PRIORITY
if ( s.my_offloaded_tasks )
orphan_offloaded_tasks( s );
#endif /* __TBB_TASK_PRIORITY */
#if __TBB_STATISTICS
++s.my_counters.arena_roundtrips;
*my_slots[index].my_counters += s.my_counters;
s.my_counters.reset();
#endif /* __TBB_STATISTICS */
__TBB_store_with_release( my_slots[index].my_scheduler, (generic_scheduler*)NULL );
s.my_arena_slot = 0; // detached from slot
s.my_inbox.detach();
__TBB_ASSERT( s.my_inbox.is_idle_state(true), NULL );
__TBB_ASSERT( s.my_innermost_running_task == s.my_dummy_task, NULL );
__TBB_ASSERT( s.worker_outermost_level(), NULL );
__TBB_ASSERT( is_alive(my_guard), NULL );
quit:
// In contrast to earlier versions of TBB (before 3.0 U5) now it is possible
// that arena may be temporarily left unpopulated by threads. See comments in
// arena::on_thread_leaving() for more details.
on_thread_leaving<ref_worker>();
}
arena::arena ( market& m, unsigned num_slots, unsigned num_reserved_slots ) {
__TBB_ASSERT( !my_guard, "improperly allocated arena?" );
__TBB_ASSERT( sizeof(my_slots[0]) % NFS_GetLineSize()==0, "arena::slot size not multiple of cache line size" );
__TBB_ASSERT( (uintptr_t)this % NFS_GetLineSize()==0, "arena misaligned" );
#if __TBB_TASK_PRIORITY
__TBB_ASSERT( !my_reload_epoch && !my_orphaned_tasks && !my_skipped_fifo_priority, "New arena object is not zeroed" );
#endif /* __TBB_TASK_PRIORITY */
my_market = &m;
my_limit = 1;
// Two slots are mandatory: for the master, and for 1 worker (required to support starvation resistant tasks).
my_num_slots = num_arena_slots(num_slots);
my_num_reserved_slots = num_reserved_slots;
my_max_num_workers = num_slots-num_reserved_slots;
my_references = ref_external; // accounts for the master
#if __TBB_TASK_PRIORITY
my_bottom_priority = my_top_priority = normalized_normal_priority;
#endif /* __TBB_TASK_PRIORITY */
my_aba_epoch = m.my_arenas_aba_epoch;
#if __TBB_ARENA_OBSERVER
my_observers.my_arena = this;
#endif
#if __TBB_PREVIEW_RESUMABLE_TASKS
my_co_cache.init(4 * num_slots);
#endif
__TBB_ASSERT ( my_max_num_workers <= my_num_slots, NULL );
// Construct slots. Mark internal synchronization elements for the tools.
for( unsigned i = 0; i < my_num_slots; ++i ) {
__TBB_ASSERT( !my_slots[i].my_scheduler && !my_slots[i].task_pool, NULL );
__TBB_ASSERT( !my_slots[i].task_pool_ptr, NULL );
__TBB_ASSERT( !my_slots[i].my_task_pool_size, NULL );
#if __TBB_PREVIEW_RESUMABLE_TASKS
__TBB_ASSERT( !my_slots[i].my_scheduler_is_recalled, NULL );
#endif
ITT_SYNC_CREATE(my_slots + i, SyncType_Scheduler, SyncObj_WorkerTaskPool);
mailbox(i+1).construct();
ITT_SYNC_CREATE(&mailbox(i+1), SyncType_Scheduler, SyncObj_Mailbox);
my_slots[i].hint_for_pop = i;
#if __TBB_PREVIEW_CRITICAL_TASKS
my_slots[i].hint_for_critical = i;
#endif
#if __TBB_STATISTICS
my_slots[i].my_counters = new ( NFS_Allocate(1, sizeof(statistics_counters), NULL) ) statistics_counters;
#endif /* __TBB_STATISTICS */
}
my_task_stream.initialize(my_num_slots);
ITT_SYNC_CREATE(&my_task_stream, SyncType_Scheduler, SyncObj_TaskStream);
#if __TBB_PREVIEW_CRITICAL_TASKS
my_critical_task_stream.initialize(my_num_slots);
ITT_SYNC_CREATE(&my_critical_task_stream, SyncType_Scheduler, SyncObj_CriticalTaskStream);
#endif
#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
my_local_concurrency_mode = false;
my_global_concurrency_mode = false;
#endif
#if !__TBB_FP_CONTEXT
my_cpu_ctl_env.get_env();
#endif
}
arena& arena::allocate_arena( market& m, unsigned num_slots, unsigned num_reserved_slots ) {
__TBB_ASSERT( sizeof(base_type) + sizeof(arena_slot) == sizeof(arena), "All arena data fields must go to arena_base" );
__TBB_ASSERT( sizeof(base_type) % NFS_GetLineSize() == 0, "arena slots area misaligned: wrong padding" );
__TBB_ASSERT( sizeof(mail_outbox) == NFS_MaxLineSize, "Mailbox padding is wrong" );
size_t n = allocation_size(num_arena_slots(num_slots));
unsigned char* storage = (unsigned char*)NFS_Allocate( 1, n, NULL );
// Zero all slots to indicate that they are empty
memset( storage, 0, n );
return *new( storage + num_arena_slots(num_slots) * sizeof(mail_outbox) ) arena(m, num_slots, num_reserved_slots);
}
void arena::free_arena () {
__TBB_ASSERT( is_alive(my_guard), NULL );
__TBB_ASSERT( !my_references, "There are threads in the dying arena" );
__TBB_ASSERT( !my_num_workers_requested && !my_num_workers_allotted, "Dying arena requests workers" );
__TBB_ASSERT( my_pool_state == SNAPSHOT_EMPTY || !my_max_num_workers, "Inconsistent state of a dying arena" );
#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
__TBB_ASSERT( !my_global_concurrency_mode, NULL );
#endif
#if !__TBB_STATISTICS_EARLY_DUMP
GATHER_STATISTIC( dump_arena_statistics() );
#endif
poison_value( my_guard );
intptr_t drained = 0;
for ( unsigned i = 0; i < my_num_slots; ++i ) {
__TBB_ASSERT( !my_slots[i].my_scheduler, "arena slot is not empty" );
// TODO: understand the assertion and modify
// __TBB_ASSERT( my_slots[i].task_pool == EmptyTaskPool, NULL );
__TBB_ASSERT( my_slots[i].head == my_slots[i].tail, NULL ); // TODO: replace by is_quiescent_local_task_pool_empty
my_slots[i].free_task_pool();
#if __TBB_STATISTICS
NFS_Free( my_slots[i].my_counters );
#endif /* __TBB_STATISTICS */
drained += mailbox(i+1).drain();
}
__TBB_ASSERT( my_task_stream.drain()==0, "Not all enqueued tasks were executed");
#if __TBB_PREVIEW_RESUMABLE_TASKS
// Cleanup coroutines/schedulers cache
my_co_cache.cleanup();
#endif
#if __TBB_PREVIEW_CRITICAL_TASKS
__TBB_ASSERT( my_critical_task_stream.drain()==0, "Not all critical tasks were executed");
#endif
#if __TBB_COUNT_TASK_NODES
my_market->update_task_node_count( -drained );
#endif /* __TBB_COUNT_TASK_NODES */
// remove an internal reference
my_market->release( /*is_public=*/false, /*blocking_terminate=*/false );
#if __TBB_TASK_GROUP_CONTEXT
__TBB_ASSERT( my_default_ctx, "Master thread never entered the arena?" );
my_default_ctx->~task_group_context();
NFS_Free(my_default_ctx);
#endif /* __TBB_TASK_GROUP_CONTEXT */
#if __TBB_ARENA_OBSERVER
if ( !my_observers.empty() )
my_observers.clear();
#endif /* __TBB_ARENA_OBSERVER */
void* storage = &mailbox(my_num_slots);
__TBB_ASSERT( my_references == 0, NULL );
__TBB_ASSERT( my_pool_state == SNAPSHOT_EMPTY || !my_max_num_workers, NULL );
this->~arena();
#if TBB_USE_ASSERT > 1
memset( storage, 0, allocation_size(my_num_slots) );
#endif /* TBB_USE_ASSERT */
NFS_Free( storage );
}
#if __TBB_STATISTICS
void arena::dump_arena_statistics () {
statistics_counters total;
for( unsigned i = 0; i < my_num_slots; ++i ) {
#if __TBB_STATISTICS_EARLY_DUMP
generic_scheduler* s = my_slots[i].my_scheduler;
if ( s )
*my_slots[i].my_counters += s->my_counters;
#else
__TBB_ASSERT( !my_slots[i].my_scheduler, NULL );
#endif
if ( i != 0 ) {
total += *my_slots[i].my_counters;
dump_statistics( *my_slots[i].my_counters, i );
}
}
dump_statistics( *my_slots[0].my_counters, 0 );
#if __TBB_STATISTICS_STDOUT
#if !__TBB_STATISTICS_TOTALS_ONLY
printf( "----------------------------------------------\n" );
#endif
dump_statistics( total, workers_counters_total );
total += *my_slots[0].my_counters;
dump_statistics( total, arena_counters_total );
#if !__TBB_STATISTICS_TOTALS_ONLY
printf( "==============================================\n" );
#endif
#endif /* __TBB_STATISTICS_STDOUT */
}
#endif /* __TBB_STATISTICS */
#if __TBB_TASK_PRIORITY
// The method inspects a scheduler to determine:
// 1. if it has tasks that can be retrieved and executed (via the return value);
// 2. if it has any tasks at all, including those of lower priority (via tasks_present);
// 3. if it is able to work with enqueued tasks (via dequeuing_possible).
inline bool arena::may_have_tasks ( generic_scheduler* s, bool& tasks_present, bool& dequeuing_possible ) {
if ( !s || s->my_arena != this )
return false;
dequeuing_possible |= s->worker_outermost_level();
if ( s->my_pool_reshuffling_pending ) {
// This primary task pool is nonempty and may contain tasks at the current
// priority level. Its owner is winnowing lower priority tasks at the moment.
tasks_present = true;
return true;
}
if ( s->my_offloaded_tasks ) {
tasks_present = true;
if ( s->my_local_reload_epoch < *s->my_ref_reload_epoch ) {
// This scheduler's offload area is nonempty and may contain tasks at the
// current priority level.
return true;
}
}
return false;
}
void arena::orphan_offloaded_tasks(generic_scheduler& s) {
__TBB_ASSERT( s.my_offloaded_tasks, NULL );
GATHER_STATISTIC( ++s.my_counters.prio_orphanings );
++my_abandonment_epoch;
__TBB_ASSERT( s.my_offloaded_task_list_tail_link && !*s.my_offloaded_task_list_tail_link, NULL );
task* orphans;
do {
orphans = const_cast<task*>(my_orphaned_tasks);
*s.my_offloaded_task_list_tail_link = orphans;
} while ( as_atomic(my_orphaned_tasks).compare_and_swap(s.my_offloaded_tasks, orphans) != orphans );
s.my_offloaded_tasks = NULL;
#if TBB_USE_ASSERT
s.my_offloaded_task_list_tail_link = NULL;
#endif /* TBB_USE_ASSERT */
}
#endif /* __TBB_TASK_PRIORITY */
bool arena::has_enqueued_tasks() {
// Look for enqueued tasks at all priority levels
for ( int p = 0; p < num_priority_levels; ++p )
if ( !my_task_stream.empty(p) )
return true;
return false;
}
void arena::restore_priority_if_need() {
// Check for the presence of enqueued tasks "lost" on some of
// priority levels because updating arena priority and switching
// arena into "populated" (FULL) state happen non-atomically.
// Imposing atomicity would require task::enqueue() to use a lock,
// which is unacceptable.
if ( has_enqueued_tasks() ) {
advertise_new_work<work_enqueued>();
#if __TBB_TASK_PRIORITY
// update_arena_priority() expects non-zero arena::my_num_workers_requested,
// so must be called after advertise_new_work<work_enqueued>()
for ( int p = 0; p < num_priority_levels; ++p )
if ( !my_task_stream.empty(p) ) {
if ( p < my_bottom_priority || p > my_top_priority )
my_market->update_arena_priority(*this, p);
}
#endif
}
}
bool arena::is_out_of_work() {
// TODO: rework it to return at least a hint about where a task was found; better if the task itself.
for(;;) {
pool_state_t snapshot = my_pool_state;
switch( snapshot ) {
case SNAPSHOT_EMPTY:
return true;
case SNAPSHOT_FULL: {
// Use unique id for "busy" in order to avoid ABA problems.
const pool_state_t busy = pool_state_t(&busy);
// Request permission to take snapshot
if( my_pool_state.compare_and_swap( busy, SNAPSHOT_FULL )==SNAPSHOT_FULL ) {
// Got permission. Take the snapshot.
// NOTE: This is not a lock, as the state can be set to FULL at
// any moment by a thread that spawns/enqueues new task.
size_t n = my_limit;
// Make local copies of volatile parameters. Their change during
// snapshot taking procedure invalidates the attempt, and returns
// this thread into the dispatch loop.
#if __TBB_TASK_PRIORITY
uintptr_t reload_epoch = __TBB_load_with_acquire( my_reload_epoch );
intptr_t top_priority = my_top_priority;
// Inspect primary task pools first
#endif /* __TBB_TASK_PRIORITY */
size_t k;
for( k=0; k<n; ++k ) {
if( my_slots[k].task_pool != EmptyTaskPool &&
__TBB_load_relaxed(my_slots[k].head) < __TBB_load_relaxed(my_slots[k].tail) )
{
// k-th primary task pool is nonempty and does contain tasks.
break;
}
if( my_pool_state!=busy )
return false; // the work was published
}
__TBB_ASSERT( k <= n, NULL );
bool work_absent = k == n;
#if __TBB_PREVIEW_CRITICAL_TASKS
bool no_critical_tasks = my_critical_task_stream.empty(0);
work_absent &= no_critical_tasks;
#endif
#if __TBB_TASK_PRIORITY
// Variable tasks_present indicates presence of tasks at any priority
// level, while work_absent refers only to the current priority.
bool tasks_present = !work_absent || my_orphaned_tasks;
bool dequeuing_possible = false;
if ( work_absent ) {
// Check for the possibility that recent priority changes
// brought some tasks to the current priority level
uintptr_t abandonment_epoch = my_abandonment_epoch;
// Master thread's scheduler needs special handling as it
// may be destroyed at any moment (workers' schedulers are
// guaranteed to be alive while at least one thread is in arena).
// The lock below excludes concurrency with task group state change
// propagation and guarantees lifetime of the master thread.
the_context_state_propagation_mutex.lock();
work_absent = !may_have_tasks( my_slots[0].my_scheduler, tasks_present, dequeuing_possible );
the_context_state_propagation_mutex.unlock();
// The following loop is subject to data races. While k-th slot's
// scheduler is being examined, corresponding worker can either
// leave to RML or migrate to another arena.
// But the races are not prevented because all of them are benign.
// First, the code relies on the fact that worker thread's scheduler
// object persists until the whole library is deinitialized.
// Second, in the worst case the races can only cause another
// round of stealing attempts to be undertaken. Introducing complex
// synchronization into this coldest part of the scheduler's control
// flow does not seem to make sense because it both is unlikely to
// ever have any observable performance effect, and will require
// additional synchronization code on the hotter paths.
for( k = 1; work_absent && k < n; ++k ) {
if( my_pool_state!=busy )
return false; // the work was published
work_absent = !may_have_tasks( my_slots[k].my_scheduler, tasks_present, dequeuing_possible );
}
// Preclude premature switching arena off because of a race in the previous loop.
work_absent = work_absent
&& !__TBB_load_with_acquire(my_orphaned_tasks)
&& abandonment_epoch == my_abandonment_epoch;
}
#endif /* __TBB_TASK_PRIORITY */
// Test and test-and-set.
if( my_pool_state==busy ) {
#if __TBB_TASK_PRIORITY
bool no_fifo_tasks = my_task_stream.empty(top_priority);
work_absent = work_absent && (!dequeuing_possible || no_fifo_tasks)
&& top_priority == my_top_priority && reload_epoch == my_reload_epoch;
#else
bool no_fifo_tasks = my_task_stream.empty(0);
work_absent = work_absent && no_fifo_tasks;
#endif /* __TBB_TASK_PRIORITY */
if( work_absent ) {
#if __TBB_TASK_PRIORITY
if ( top_priority > my_bottom_priority ) {
if ( my_market->lower_arena_priority(*this, top_priority - 1, reload_epoch)
&& !my_task_stream.empty(top_priority) )
{
atomic_update( my_skipped_fifo_priority, top_priority, std::less<intptr_t>());
}
}
else if ( !tasks_present && !my_orphaned_tasks && no_fifo_tasks ) {
#endif /* __TBB_TASK_PRIORITY */
// save current demand value before setting SNAPSHOT_EMPTY,
// to avoid race with advertise_new_work.
int current_demand = (int)my_max_num_workers;
if( my_pool_state.compare_and_swap( SNAPSHOT_EMPTY, busy )==busy ) {
// This thread transitioned pool to empty state, and thus is
// responsible for telling the market that there is no work to do.
my_market->adjust_demand( *this, -current_demand );
restore_priority_if_need();
return true;
}
return false;
#if __TBB_TASK_PRIORITY
}
#endif /* __TBB_TASK_PRIORITY */
}
// Undo previous transition SNAPSHOT_FULL-->busy, unless another thread undid it.
my_pool_state.compare_and_swap( SNAPSHOT_FULL, busy );
}
}
return false;
}
default:
// Another thread is taking a snapshot.
return false;
}
}
}
#if __TBB_COUNT_TASK_NODES
intptr_t arena::workers_task_node_count() {
intptr_t result = 0;
for( unsigned i = 1; i < my_num_slots; ++i ) {
generic_scheduler* s = my_slots[i].my_scheduler;
if( s )
result += s->my_task_node_count;
}
return result;
}
#endif /* __TBB_COUNT_TASK_NODES */
void arena::enqueue_task( task& t, intptr_t prio, FastRandom &random )
{
#if __TBB_RECYCLE_TO_ENQUEUE
__TBB_ASSERT( t.state()==task::allocated || t.state()==task::to_enqueue, "attempt to enqueue task with inappropriate state" );
#else
__TBB_ASSERT( t.state()==task::allocated, "attempt to enqueue task that is not in 'allocated' state" );
#endif
t.prefix().state = task::ready;
t.prefix().extra_state |= es_task_enqueued; // enqueued task marker
#if TBB_USE_ASSERT
if( task* parent = t.parent() ) {
internal::reference_count ref_count = parent->prefix().ref_count;
__TBB_ASSERT( ref_count!=0, "attempt to enqueue task whose parent has a ref_count==0 (forgot to set_ref_count?)" );
__TBB_ASSERT( ref_count>0, "attempt to enqueue task whose parent has a ref_count<0" );
parent->prefix().extra_state |= es_ref_count_active;
}
__TBB_ASSERT(t.prefix().affinity==affinity_id(0), "affinity is ignored for enqueued tasks");
#endif /* TBB_USE_ASSERT */
#if __TBB_PREVIEW_CRITICAL_TASKS
#if __TBB_TASK_PRIORITY
bool is_critical = internal::is_critical( t ) || prio == internal::priority_critical;
#else
bool is_critical = internal::is_critical( t );
#endif
if( is_critical ) {
// TODO: consider using of 'scheduler::handled_as_critical'
internal::make_critical( t );
generic_scheduler* s = governor::local_scheduler_if_initialized();
ITT_NOTIFY(sync_releasing, &my_critical_task_stream);
if( s && s->my_arena_slot ) {
// Scheduler is initialized and it is attached to the arena,
// propagate isolation level to critical task
#if __TBB_TASK_ISOLATION
t.prefix().isolation = s->my_innermost_running_task->prefix().isolation;
#endif
unsigned& lane = s->my_arena_slot->hint_for_critical;
my_critical_task_stream.push( &t, 0, tbb::internal::subsequent_lane_selector(lane) );
} else {
// Either scheduler is not initialized or it is not attached to the arena
// use random lane for the task
my_critical_task_stream.push( &t, 0, internal::random_lane_selector(random) );
}
advertise_new_work<work_spawned>();
return;
}
#endif /* __TBB_PREVIEW_CRITICAL_TASKS */
ITT_NOTIFY(sync_releasing, &my_task_stream);
#if __TBB_TASK_PRIORITY
intptr_t p = prio ? normalize_priority(priority_t(prio)) : normalized_normal_priority;
assert_priority_valid(p);
#if __TBB_PREVIEW_CRITICAL_TASKS && __TBB_CPF_BUILD
my_task_stream.push( &t, p, internal::random_lane_selector(random) );
#else
my_task_stream.push( &t, p, random );
#endif
if ( p != my_top_priority )
my_market->update_arena_priority( *this, p );
#else /* !__TBB_TASK_PRIORITY */
__TBB_ASSERT_EX(prio == 0, "the library is not configured to respect the task priority");
#if __TBB_PREVIEW_CRITICAL_TASKS && __TBB_CPF_BUILD
my_task_stream.push( &t, 0, internal::random_lane_selector(random) );
#else
my_task_stream.push( &t, 0, random );
#endif
#endif /* !__TBB_TASK_PRIORITY */
advertise_new_work<work_enqueued>();
#if __TBB_TASK_PRIORITY
if ( p != my_top_priority )
my_market->update_arena_priority( *this, p );
#endif /* __TBB_TASK_PRIORITY */
}
class nested_arena_context : no_copy {
public:
nested_arena_context(generic_scheduler *s, arena* a, size_t slot_index, bool type, bool same)
: my_scheduler(*s), my_orig_ctx(NULL), same_arena(same) {
if (same_arena) {
my_orig_state.my_properties = my_scheduler.my_properties;
my_orig_state.my_innermost_running_task = my_scheduler.my_innermost_running_task;
mimic_outermost_level(a, type);
} else {
my_orig_state = *s;
#if __TBB_PREVIEW_RESUMABLE_TASKS
my_scheduler.my_properties.genuine = true;
my_scheduler.my_current_is_recalled = NULL;
#endif
mimic_outermost_level(a, type);
s->nested_arena_entry(a, slot_index);
}
}
~nested_arena_context() {
#if __TBB_TASK_GROUP_CONTEXT
my_scheduler.my_dummy_task->prefix().context = my_orig_ctx; // restore context of dummy task
#endif
if (same_arena) {
my_scheduler.my_properties = my_orig_state.my_properties;
my_scheduler.my_innermost_running_task = my_orig_state.my_innermost_running_task;
} else {
my_scheduler.nested_arena_exit();
static_cast<scheduler_state&>(my_scheduler) = my_orig_state; // restore arena settings
#if __TBB_TASK_PRIORITY
my_scheduler.my_local_reload_epoch = *my_orig_state.my_ref_reload_epoch;
#endif
governor::assume_scheduler(&my_scheduler);
}
}
private:
generic_scheduler &my_scheduler;
scheduler_state my_orig_state;
task_group_context *my_orig_ctx;
const bool same_arena;
void mimic_outermost_level(arena* a, bool type) {
my_scheduler.my_properties.outermost = true;
my_scheduler.my_properties.type = type;
my_scheduler.my_innermost_running_task = my_scheduler.my_dummy_task;
#if __TBB_PREVIEW_CRITICAL_TASKS
my_scheduler.my_properties.has_taken_critical_task = false;
#endif
#if __TBB_TASK_GROUP_CONTEXT
// Save dummy's context and replace it by arena's context
my_orig_ctx = my_scheduler.my_dummy_task->prefix().context;
my_scheduler.my_dummy_task->prefix().context = a->my_default_ctx;
#endif
}
};
void generic_scheduler::nested_arena_entry(arena* a, size_t slot_index) {
__TBB_ASSERT( is_alive(a->my_guard), NULL );
__TBB_ASSERT( a!=my_arena, NULL);
// overwrite arena settings
#if __TBB_TASK_PRIORITY
if ( my_offloaded_tasks )
my_arena->orphan_offloaded_tasks( *this );
my_offloaded_tasks = NULL;
#endif /* __TBB_TASK_PRIORITY */
attach_arena( a, slot_index, /*is_master*/true );
__TBB_ASSERT( my_arena == a, NULL );
governor::assume_scheduler( this );
// TODO? ITT_NOTIFY(sync_acquired, a->my_slots + index);
// TODO: it requires market to have P workers (not P-1)
// TODO: a preempted worker should be excluded from assignment to other arenas e.g. my_slack--
if( !is_worker() && slot_index >= my_arena->my_num_reserved_slots )
my_arena->my_market->adjust_demand(*my_arena, -1);
#if __TBB_ARENA_OBSERVER
my_last_local_observer = 0; // TODO: try optimize number of calls
my_arena->my_observers.notify_entry_observers( my_last_local_observer, /*worker=*/false );
#endif
#if __TBB_PREVIEW_RESUMABLE_TASKS
my_wait_task = NULL;
#endif
}
void generic_scheduler::nested_arena_exit() {
#if __TBB_ARENA_OBSERVER
my_arena->my_observers.notify_exit_observers( my_last_local_observer, /*worker=*/false );
#endif /* __TBB_ARENA_OBSERVER */
#if __TBB_TASK_PRIORITY
if ( my_offloaded_tasks )
my_arena->orphan_offloaded_tasks( *this );
#endif
if( !is_worker() && my_arena_index >= my_arena->my_num_reserved_slots )
my_arena->my_market->adjust_demand(*my_arena, 1);
// Free the master slot.
__TBB_ASSERT(my_arena->my_slots[my_arena_index].my_scheduler, "A slot is already empty");
__TBB_store_with_release(my_arena->my_slots[my_arena_index].my_scheduler, (generic_scheduler*)NULL);
my_arena->my_exit_monitors.notify_one(); // do not relax!
}
void generic_scheduler::wait_until_empty() {
my_dummy_task->prefix().ref_count++; // prevents exit from local_wait_for_all when local work is done enforcing the stealing
while( my_arena->my_pool_state != arena::SNAPSHOT_EMPTY )
local_wait_for_all(*my_dummy_task, NULL);
my_dummy_task->prefix().ref_count--;
}
#if __TBB_PREVIEW_RESUMABLE_TASKS
class resume_task : public task {
generic_scheduler& my_target;
public:
resume_task(generic_scheduler& target) : my_target(target) {}
task* execute() __TBB_override {
generic_scheduler* s = governor::local_scheduler_if_initialized();
__TBB_ASSERT(s, NULL);
if (s->prepare_resume(my_target)) {
s->resume(my_target);
} else {
__TBB_ASSERT(prefix().state == task::executing, NULL);
// Request the dispatch loop to exit (because we in a coroutine on the outermost level).
prefix().state = task::to_resume;
}
return NULL;
}
};
static generic_scheduler& create_coroutine(generic_scheduler& curr) {
// We may have some coroutines cached
generic_scheduler* co_sched = curr.my_arena->my_co_cache.pop();
if (!co_sched) {
// TODO: avoid setting/unsetting the scheduler.
governor::assume_scheduler(NULL);
co_sched = generic_scheduler::create_worker(*curr.my_market, curr.my_arena_index, /* genuine = */ false);
governor::assume_scheduler(&curr);
// Prepare newly created scheduler
co_sched->my_arena = curr.my_arena;
}
// Prepare scheduler (general)
co_sched->my_dummy_task->prefix().context = co_sched->my_arena->my_default_ctx;
// Prolong the arena's lifetime until all coroutines is alive
// (otherwise the arena can be destroyed while some tasks are suspended).
co_sched->my_arena->my_references += arena::ref_external;
return *co_sched;
}
void internal_suspend(void* suspend_callback, void* user_callback) {
generic_scheduler& s = *governor::local_scheduler();
__TBB_ASSERT(s.my_arena_slot->my_scheduler_is_recalled != NULL, NULL);
bool is_recalled = *s.my_arena_slot->my_scheduler_is_recalled;
generic_scheduler& target = is_recalled ? *s.my_arena_slot->my_scheduler : create_coroutine(s);
generic_scheduler::callback_t callback = {
(generic_scheduler::suspend_callback_t)suspend_callback, user_callback, &s };
target.set_post_resume_action(generic_scheduler::PRA_CALLBACK, &callback);
s.resume(target);
}
void internal_resume(task::suspend_point tag) {
generic_scheduler& s = *static_cast<generic_scheduler*>(tag);
task* t = new(&s.allocate_task(sizeof(resume_task), __TBB_CONTEXT_ARG(NULL, s.my_dummy_task->context()))) resume_task(s);
make_critical(*t);
// TODO: remove this work-around
// Prolong the arena's lifetime until all coroutines is alive
// (otherwise the arena can be destroyed while some tasks are suspended).
arena& a = *s.my_arena;
a.my_references += arena::ref_external;
a.my_critical_task_stream.push(t, 0, tbb::internal::random_lane_selector(s.my_random));
// Do not access 's' after that point.
a.advertise_new_work<arena::work_spawned>();
// Release our reference to my_arena.
a.on_thread_leaving<arena::ref_external>();
}
task::suspend_point internal_current_suspend_point() {
return governor::local_scheduler();
}
#endif /* __TBB_PREVIEW_RESUMABLE_TASKS */
} // namespace internal
} // namespace tbb
#include "scheduler_utility.h"
#include "tbb/task_arena.h" // task_arena_base
namespace tbb {
namespace interface7 {
namespace internal {
void task_arena_base::internal_initialize( ) {
governor::one_time_init();
if( my_max_concurrency < 1 )
#if __TBB_NUMA_SUPPORT
my_max_concurrency = tbb::internal::numa_topology::default_concurrency(numa_id());
#else /*__TBB_NUMA_SUPPORT*/
my_max_concurrency = (int)governor::default_num_threads();
#endif /*__TBB_NUMA_SUPPORT*/
__TBB_ASSERT( my_master_slots <= (unsigned)my_max_concurrency, "Number of slots reserved for master should not exceed arena concurrency");
arena* new_arena = market::create_arena( my_max_concurrency, my_master_slots, 0 );
// add an internal market reference; a public reference was added in create_arena
market &m = market::global_market( /*is_public=*/false );
// allocate default context for task_arena
#if __TBB_TASK_GROUP_CONTEXT
new_arena->my_default_ctx = new ( NFS_Allocate(1, sizeof(task_group_context), NULL) )
task_group_context( task_group_context::isolated, task_group_context::default_traits );
#if __TBB_FP_CONTEXT
new_arena->my_default_ctx->capture_fp_settings();
#endif
#endif /* __TBB_TASK_GROUP_CONTEXT */
// threads might race to initialize the arena
if(as_atomic(my_arena).compare_and_swap(new_arena, NULL) != NULL) {
__TBB_ASSERT(my_arena, NULL); // another thread won the race
// release public market reference
m.release( /*is_public=*/true, /*blocking_terminate=*/false );
new_arena->on_thread_leaving<arena::ref_external>(); // destroy unneeded arena
#if __TBB_TASK_GROUP_CONTEXT
spin_wait_while_eq(my_context, (task_group_context*)NULL);
#endif /*__TBB_TASK_GROUP_CONTEXT*/
#if __TBB_TASK_GROUP_CONTEXT || __TBB_NUMA_SUPPORT
} else {
#if __TBB_TASK_GROUP_CONTEXT
new_arena->my_default_ctx->my_version_and_traits |= my_version_and_traits & exact_exception_flag;
as_atomic(my_context) = new_arena->my_default_ctx;
#endif /*__TBB_TASK_GROUP_CONTEXT*/
#if __TBB_NUMA_SUPPORT
my_arena->my_numa_binding_observer = tbb::internal::construct_binding_observer(
static_cast<task_arena*>(this), numa_id(), my_arena->my_num_slots);
#endif /*__TBB_NUMA_SUPPORT*/
}
#endif /*__TBB_TASK_GROUP_CONTEXT || __TBB_NUMA_SUPPORT*/
// TODO: should it trigger automatic initialization of this thread?
governor::local_scheduler_weak();
}
void task_arena_base::internal_terminate( ) {
if( my_arena ) {// task_arena was initialized
#if __TBB_NUMA_SUPPORT
if( my_arena->my_numa_binding_observer != NULL ) {
tbb::internal::destroy_binding_observer(my_arena->my_numa_binding_observer);
my_arena->my_numa_binding_observer = NULL;
}
#endif /*__TBB_NUMA_SUPPORT*/
my_arena->my_market->release( /*is_public=*/true, /*blocking_terminate=*/false );
my_arena->on_thread_leaving<arena::ref_external>();
my_arena = 0;
#if __TBB_TASK_GROUP_CONTEXT
my_context = 0;
#endif
}
}
void task_arena_base::internal_attach( ) {
__TBB_ASSERT(!my_arena, NULL);
generic_scheduler* s = governor::local_scheduler_if_initialized();
if( s && s->my_arena ) {
// There is an active arena to attach to.
// It's still used by s, so won't be destroyed right away.
my_arena = s->my_arena;
__TBB_ASSERT( my_arena->my_references > 0, NULL );
my_arena->my_references += arena::ref_external;
#if __TBB_TASK_GROUP_CONTEXT
my_context = my_arena->my_default_ctx;
my_version_and_traits |= my_context->my_version_and_traits & exact_exception_flag;
#endif
my_master_slots = my_arena->my_num_reserved_slots;
my_max_concurrency = my_master_slots + my_arena->my_max_num_workers;
__TBB_ASSERT(arena::num_arena_slots(my_max_concurrency)==my_arena->my_num_slots, NULL);
// increases market's ref count for task_arena
market::global_market( /*is_public=*/true );
}
}
void task_arena_base::internal_enqueue( task& t, intptr_t prio ) const {
__TBB_ASSERT(my_arena, NULL);
generic_scheduler* s = governor::local_scheduler_weak(); // scheduler is only needed for FastRandom instance
__TBB_ASSERT(s, "Scheduler is not initialized"); // we allocated a task so can expect the scheduler
#if __TBB_TASK_GROUP_CONTEXT
// Is there a better place for checking the state of my_default_ctx?
__TBB_ASSERT(!(my_arena->my_default_ctx == t.prefix().context && my_arena->my_default_ctx->is_group_execution_cancelled()),
"The task will not be executed because default task_group_context of task_arena is cancelled. Has previously enqueued task thrown an exception?");
#endif
my_arena->enqueue_task( t, prio, s->my_random );
}
class delegated_task : public task {
internal::delegate_base & my_delegate;
concurrent_monitor & my_monitor;
task * my_root;
task* execute() __TBB_override {
generic_scheduler& s = *(generic_scheduler*)prefix().owner;
__TBB_ASSERT(s.outermost_level(), "expected to be enqueued and received on the outermost level");
struct outermost_context : internal::no_copy {
delegated_task * t;
generic_scheduler & s;
task * orig_dummy;
task_group_context * orig_ctx;
scheduler_properties orig_props;
outermost_context(delegated_task *_t, generic_scheduler &_s)
: t(_t), s(_s), orig_dummy(s.my_dummy_task), orig_props(s.my_properties) {
__TBB_ASSERT(s.my_innermost_running_task == t, NULL);
#if __TBB_TASK_GROUP_CONTEXT
orig_ctx = t->prefix().context;
t->prefix().context = s.my_arena->my_default_ctx;
#endif
// Mimics outermost master
s.my_dummy_task = t;
s.my_properties.type = scheduler_properties::master;
}
~outermost_context() {
#if __TBB_TASK_GROUP_CONTEXT
// Restore context for sake of registering potential exception
t->prefix().context = orig_ctx;
#endif
// Restore scheduler state
s.my_properties = orig_props;
s.my_dummy_task = orig_dummy;
}
} scope(this, s);
my_delegate();
return NULL;
}
~delegated_task() {
// potential exception was already registered. It must happen before the notification
__TBB_ASSERT(my_root->ref_count() == 2, NULL);
task_prefix& prefix = my_root->prefix();
#if __TBB_PREVIEW_RESUMABLE_TASKS
reference_count old_ref_count = __TBB_FetchAndStoreW(&prefix.ref_count, 1);
// Check if the scheduler was abandoned.
if (old_ref_count == internal::abandon_flag + 2) {
__TBB_ASSERT(prefix.abandoned_scheduler, NULL);
// The wait has been completed. Spawn a resume task.
tbb::task::resume(prefix.abandoned_scheduler);
}
#else
__TBB_store_with_release(prefix.ref_count, 1); // must precede the wakeup
#endif
my_monitor.notify(*this); // do not relax, it needs a fence!
}
public:
delegated_task( internal::delegate_base & d, concurrent_monitor & s, task * t )
: my_delegate(d), my_monitor(s), my_root(t) {}
// predicate for concurrent_monitor notification
bool operator()(uintptr_t ctx) const { return (void*)ctx == (void*)&my_delegate; }
};
void task_arena_base::internal_execute(internal::delegate_base& d) const {
__TBB_ASSERT(my_arena, NULL);
generic_scheduler* s = governor::local_scheduler_weak();
__TBB_ASSERT(s, "Scheduler is not initialized");
bool same_arena = s->my_arena == my_arena;
size_t index1 = s->my_arena_index;
if (!same_arena) {
index1 = my_arena->occupy_free_slot</* as_worker*/false>(*s);
if (index1 == arena::out_of_arena) {
#if __TBB_USE_OPTIONAL_RTTI
// Workaround for the bug inside graph. If the thread can not occupy arena slot during task_arena::execute()
// and all aggregator operations depend on this task completion (all other threads are inside arena already)
// deadlock appears, because enqueued task will never enter arena.
// Workaround: check if the task came from graph via RTTI (casting to graph::spawn_functor)
// and enqueue this task with non-blocking internal_enqueue method.
// TODO: have to change behaviour later in next GOLD release (maybe to add new library entry point - try_execute)
typedef tbb::flow::interface10::graph::spawn_functor graph_funct;
internal::delegated_function< graph_funct, void >* deleg_funct =
dynamic_cast< internal::delegated_function< graph_funct, void>* >(&d);
if (deleg_funct) {
internal_enqueue(*new(task::allocate_root(__TBB_CONTEXT_ARG1(*my_context)))
internal::function_task< internal::strip< graph_funct >::type >
(internal::forward< graph_funct >(deleg_funct->my_func)), 0);
return;
} else {
#endif /* __TBB_USE_OPTIONAL_RTTI */
concurrent_monitor::thread_context waiter;
#if __TBB_TASK_GROUP_CONTEXT
task_group_context exec_context(task_group_context::isolated, my_version_and_traits & exact_exception_flag);
#if __TBB_FP_CONTEXT
exec_context.copy_fp_settings(*my_context);
#endif
#endif
auto_empty_task root(__TBB_CONTEXT_ARG(s, &exec_context));
root.prefix().ref_count = 2;
my_arena->enqueue_task(*new(task::allocate_root(__TBB_CONTEXT_ARG1(exec_context)))
delegated_task(d, my_arena->my_exit_monitors, &root),
0, s->my_random); // TODO: priority?
size_t index2 = arena::out_of_arena;
do {
my_arena->my_exit_monitors.prepare_wait(waiter, (uintptr_t)&d);
if (__TBB_load_with_acquire(root.prefix().ref_count) < 2) {
my_arena->my_exit_monitors.cancel_wait(waiter);
break;
}
index2 = my_arena->occupy_free_slot</*as_worker*/false>(*s);
if (index2 != arena::out_of_arena) {
my_arena->my_exit_monitors.cancel_wait(waiter);
nested_arena_context scope(s, my_arena, index2, scheduler_properties::master, same_arena);
s->local_wait_for_all(root, NULL);
#if TBB_USE_EXCEPTIONS
__TBB_ASSERT(!exec_context.my_exception, NULL); // exception can be thrown above, not deferred
#endif
__TBB_ASSERT(root.prefix().ref_count == 0, NULL);
break;
}
my_arena->my_exit_monitors.commit_wait(waiter);
} while (__TBB_load_with_acquire(root.prefix().ref_count) == 2);
if (index2 == arena::out_of_arena) {
// notify a waiting thread even if this thread did not enter arena,
// in case it was woken by a leaving thread but did not need to enter
my_arena->my_exit_monitors.notify_one(); // do not relax!
}
#if TBB_USE_EXCEPTIONS
// process possible exception
if (task_group_context::exception_container_type *pe = exec_context.my_exception)
TbbRethrowException(pe);
#endif
return;
#if __TBB_USE_OPTIONAL_RTTI
} // if task came from graph
#endif
} // if (index1 == arena::out_of_arena)
} // if (!same_arena)
context_guard_helper</*report_tasks=*/false> context_guard;
context_guard.set_ctx(__TBB_CONTEXT_ARG1(my_context));
#if TBB_USE_EXCEPTIONS
try {
#endif
//TODO: replace dummy tasks for workers as well to avoid using of the_dummy_context
nested_arena_context scope(s, my_arena, index1, scheduler_properties::master, same_arena);
d();
#if TBB_USE_EXCEPTIONS
}
catch (...) {
context_guard.restore_default(); // TODO: is it needed on Windows?
if (my_version_and_traits & exact_exception_flag) throw;
else {
task_group_context exception_container(task_group_context::isolated,
task_group_context::default_traits & ~task_group_context::exact_exception);
exception_container.register_pending_exception();
__TBB_ASSERT(exception_container.my_exception, NULL);
TbbRethrowException(exception_container.my_exception);
}
}
#endif
}
// this wait task is a temporary approach to wait for arena emptiness for masters without slots
// TODO: it will be rather reworked for one source of notification from is_out_of_work
class wait_task : public task {
binary_semaphore & my_signal;
task* execute() __TBB_override {
generic_scheduler* s = governor::local_scheduler_if_initialized();
__TBB_ASSERT( s, NULL );
__TBB_ASSERT( s->outermost_level(), "The enqueued task can be processed only on outermost level" );
if ( s->is_worker() ) {
__TBB_ASSERT( s->my_innermost_running_task == this, NULL );
// Mimic worker on outermost level to run remaining tasks
s->my_innermost_running_task = s->my_dummy_task;
s->local_wait_for_all( *s->my_dummy_task, NULL );
s->my_innermost_running_task = this;
} else s->my_arena->is_out_of_work(); // avoids starvation of internal_wait: issuing this task makes arena full
my_signal.V();
return NULL;
}
public:
wait_task ( binary_semaphore & sema ) : my_signal(sema) {}
};
void task_arena_base::internal_wait() const {
__TBB_ASSERT(my_arena, NULL);
generic_scheduler* s = governor::local_scheduler_weak();
__TBB_ASSERT(s, "Scheduler is not initialized");
__TBB_ASSERT(s->my_arena != my_arena || s->my_arena_index == 0, "task_arena::wait_until_empty() is not supported within a worker context" );
if( s->my_arena == my_arena ) {
//unsupported, but try do something for outermost master
__TBB_ASSERT(s->master_outermost_level(), "unsupported");
if( !s->my_arena_index )
while( my_arena->num_workers_active() )
s->wait_until_empty();
} else for(;;) {
while( my_arena->my_pool_state != arena::SNAPSHOT_EMPTY ) {
if( !__TBB_load_with_acquire(my_arena->my_slots[0].my_scheduler) // TODO TEMP: one master, make more masters
&& as_atomic(my_arena->my_slots[0].my_scheduler).compare_and_swap(s, NULL) == NULL ) {
nested_arena_context a(s, my_arena, 0, scheduler_properties::worker, false);
s->wait_until_empty();
} else {
binary_semaphore waiter; // TODO: replace by a single event notification from is_out_of_work
internal_enqueue( *new( task::allocate_root(__TBB_CONTEXT_ARG1(*my_context)) ) wait_task(waiter), 0 ); // TODO: priority?
waiter.P(); // TODO: concurrent_monitor
}
}
if( !my_arena->num_workers_active() && !my_arena->my_slots[0].my_scheduler) // no activity
break; // spin until workers active but avoid spinning in a worker
__TBB_Yield(); // wait until workers and master leave
}
}
/*static*/ int task_arena_base::internal_current_slot() {
generic_scheduler* s = governor::local_scheduler_if_initialized();
return s? int(s->my_arena_index) : -1;
}
#if __TBB_TASK_ISOLATION
class isolation_guard : tbb::internal::no_copy {
isolation_tag &guarded;
isolation_tag previous_value;
public:
isolation_guard( isolation_tag &isolation ) : guarded( isolation ), previous_value( isolation ) {}
~isolation_guard() {
guarded = previous_value;
}
};
void isolate_within_arena( delegate_base& d, intptr_t isolation ) {
// TODO: Decide what to do if the scheduler is not initialized. Is there a use case for it?
generic_scheduler* s = governor::local_scheduler_weak();
__TBB_ASSERT( s, "this_task_arena::isolate() needs an initialized scheduler" );
// Theoretically, we can keep the current isolation in the scheduler; however, it makes sense to store it in innermost
// running task because it can in principle be queried via task::self().
isolation_tag& current_isolation = s->my_innermost_running_task->prefix().isolation;
// We temporarily change the isolation tag of the currently running task. It will be restored in the destructor of the guard.
isolation_guard guard( current_isolation );
current_isolation = isolation? isolation : reinterpret_cast<isolation_tag>(&d);
d();
}
#endif /* __TBB_TASK_ISOLATION */
int task_arena_base::internal_max_concurrency(const task_arena *ta) {
arena* a = NULL;
if( ta ) // for special cases of ta->max_concurrency()
a = ta->my_arena;
else if( generic_scheduler* s = governor::local_scheduler_if_initialized() )
a = s->my_arena; // the current arena if any
if( a ) { // Get parameters from the arena
__TBB_ASSERT( !ta || ta->my_max_concurrency==1, NULL );
return a->my_num_reserved_slots + a->my_max_num_workers;
} else {
__TBB_ASSERT( !ta || ta->my_max_concurrency==automatic, NULL );
return int(governor::default_num_threads());
}
}
} // tbb::interfaceX::internal
} // tbb::interfaceX
} // tbb
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< arena_base >
tbb::interface7::internal::task_arena_base::internal_initialize
void __TBB_EXPORTED_METHOD internal_initialize()
tbb::internal::arena_slot
Definition: scheduler_common.h:377
tbb::internal::__TBB_load_with_acquire
T __TBB_load_with_acquire(const volatile T &location)
Definition: tbb_machine.h:712
tbb::interface7::internal::task_arena_base::internal_terminate
void __TBB_EXPORTED_METHOD internal_terminate()
governor.h
head
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void ITT_FORMAT p const __itt_domain __itt_id __itt_string_handle const wchar_t size_t ITT_FORMAT lu const __itt_domain __itt_id head
Definition: ittnotify_static.h:207
tbb::interface7::internal::task_arena_base::automatic
static const int automatic
Typedef for number of threads that is automatic.
Definition: task_arena.h:205
tbb::internal::concurrent_monitor::cancel_wait
void cancel_wait(thread_context &thr)
Cancel the wait. Removes the thread from the wait queue if not removed yet.
Definition: concurrent_monitor.cpp:50
tbb::internal::arena::occupy_free_slot_in_range
size_t occupy_free_slot_in_range(generic_scheduler &s, size_t lower, size_t upper)
Tries to occupy a slot in the specified range.
Definition: arena.cpp:115
tbb::internal::arena::on_thread_leaving
void on_thread_leaving()
Notification that worker or master leaves its arena.
Definition: arena.h:394
tbb::internal::create_coroutine
void create_coroutine(coroutine_type &c, size_t stack_size, void *arg)
Definition: co_context.h:154
scope
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void ITT_FORMAT p const __itt_domain __itt_id __itt_string_handle const wchar_t size_t ITT_FORMAT lu const __itt_domain __itt_id __itt_relation __itt_id ITT_FORMAT p const wchar_t int ITT_FORMAT __itt_group_mark d __itt_event ITT_FORMAT __itt_group_mark d void const wchar_t const wchar_t int ITT_FORMAT __itt_group_sync __itt_group_fsync x void const wchar_t int const wchar_t int int ITT_FORMAT __itt_group_sync __itt_group_fsync x void ITT_FORMAT __itt_group_sync __itt_group_fsync p void ITT_FORMAT __itt_group_sync __itt_group_fsync p void size_t ITT_FORMAT lu no args __itt_obj_prop_t __itt_obj_state_t ITT_FORMAT d const char ITT_FORMAT s const char ITT_FORMAT s __itt_frame ITT_FORMAT p __itt_counter ITT_FORMAT p __itt_counter unsigned long long ITT_FORMAT lu __itt_counter unsigned long long ITT_FORMAT lu __itt_counter __itt_clock_domain unsigned long long void ITT_FORMAT p const wchar_t ITT_FORMAT S __itt_mark_type const wchar_t ITT_FORMAT S __itt_mark_type const char ITT_FORMAT s __itt_mark_type ITT_FORMAT d __itt_caller ITT_FORMAT p __itt_caller ITT_FORMAT p no args const __itt_domain __itt_clock_domain unsigned long long __itt_id ITT_FORMAT lu const __itt_domain __itt_clock_domain unsigned long long __itt_id __itt_id void ITT_FORMAT p const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_clock_domain unsigned long long __itt_id __itt_string_handle __itt_scope scope
Definition: ittnotify_static.h:317
tbb::internal::arena_base::my_references
atomic< unsigned > my_references
Reference counter for the arena.
Definition: arena.h:153
tbb::interface7::internal::task_arena_base::my_context
task_group_context * my_context
default context of the arena
Definition: task_arena.h:124
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::internal::arena::out_of_arena
static const size_t out_of_arena
Definition: arena.h:382
tbb::interface7::internal::task_arena_base::my_max_concurrency
int my_max_concurrency
Concurrency level for deferred initialization.
Definition: task_arena.h:128
tbb::internal::generic_scheduler::nested_arena_exit
void nested_arena_exit()
Definition: arena.cpp:756
tbb::interface7::internal::task_arena_base::internal_max_concurrency
static int __TBB_EXPORTED_FUNC internal_max_concurrency(const task_arena *)
task_arena.h
tbb::internal::arena_base::my_aba_epoch
uintptr_t my_aba_epoch
ABA prevention marker.
Definition: arena.h:235
tbb::internal::task_stream::initialize
void initialize(unsigned n_lanes)
Definition: task_stream.h:83
tbb::internal::market::my_arenas_aba_epoch
uintptr_t my_arenas_aba_epoch
ABA prevention marker to assign to newly created arenas.
Definition: market.h:143
tbb::internal::scheduler_state
Definition: scheduler.h:77
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::atomic_update
T1 atomic_update(tbb::atomic< T1 > &dst, T2 newValue, Pred compare)
Atomically replaces value of dst with newValue if they satisfy condition of compare predicate.
Definition: tbb_misc.h:186
tbb::internal::arena_base::my_task_stream
task_stream< num_priority_levels > my_task_stream
Task pool for the tasks scheduled via task::enqueue() method.
Definition: arena.h:172
ITT_NOTIFY
#define ITT_NOTIFY(name, obj)
Definition: itt_notify.h:112
tbb::internal::market::global_market
static market & global_market(bool is_public, unsigned max_num_workers=0, size_t stack_size=0)
Factory method creating new market object.
Definition: market.cpp:96
tbb::internal::generic_scheduler::my_dummy_task
task * my_dummy_task
Fake root task created by slave threads.
Definition: scheduler.h:186
tbb::task_group_context::exception_container_type
internal::tbb_exception_ptr exception_container_type
Definition: task.h:356
tbb::interface7::internal::task_arena_base::my_arena
internal::arena * my_arena
NULL if not currently initialized.
Definition: task_arena.h:120
tbb::internal::priority_critical
static const int priority_critical
Definition: task.h:302
tbb::internal::arena::ref_external
static const unsigned ref_external
Reference increment values for externals and workers.
Definition: arena.h:327
tbb::internal::arena::num_arena_slots
static int unsigned num_arena_slots(unsigned num_slots)
Definition: arena.h:296
global_control.h
tbb::internal::arena_base::my_exit_monitors
concurrent_monitor my_exit_monitors
Waiting object for master threads that cannot join the arena.
Definition: arena.h:263
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::arena::occupy_free_slot
size_t occupy_free_slot(generic_scheduler &s)
Tries to occupy a slot in the arena. On success, returns the slot index; if no slot is available,...
Definition: arena.cpp:130
tbb::internal::governor::one_time_init
static void one_time_init()
Definition: governor.cpp:156
tbb::interface7::internal::isolate_within_arena
void __TBB_EXPORTED_FUNC isolate_within_arena(delegate_base &d, intptr_t isolation=0)
tbb::task_group_context
Used to form groups of tasks.
Definition: task.h:347
tbb::internal::generic_scheduler::local_wait_for_all
virtual void local_wait_for_all(task &parent, task *child)=0
tbb::internal::arena::is_recall_requested
bool is_recall_requested() const
Check if the recall is requested by the market.
Definition: arena.h:339
tbb::interface7::internal::task_arena_base::internal_enqueue
void __TBB_EXPORTED_METHOD internal_enqueue(task &, intptr_t) const
tbb::internal::arena_base::my_max_num_workers
unsigned my_max_num_workers
The number of workers requested by the master thread owning the arena.
Definition: arena.h:185
tbb::task_group_context::isolated
Definition: task.h:367
tbb::internal::scheduler_state::my_arena
arena * my_arena
The arena that I own (if master) or am servicing at the moment (if worker)
Definition: scheduler.h:85
tbb::internal::mail_inbox::is_idle_state
bool is_idle_state(bool value) const
Indicate whether thread that reads this mailbox is idle.
Definition: mailbox.h:218
tbb::task_group_context::default_traits
Definition: task.h:380
__TBB_CONTEXT_ARG1
#define __TBB_CONTEXT_ARG1(context)
Definition: scheduler_common.h:59
tbb::internal::spin_wait_while_eq
void spin_wait_while_eq(const volatile T &location, U value)
Spin WHILE the value of the variable is equal to a given value.
Definition: tbb_machine.h:394
tbb::internal::task_stream::empty
bool empty(int level)
Checks existence of a task.
Definition: task_stream.h:138
d
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
Definition: ittnotify_static.h:109
tbb::interface7::internal::task_arena_base::internal_attach
void __TBB_EXPORTED_METHOD internal_attach()
tbb::internal::arena_slot::free_task_pool
void free_task_pool()
Deallocate task pool that was allocated by means of allocate_task_pool.
Definition: scheduler_common.h:397
tbb::internal::arena::has_enqueued_tasks
bool has_enqueued_tasks()
Check for the presence of enqueued tasks at all priority levels.
Definition: arena.cpp:426
tbb::internal::arena::is_out_of_work
bool is_out_of_work()
Check if there is job anywhere in arena.
Definition: arena.cpp:454
tbb::internal::governor::local_scheduler_if_initialized
static generic_scheduler * local_scheduler_if_initialized()
Definition: governor.h:139
semaphore.h
itt_notify.h
tbb::internal::arena_base::my_cpu_ctl_env
cpu_ctl_env my_cpu_ctl_env
FPU control settings of arena's master thread captured at the moment of arena instantiation.
Definition: arena.h:239
_flow_graph_impl.h
scheduler.h
tbb::internal::arena::pool_state_t
uintptr_t pool_state_t
Definition: arena.h:315
tbb::task::executing
task is running, and will be destroyed after method execute() completes.
Definition: task.h:626
tbb::internal::concurrent_monitor::prepare_wait
void prepare_wait(thread_context &thr, uintptr_t ctx=0)
prepare wait by inserting 'thr' into the wait queue
Definition: concurrent_monitor.cpp:32
tbb::internal::scheduler_state::my_arena_index
size_t my_arena_index
Index of the arena slot the scheduler occupies now, or occupied last time.
Definition: scheduler.h:79
tbb::internal::arena_slot_line1::my_scheduler
generic_scheduler * my_scheduler
Scheduler of the thread attached to the slot.
Definition: scheduler_common.h:333
tbb::interface6::task_scheduler_observer
Definition: task_scheduler_observer.h:92
tbb::internal::governor::default_num_threads
static unsigned default_num_threads()
Definition: governor.h:84
tbb::internal::mail_outbox::drain
intptr_t drain()
Drain the mailbox.
Definition: mailbox.h:168
tbb::priority_t
priority_t
Definition: task.h:306
tbb::interface7::internal::task_arena_base::internal_current_slot
static int __TBB_EXPORTED_FUNC internal_current_slot()
EmptyTaskPool
#define EmptyTaskPool
Definition: scheduler.h:46
tbb::internal::task_prefix::ref_count
__TBB_atomic reference_count ref_count
Reference count used for synchronization.
Definition: task.h:263
tbb::internal::is_critical
bool is_critical(task &t)
Definition: task.h:1003
tbb::internal::arena::SNAPSHOT_EMPTY
static const pool_state_t SNAPSHOT_EMPTY
No tasks to steal since last snapshot was taken.
Definition: arena.h:318
tbb::internal::es_ref_count_active
Set if ref_count might be changed by another thread. Used for debugging.
Definition: scheduler_common.h:176
tbb::internal::market::create_arena
static arena * create_arena(int num_slots, int num_reserved_slots, size_t stack_size)
Creates an arena object.
Definition: market.cpp:308
tbb::interface7::internal::task_arena_base::internal_wait
void __TBB_EXPORTED_METHOD internal_wait() const
tbb::internal::arena::enqueue_task
void enqueue_task(task &, intptr_t, FastRandom &)
enqueue a task into starvation-resistance queue
Definition: arena.cpp:597
GATHER_STATISTIC
#define GATHER_STATISTIC(x)
Definition: tbb_statistics.h:232
ITT_SYNC_CREATE
#define ITT_SYNC_CREATE(obj, type, name)
Definition: itt_notify.h:115
tbb::internal::affinity_id
unsigned short affinity_id
An id as used for specifying affinity.
Definition: task.h:128
scheduler_utility.h
tbb::interface7::internal::task_arena_base::exact_exception_flag
Definition: task_arena.h:156
int
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void ITT_FORMAT p const __itt_domain __itt_id __itt_string_handle const wchar_t size_t ITT_FORMAT lu const __itt_domain __itt_id __itt_relation __itt_id ITT_FORMAT p const wchar_t int ITT_FORMAT __itt_group_mark d int
Definition: ittnotify_static.h:217
tbb::internal::arena::mailbox
mail_outbox & mailbox(affinity_id id)
Get reference to mailbox corresponding to given affinity_id.
Definition: arena.h:305
tbb::internal::__TBB_store_with_release
void __TBB_store_with_release(volatile T &location, V value)
Definition: tbb_machine.h:716
tbb::internal::market::release
bool release(bool is_public, bool blocking_terminate)
Decrements market's refcount and destroys it in the end.
Definition: market.cpp:175
tbb::internal::arena::my_slots
arena_slot my_slots[1]
Definition: arena.h:390
tbb::internal::strip::type
T type
Definition: _template_helpers.h:38
tbb::internal::generic_scheduler::attach_arena
void attach_arena(arena *, size_t index, bool is_master)
Definition: arena.cpp:80
tbb::internal::generic_scheduler::nested_arena_entry
void nested_arena_entry(arena *, size_t)
Definition: arena.cpp:729
tbb::interface7::internal::task_arena_base::internal_execute
void __TBB_EXPORTED_METHOD internal_execute(delegate_base &) const
tbb::internal::generic_scheduler::attach_mailbox
void attach_mailbox(affinity_id id)
Definition: scheduler.h:667
tbb::internal::cpu_ctl_env::set_env
void set_env() const
Definition: gcc_ia32_common.h:95
tbb::internal::no_isolation
const isolation_tag no_isolation
Definition: task.h:133
tbb::internal::occupy_slot
static bool occupy_slot(generic_scheduler *&slot, generic_scheduler &s)
Definition: arena.cpp:111
__TBB_ISOLATION_ARG
#define __TBB_ISOLATION_ARG(arg1, isolation)
Definition: scheduler_common.h:68
tbb::internal::task_prefix::affinity
affinity_id affinity
Definition: task.h:283
tbb::this_task_arena::current_thread_index
int current_thread_index()
Returns the index, aka slot number, of the calling thread in its current arena.
Definition: task_arena.h:484
tbb::interface7::internal::task_arena_base::my_master_slots
unsigned my_master_slots
Reserved master slots.
Definition: task_arena.h:131
__TBB_Yield
#define __TBB_Yield()
Definition: ibm_aix51.h:44
tbb::internal::generic_scheduler::is_worker
bool is_worker() const
True if running on a worker thread, false otherwise.
Definition: scheduler.h:673
tbb::internal::arena_base::my_market
market * my_market
The market that owns this arena.
Definition: arena.h:232
tbb::internal::arena::allocate_arena
static arena & allocate_arena(market &, unsigned num_slots, unsigned num_reserved_slots)
Allocate an instance of arena.
Definition: arena.cpp:285
tbb::internal::arena_base::my_num_reserved_slots
unsigned my_num_reserved_slots
The number of reserved slots (can be occupied only by masters).
Definition: arena.h:253
tbb::internal::nested_arena_context::~nested_arena_context
~nested_arena_context()
Definition: arena.cpp:691
arena.h
tbb::internal::mail_inbox::set_is_idle
void set_is_idle(bool value)
Indicate whether thread that reads this mailbox is idle.
Definition: mailbox.h:211
tbb::task::state
state_type state() const
Current execution state.
Definition: task.h:901
tbb::internal::cpu_ctl_env::get_env
void get_env()
Definition: gcc_ia32_common.h:77
tbb::internal::NFS_MaxLineSize
const size_t NFS_MaxLineSize
Compile-time constant that is upper bound on cache line/sector size.
Definition: tbb_stddef.h:216
tbb::internal::arena::SNAPSHOT_FULL
static const pool_state_t SNAPSHOT_FULL
At least one task has been offered for stealing since the last snapshot started.
Definition: arena.h:321
tbb::internal::scheduler_properties::worker
static const bool worker
Definition: scheduler.h:51
tbb::internal::arena::restore_priority_if_need
void restore_priority_if_need()
If enqueued tasks found, restore arena priority and task presence status.
Definition: arena.cpp:434
tbb::internal::arena_base::my_num_workers_allotted
unsigned my_num_workers_allotted
The number of workers that have been marked out by the resource manager to service the arena.
Definition: arena.h:147
parent
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id parent
Definition: ittnotify_static.h:176
tbb::task_group_context::my_version_and_traits
uintptr_t my_version_and_traits
Version for run-time checks and behavioral traits of the context.
Definition: task.h:435
tbb::internal::nested_arena_context::my_orig_state
scheduler_state my_orig_state
Definition: arena.cpp:710
tbb::internal::isolation_tag
intptr_t isolation_tag
A tag for task isolation.
Definition: task.h:132
tbb::internal::scheduler_state::my_innermost_running_task
task * my_innermost_running_task
Innermost task whose task::execute() is running. A dummy task on the outermost level.
Definition: scheduler.h:88
tbb::internal::scheduler_state::my_arena_slot
arena_slot * my_arena_slot
Pointer to the slot in the arena we own at the moment.
Definition: scheduler.h:82
tbb::task_group_context::exact_exception
Definition: task.h:372
__TBB_override
#define __TBB_override
Definition: tbb_stddef.h:240
tbb::internal::arena_base::my_limit
atomic< unsigned > my_limit
The maximal number of currently busy slots.
Definition: arena.h:161
tbb::task::parent
task * parent() const
task on whose behalf this task is working, or NULL if this is a root.
Definition: task.h:854
tbb::internal::nested_arena_context::mimic_outermost_level
void mimic_outermost_level(arena *a, bool type)
Definition: arena.cpp:714
tbb::internal::market
Definition: market.h:45
s
void const char const char int ITT_FORMAT __itt_group_sync s
Definition: ittnotify_static.h:91
tbb::internal::subsequent_lane_selector
Definition: task_stream_extended.h:110
tbb::internal::scheduler_properties::outermost
bool outermost
Indicates that a scheduler is on outermost level.
Definition: scheduler.h:57
tbb::task::allocated
task object is freshly allocated or recycled.
Definition: task.h:632
tbb::internal::governor::assume_scheduler
static void assume_scheduler(generic_scheduler *s)
Temporarily set TLS slot to the given scheduler.
Definition: governor.cpp:116
tbb::internal::governor::local_scheduler
static generic_scheduler * local_scheduler()
Obtain the thread-local instance of the TBB scheduler.
Definition: governor.h:129
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::governor::is_set
static bool is_set(generic_scheduler *s)
Used to check validity of the local scheduler TLS contents.
Definition: governor.cpp:120
tbb::internal::task_prefix::extra_state
unsigned char extra_state
Miscellaneous state that is not directly visible to users, stored as a byte for compactness.
Definition: task.h:281
tbb::internal::arena::process
void process(generic_scheduler &)
Registers the worker with the arena and enters TBB scheduler dispatch loop.
Definition: arena.cpp:146
tbb::internal::arena
Definition: arena.h:276
tbb::internal::mail_outbox::construct
void construct()
Construct *this as a mailbox from zeroed memory.
Definition: mailbox.h:158
tbb::internal::scheduler_state::my_inbox
mail_inbox my_inbox
Definition: scheduler.h:90
poison_value
#define poison_value(g)
Definition: scheduler_common.h:235
tbb::internal::scheduler_state::my_properties
scheduler_properties my_properties
Definition: scheduler.h:101
tbb::task::ready
task is in ready pool, or is going to be put there, or was just taken off.
Definition: task.h:630
tbb::internal::mail_outbox
Class representing where mail is put.
Definition: mailbox.h:96
tbb::internal::nested_arena_context::my_orig_ctx
task_group_context * my_orig_ctx
Definition: arena.cpp:711
tbb::internal::num_priority_levels
static const intptr_t num_priority_levels
Definition: scheduler_common.h:129
tbb::internal::NFS_GetLineSize
size_t __TBB_EXPORTED_FUNC NFS_GetLineSize()
Cache/sector line size.
Definition: cache_aligned_allocator.cpp:167
tbb::internal::task_prefix::state
unsigned char state
A task::state_type, stored as a byte for compactness.
Definition: task.h:272
tbb::task::allocate_root
static internal::allocate_root_proxy allocate_root()
Returns proxy for overloaded new that allocates a root task.
Definition: task.h:652
tbb::internal::random_lane_selector
Definition: task_stream_extended.h:83
tbb::interface7::internal::task_arena_base::my_version_and_traits
intptr_t my_version_and_traits
Special settings.
Definition: task_arena.h:134
tbb::internal::as_atomic
atomic< T > & as_atomic(T &t)
Definition: atomic.h:572
tbb::internal::task_prefix::isolation
isolation_tag isolation
The tag used for task isolation.
Definition: task.h:209
tbb::internal::reference_count
intptr_t reference_count
A reference count.
Definition: task.h:120
tbb::internal::nested_arena_context
Definition: arena.cpp:673
tbb::internal::nested_arena_context::same_arena
const bool same_arena
Definition: arena.cpp:712
tbb::task
Base class for user-defined tasks.
Definition: task.h:604
tbb::internal::arena_base::my_num_workers_requested
int my_num_workers_requested
The number of workers that are currently requested from the resource manager.
Definition: arena.h:188
tbb::internal::nested_arena_context::my_scheduler
generic_scheduler & my_scheduler
Definition: arena.cpp:709
tbb::internal::generic_scheduler
Work stealing task scheduler.
Definition: scheduler.h:137
tbb::internal::governor::local_scheduler_weak
static generic_scheduler * local_scheduler_weak()
Definition: governor.h:134
sync_releasing
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 sync_releasing
Definition: ittnotify_static.h:104
tbb::internal::arena::allocation_size
static int allocation_size(unsigned num_slots)
Definition: arena.h:300
tbb::internal::FastRandom
A fast random number generator.
Definition: tbb_misc.h:135
tbb::internal::concurrent_monitor::notify_one
void notify_one()
Notify one thread about the event.
Definition: concurrent_monitor.h:157
tbb::internal::generic_scheduler::my_market
market * my_market
The market I am in.
Definition: scheduler.h:172
tbb::internal::arena_base::my_pool_state
tbb::atomic< uintptr_t > my_pool_state
Current task pool state and estimate of available tasks amount.
Definition: arena.h:195
p
void const char const char int ITT_FORMAT __itt_group_sync p
Definition: ittnotify_static.h:91
tbb::internal::NFS_Free
void __TBB_EXPORTED_FUNC NFS_Free(void *)
Free memory allocated by NFS_Allocate.
Definition: cache_aligned_allocator.cpp:198
tail
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void ITT_FORMAT p const __itt_domain __itt_id __itt_string_handle const wchar_t size_t ITT_FORMAT lu const __itt_domain __itt_id __itt_relation __itt_id tail
Definition: ittnotify_static.h:207
tbb::internal::task_prefix::context
task_group_context * context
Shared context that is used to communicate asynchronous state changes.
Definition: task.h:219
tbb::internal::arena::arena
arena(market &, unsigned max_num_workers, unsigned num_reserved_slots)
Constructor.
Definition: arena.cpp:226
tbb::internal::market::adjust_demand
void adjust_demand(arena &, int delta)
Request that arena's need in workers should be adjusted.
Definition: market.cpp:557
tbb::internal::concurrent_monitor::commit_wait
bool commit_wait(thread_context &thr)
Commit wait if event count has not changed; otherwise, cancel wait.
Definition: concurrent_monitor.h:135
tbb::internal::arena::free_arena
void free_arena()
Completes arena shutdown, destructs and deallocates it.
Definition: arena.cpp:296
tbb::internal::make_critical
void make_critical(task &t)
Definition: task.h:1002
tbb::internal::scheduler_properties::master
static const bool master
Definition: scheduler.h:52
tbb::internal::nested_arena_context::nested_arena_context
nested_arena_context(generic_scheduler *s, arena *a, size_t slot_index, bool type, bool same)
Definition: arena.cpp:675
tbb::interface7::task_arena
Definition: task_arena.h:228
tbb::internal::es_task_enqueued
Tag for enqueued tasks.
Definition: scheduler_common.h:172
tbb::internal::arena_base::my_num_slots
unsigned my_num_slots
The number of slots in the arena.
Definition: arena.h:250
tbb::internal::generic_scheduler::wait_until_empty
void wait_until_empty()
Definition: arena.cpp:772
tbb::internal::arena::work_spawned
Definition: arena.h:285
tbb::internal::NFS_Allocate
void *__TBB_EXPORTED_FUNC NFS_Allocate(size_t n_element, size_t element_size, void *hint)
Allocate memory on cache/sector line boundary.
Definition: cache_aligned_allocator.cpp:176
type
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type type
Definition: ittnotify_static.h:198
tbb::internal::arena::num_workers_active
unsigned num_workers_active() const
The number of workers active in the arena.
Definition: arena.h:334
tbb::internal::generic_scheduler::create_worker
static generic_scheduler * create_worker(market &m, size_t index, bool geniune)
Initialize a scheduler for a worker thread.
Definition: scheduler.cpp:1271
__TBB_ASSERT_EX
#define __TBB_ASSERT_EX(predicate, comment)
"Extended" version is useful to suppress warnings if a variable is only used with an assert
Definition: tbb_stddef.h:167
tbb::internal::__TBB_load_relaxed
T __TBB_load_relaxed(const volatile T &location)
Definition: tbb_machine.h:738
tbb::internal::arena_slot_line2::hint_for_pop
unsigned hint_for_pop
Hint provided for operations with the container of starvation-resistant tasks.
Definition: scheduler_common.h:354
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::scheduler_properties::type
bool type
Indicates that a scheduler acts as a master or a worker.
Definition: scheduler.h:54
tbb::flow::interface10::graph::spawn_functor
A functor that spawns a task.
Definition: _flow_graph_impl.h:256
__TBB_CONTEXT_ARG
#define __TBB_CONTEXT_ARG(arg1, context)
Definition: scheduler_common.h:60

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.