| /* |
| Copyright 2005-2010 Intel Corporation. All Rights Reserved. |
| |
| This file is part of Threading Building Blocks. |
| |
| Threading Building Blocks is free software; you can redistribute it |
| and/or modify it under the terms of the GNU General Public License |
| version 2 as published by the Free Software Foundation. |
| |
| Threading Building Blocks is distributed in the hope that it will be |
| useful, but WITHOUT ANY WARRANTY; without even the implied warranty |
| of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| GNU General Public License for more details. |
| |
| You should have received a copy of the GNU General Public License |
| along with Threading Building Blocks; if not, write to the Free Software |
| Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA |
| |
| As a special exception, you may use this file as part of a free software |
| library without restriction. Specifically, if other files instantiate |
| templates or use macros or inline functions from this file, or you compile |
| this file and link it with other files to produce an executable, this |
| file does not by itself cause the resulting executable to be covered by |
| the GNU General Public License. This exception does not however |
| invalidate any other reasons why the executable file might be covered by |
| the GNU General Public License. |
| */ |
| |
| #include "tbb/tbb_machine.h" |
| |
| #include "custom_scheduler.h" |
| #include "scheduler_utility.h" |
| #include "governor.h" |
| #include "market.h" |
| #include "arena.h" |
| #include "mailbox.h" |
| #include "observer_proxy.h" |
| #include "itt_notify.h" |
| |
| namespace tbb { |
| namespace internal { |
| |
| /** Defined in tbb_main.cpp **/ |
| extern generic_scheduler* (*AllocateSchedulerPtr)( arena*, size_t index ); |
| |
| inline generic_scheduler* allocate_scheduler ( arena* a, size_t index ) { |
| return AllocateSchedulerPtr(a, index); |
| } |
| |
| #if __TBB_TASK_GROUP_CONTEXT |
| #if !__TBB_ARENA_PER_MASTER |
| //! Head of the list of master thread schedulers. |
| static scheduler_list_node_t the_scheduler_list_head; |
| |
| //! Mutex protecting access to the list of schedulers. |
| static mutex the_scheduler_list_mutex; |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| |
| //! Counter that is incremented whenever new cancellation signal is sent to a task group. |
| /** Together with generic_scheduler::local_cancel_count forms cross-thread signaling |
| mechanism that allows to avoid locking at the hot path of normal execution flow. |
| |
| When a descendant task group context is being registered or unregistered, |
| the global and local counters are compared. If they differ, it means that |
| a cancellation signal is being propagated, and registration/deregistration |
| routines take slower branch that may block (at most one thread of the pool |
| can be blocked at any moment). Otherwise the control path is lock-free and fast. **/ |
| uintptr_t global_cancel_count = 0; |
| |
| //! Context to be associated with dummy tasks of worker threads schedulers. |
| /** It is never used for its direct purpose, and is introduced solely for the sake |
| of avoiding one extra conditional branch in the end of wait_for_all method. **/ |
| static task_group_context dummy_context(task_group_context::isolated); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| |
| #if STATISTICS |
| static statistics the_statistics; |
| #endif /* STATISTICS */ |
| |
| void Scheduler_OneTimeInitialization ( bool itt_present ) { |
| AllocateSchedulerPtr = itt_present ? &custom_scheduler<DefaultSchedulerTraits>::allocate_scheduler : |
| &custom_scheduler<IntelSchedulerTraits>::allocate_scheduler; |
| #if __TBB_TASK_GROUP_CONTEXT && !__TBB_ARENA_PER_MASTER |
| ITT_SYNC_CREATE(&the_scheduler_list_mutex, SyncType_GlobalLock, SyncObj_SchedulersList); |
| the_scheduler_list_head.my_next = &the_scheduler_list_head; |
| the_scheduler_list_head.my_prev = &the_scheduler_list_head; |
| #endif /* __TBB_TASK_GROUP_CONTEXT && !__TBB_ARENA_PER_MASTER */ |
| } |
| |
| //------------------------------------------------------------------------ |
| // scheduler interface |
| //------------------------------------------------------------------------ |
| |
| // A pure virtual destructor should still have a body |
| // so the one for tbb::internal::scheduler::~scheduler() is provided here |
| scheduler::~scheduler( ) {} |
| |
| //------------------------------------------------------------------------ |
| // generic_scheduler |
| //------------------------------------------------------------------------ |
| |
| #if _MSC_VER && !defined(__INTEL_COMPILER) |
| // Suppress overzealous compiler warning about using 'this' in base initializer list. |
| #pragma warning(push) |
| #pragma warning(disable:4355) |
| #endif |
| |
| generic_scheduler::generic_scheduler( arena* a, size_t index ) : |
| my_stealing_threshold(0), |
| arena_index(index), |
| task_pool_size(0), |
| my_arena_slot(&dummy_slot), |
| #if __TBB_ARENA_PER_MASTER |
| my_market(NULL), |
| #endif /* __TBB_ARENA_PER_MASTER */ |
| my_arena(a), |
| random( unsigned(this-(generic_scheduler*)NULL) ), |
| free_list(NULL), |
| innermost_running_task(NULL), |
| dummy_task(NULL), |
| ref_count(1), |
| my_affinity_id(0), |
| is_registered(false), |
| is_auto_initialized(false), |
| #if __TBB_SCHEDULER_OBSERVER |
| local_last_observer_proxy(NULL), |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| #if __TBB_COUNT_TASK_NODES |
| task_node_count(0), |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| #if STATISTICS |
| current_active(0), |
| current_length(0), |
| current_big_malloc(0), |
| execute_count(0), |
| steal_count(0), |
| mail_received_count(0), |
| proxy_execute_count(0), |
| proxy_steal_count(0), |
| proxy_bypass_count(0), |
| #endif /* STATISTICS */ |
| small_task_count(1), // Extra 1 is a guard reference |
| return_list(NULL), |
| #if __TBB_TASK_GROUP_CONTEXT |
| local_ctx_list_update(0), |
| nonlocal_ctx_list_update(0) |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| { |
| dummy_slot.task_pool = allocate_task_pool( min_task_pool_size ); |
| dummy_slot.head = dummy_slot.tail = 0; |
| dummy_task = &allocate_task( sizeof(task), __TBB_CONTEXT_ARG(NULL, NULL) ); |
| #if __TBB_TASK_GROUP_CONTEXT |
| context_list_head.my_prev = &context_list_head; |
| context_list_head.my_next = &context_list_head; |
| ITT_SYNC_CREATE(&context_list_mutex, SyncType_Scheduler, SyncObj_ContextsList); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| dummy_task->prefix().ref_count = 2; |
| ITT_SYNC_CREATE(&dummy_task->prefix().ref_count, SyncType_Scheduler, SyncObj_WorkerLifeCycleMgmt); |
| ITT_SYNC_CREATE(&return_list, SyncType_Scheduler, SyncObj_TaskReturnList); |
| __TBB_ASSERT( assert_okay(), "constructor error" ); |
| } |
| |
| #if _MSC_VER && !defined(__INTEL_COMPILER) |
| #pragma warning(pop) |
| #endif // warning 4355 is back |
| |
| #if TBB_USE_ASSERT |
| bool generic_scheduler::assert_okay() const { |
| #if TBB_USE_ASSERT >= 2 |
| acquire_task_pool(); |
| task** tp = dummy_slot.task_pool; |
| __TBB_ASSERT( task_pool_size >= min_task_pool_size, NULL ); |
| __TBB_ASSERT( my_arena_slot->head <= my_arena_slot->tail, NULL ); |
| for ( size_t i = my_arena_slot->head; i < my_arena_slot->tail; ++i ) { |
| __TBB_ASSERT( (uintptr_t)tp[i] + 1 > 1u, "nil or invalid task pointer in the deque" ); |
| __TBB_ASSERT( tp[i]->prefix().state == task::ready || |
| tp[i]->prefix().extra_state == es_task_proxy, "task in the deque has invalid state" ); |
| } |
| release_task_pool(); |
| #endif /* TBB_USE_ASSERT >=2 */ |
| return true; |
| } |
| #endif /* TBB_USE_ASSERT */ |
| |
| #if __TBB_TASK_GROUP_CONTEXT |
| void generic_scheduler::propagate_cancellation () { |
| spin_mutex::scoped_lock lock(context_list_mutex); |
| // Acquire fence is necessary to ensure that the subsequent node->my_next load |
| // returned the correct value in case it was just inserted in another thread. |
| // The fence also ensures visibility of the correct my_parent value. |
| context_list_node_t *node = __TBB_load_with_acquire(context_list_head.my_next); |
| while ( node != &context_list_head ) { |
| task_group_context &ctx = __TBB_get_object_ref(task_group_context, my_node, node); |
| // The absence of acquire fence while reading my_cancellation_requested may result |
| // in repeated traversals of the same parents chain if another group (precedent or |
| // descendant) belonging to the tree being canceled sends cancellation request of |
| // its own around the same time. |
| if ( !ctx.my_cancellation_requested ) |
| ctx.propagate_cancellation_from_ancestors(); |
| node = node->my_next; |
| __TBB_ASSERT( ctx.is_alive(), "Walked into a destroyed context while propagating cancellation" ); |
| } |
| } |
| |
| #if !__TBB_ARENA_PER_MASTER |
| /** Propagates cancellation down the tree of dependent contexts by walking each |
| thread's local list of contexts **/ |
| void generic_scheduler::propagate_cancellation ( task_group_context& ctx ) { |
| __TBB_ASSERT ( ctx.my_cancellation_requested, "No cancellation request in the context" ); |
| // The whole propagation algorithm is under the lock in order to ensure correctness |
| // in case of parallel cancellations at the different levels of the context tree. |
| // See the note 2 at the bottom of the file. |
| mutex::scoped_lock lock(the_scheduler_list_mutex); |
| // Advance global cancellation state |
| __TBB_FetchAndAddWrelease(&global_cancel_count, 1); |
| // First propagate to workers using arena to access their context lists |
| size_t num_workers = my_arena->prefix().number_of_workers; |
| for ( size_t i = 0; i < num_workers; ++i ) { |
| // No fence is necessary here since the context list of worker's scheduler |
| // can contain anything of interest only after the first stealing was done |
| // by that worker. And doing it applies the necessary fence |
| generic_scheduler *s = my_arena->prefix().worker_list[i].scheduler; |
| // If the worker is in the middle of its startup sequence, skip it. |
| if ( s ) |
| s->propagate_cancellation(); |
| } |
| // Then propagate to masters using the global list of master's schedulers |
| scheduler_list_node_t *node = the_scheduler_list_head.my_next; |
| while ( node != &the_scheduler_list_head ) { |
| __TBB_get_object_ref(generic_scheduler, my_node, node).propagate_cancellation(); |
| node = node->my_next; |
| } |
| // Now sync up the local counters |
| for ( size_t i = 0; i < num_workers; ++i ) { |
| generic_scheduler *s = my_arena->prefix().worker_list[i].scheduler; |
| // If the worker is in the middle of its startup sequence, skip it. |
| if ( s ) |
| s->local_cancel_count = global_cancel_count; |
| } |
| node = the_scheduler_list_head.my_next; |
| while ( node != &the_scheduler_list_head ) { |
| __TBB_get_object_ref(generic_scheduler, my_node, node).local_cancel_count = global_cancel_count; |
| node = node->my_next; |
| } |
| } |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| |
| |
| void generic_scheduler::init_stack_info () { |
| // Stacks are growing top-down. Highest address is called "stack base", |
| // and the lowest is "stack limit". |
| #if __TBB_ARENA_PER_MASTER |
| __TBB_ASSERT( !my_stealing_threshold, "Stealing threshold has already been calculated" ); |
| size_t stack_size = my_market->worker_stack_size(); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| size_t stack_size = my_arena->prefix().stack_size; |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| #if USE_WINTHREAD |
| #if defined(_MSC_VER)&&_MSC_VER<1400 && !_WIN64 |
| NT_TIB *pteb = (NT_TIB*)__TBB_machine_get_current_teb(); |
| #else |
| NT_TIB *pteb = (NT_TIB*)NtCurrentTeb(); |
| #endif |
| __TBB_ASSERT( &pteb < pteb->StackBase && &pteb > pteb->StackLimit, "invalid stack info in TEB" ); |
| __TBB_ASSERT( stack_size >0, "stack_size not initialized?" ); |
| // When a thread is created with the attribute STACK_SIZE_PARAM_IS_A_RESERVATION, stack limit |
| // in the TIB points to the committed part of the stack only. This renders the expression |
| // "(uintptr_t)pteb->StackBase / 2 + (uintptr_t)pteb->StackLimit / 2" virtually useless. |
| // Thus for worker threads we use the explicit stack size we used while creating them. |
| // And for master threads we rely on the following fact and assumption: |
| // - the default stack size of a master thread on Windows is 1M; |
| // - if it was explicitly set by the application it is at least as large as the size of a worker stack. |
| if ( is_worker() || stack_size < MByte ) |
| my_stealing_threshold = (uintptr_t)pteb->StackBase - stack_size / 2; |
| else |
| my_stealing_threshold = (uintptr_t)pteb->StackBase - MByte / 2; |
| #else /* USE_PTHREAD */ |
| // There is no portable way to get stack base address in Posix, so we use |
| // non-portable method (on all modern Linux) or the simplified approach |
| // based on the common sense assumptions. The most important assumption |
| // is that the main thread's stack size is not less than that of other threads. |
| void *stack_base = &stack_size; |
| #if __TBB_ipf |
| void *rsb_base = __TBB_get_bsp(); |
| #endif |
| #if __linux__ |
| size_t np_stack_size = 0; |
| void *stack_limit = NULL; |
| pthread_attr_t attr_stack, np_attr_stack; |
| if( 0 == pthread_getattr_np(pthread_self(), &np_attr_stack) ) { |
| if ( 0 == pthread_attr_getstack(&np_attr_stack, &stack_limit, &np_stack_size) ) { |
| if ( 0 == pthread_attr_init(&attr_stack) ) { |
| if ( 0 == pthread_attr_getstacksize(&attr_stack, &stack_size) ) |
| { |
| stack_base = (char*)stack_limit + np_stack_size; |
| if ( np_stack_size < stack_size ) { |
| // We are in a secondary thread. Use reliable data. |
| #if __TBB_ipf |
| // IA64 stack is split into RSE backup and memory parts |
| rsb_base = stack_limit; |
| stack_size = np_stack_size/2; |
| #else |
| stack_size = np_stack_size; |
| #endif /* !__TBB_ipf */ |
| } |
| // We are either in the main thread or this thread stack |
| // is bigger that that of the main one. As we cannot discern |
| // these cases we fall back to the default (heuristic) values. |
| } |
| pthread_attr_destroy(&attr_stack); |
| } |
| } |
| pthread_attr_destroy(&np_attr_stack); |
| } |
| #endif /* __linux__ */ |
| __TBB_ASSERT( stack_size>0, "stack size must be positive" ); |
| my_stealing_threshold = (uintptr_t)((char*)stack_base - stack_size/2); |
| #if __TBB_ipf |
| my_rsb_stealing_threshold = (uintptr_t)((char*)rsb_base + stack_size/2); |
| #endif |
| #endif /* USE_PTHREAD */ |
| } |
| |
| /** The function uses synchronization scheme similar to the one in the destructor |
| of task_group_context augmented with interlocked state change of each context |
| object. The purpose of this algo is to prevent threads doing nonlocal context |
| destruction from accessing destroyed owner-scheduler instance still pointed to |
| by the context object. **/ |
| void generic_scheduler::cleanup_local_context_list () { |
| // Detach contexts remaining in the local list |
| bool wait_for_concurrent_destroyers_to_leave = false; |
| uintptr_t local_count_snapshot = local_cancel_count; |
| local_ctx_list_update = 1; |
| { |
| spin_mutex::scoped_lock lock; // The lock is necessary in acse of a conflict only |
| __TBB_rel_acq_fence(); |
| // Check for the conflict with concurrent destroyer or cancelation propagator |
| if ( nonlocal_ctx_list_update || local_count_snapshot != global_cancel_count ) |
| lock.acquire(context_list_mutex); |
| context_list_node_t *node = context_list_head.my_next; |
| while ( node != &context_list_head ) { |
| task_group_context &ctx = __TBB_get_object_ref(task_group_context, my_node, node); |
| __TBB_ASSERT( ctx.my_kind != task_group_context::binding_required, "Only a context bound to a root task can be detached" ); |
| node = node->my_next; |
| __TBB_ASSERT( ctx.is_alive(), "Walked into a destroyed context while detaching contexts from the local list" ); |
| // On 64-bit systems my_kind can be a 32-bit value padded with 32 uninitialized bits. |
| // So the cast below is necessary to throw off the higher bytes containing garbage |
| if ( (task_group_context::kind_type)(uintptr_t)__TBB_FetchAndStoreW(&ctx.my_kind, task_group_context::detached) == task_group_context::dying ) |
| wait_for_concurrent_destroyers_to_leave = true; |
| } |
| } |
| __TBB_store_with_release( local_ctx_list_update, 0 ); |
| // Wait until other threads referencing this scheduler object finish with it |
| if ( wait_for_concurrent_destroyers_to_leave ) |
| spin_wait_until_eq( nonlocal_ctx_list_update, 0u ); |
| } |
| |
| void generic_scheduler::free_scheduler() { |
| if( in_arena() ) { |
| acquire_task_pool(); |
| leave_arena(); |
| } |
| #if __TBB_TASK_GROUP_CONTEXT |
| cleanup_local_context_list(); |
| #if !__TBB_ARENA_PER_MASTER |
| task_group_context* default_context = dummy_task->prefix().context; |
| if ( default_context != &dummy_context) { |
| // Only master thread's dummy task has a dynamically allocated context |
| default_context->task_group_context::~task_group_context(); |
| NFS_Free(default_context); |
| { |
| mutex::scoped_lock lock(the_scheduler_list_mutex); |
| my_node.my_next->my_prev = my_node.my_prev; |
| my_node.my_prev->my_next = my_node.my_next; |
| } |
| } |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| free_task<small_local_task>( *dummy_task ); |
| |
| // k accounts for a guard reference and each task that we deallocate. |
| intptr_t k = 1; |
| for(;;) { |
| while( task* t = free_list ) { |
| free_list = t->prefix().next; |
| deallocate_task(*t); |
| ++k; |
| } |
| if( return_list==plugged_return_list() ) |
| break; |
| free_list = (task*)__TBB_FetchAndStoreW( &return_list, (intptr_t)plugged_return_list() ); |
| } |
| #if __TBB_COUNT_TASK_NODES |
| #if __TBB_ARENA_PER_MASTER |
| my_market->update_task_node_count( task_node_count ); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| my_arena->prefix().task_node_count += task_node_count; |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| #if STATISTICS |
| the_statistics.record( execute_count, steal_count, mail_received_count, |
| proxy_execute_count, proxy_steal_count, proxy_bypass_count ); |
| #endif /* STATISTICS */ |
| free_task_pool( dummy_slot.task_pool ); |
| dummy_slot.task_pool = NULL; |
| // Update small_task_count last. Doing so sooner might cause another thread to free *this. |
| __TBB_ASSERT( small_task_count>=k, "small_task_count corrupted" ); |
| governor::sign_off(this); |
| if( __TBB_FetchAndAddW( &small_task_count, -k )==k ) |
| NFS_Free( this ); |
| } |
| |
| task& generic_scheduler::allocate_task( size_t number_of_bytes, |
| __TBB_CONTEXT_ARG(task* parent, task_group_context* context) ) { |
| GATHER_STATISTIC(current_active+=1); |
| task* t = free_list; |
| if( number_of_bytes<=quick_task_size ) { |
| if( t ) { |
| GATHER_STATISTIC(current_length-=1); |
| __TBB_ASSERT( t->state()==task::freed, "free list of tasks is corrupted" ); |
| free_list = t->prefix().next; |
| } else if( return_list ) { |
| // No fence required for read of return_list above, because __TBB_FetchAndStoreW has a fence. |
| t = (task*)__TBB_FetchAndStoreW( &return_list, 0 ); |
| __TBB_ASSERT( t, "another thread emptied the return_list" ); |
| __TBB_ASSERT( t->prefix().origin==this, "task returned to wrong return_list" ); |
| ITT_NOTIFY( sync_acquired, &return_list ); |
| free_list = t->prefix().next; |
| } else { |
| t = (task*)((char*)NFS_Allocate( task_prefix_reservation_size+quick_task_size, 1, NULL ) + task_prefix_reservation_size ); |
| #if __TBB_COUNT_TASK_NODES |
| ++task_node_count; |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| t->prefix().origin = this; |
| ++small_task_count; |
| } |
| } else { |
| GATHER_STATISTIC(current_big_malloc+=1); |
| t = (task*)((char*)NFS_Allocate( task_prefix_reservation_size+number_of_bytes, 1, NULL ) + task_prefix_reservation_size ); |
| #if __TBB_COUNT_TASK_NODES |
| ++task_node_count; |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| t->prefix().origin = NULL; |
| } |
| task_prefix& p = t->prefix(); |
| #if __TBB_TASK_GROUP_CONTEXT |
| p.context = context; |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| p.owner = this; |
| p.ref_count = 0; |
| // Assign some not outrageously out-of-place value for a while |
| p.depth = 0; |
| p.parent = parent; |
| // In TBB 2.1 and later, the constructor for task sets extra_state to indicate the version of the tbb/task.h header. |
| // In TBB 2.0 and earlier, the constructor leaves extra_state as zero. |
| p.extra_state = 0; |
| p.affinity = 0; |
| p.state = task::allocated; |
| return *t; |
| } |
| |
| void generic_scheduler::free_nonlocal_small_task( task& t ) { |
| __TBB_ASSERT( t.state()==task::freed, NULL ); |
| generic_scheduler& s = *static_cast<generic_scheduler*>(t.prefix().origin); |
| __TBB_ASSERT( &s!=this, NULL ); |
| for(;;) { |
| task* old = s.return_list; |
| if( old==plugged_return_list() ) |
| break; |
| // Atomically insert t at head of s.return_list |
| t.prefix().next = old; |
| ITT_NOTIFY( sync_releasing, &s.return_list ); |
| if( __TBB_CompareAndSwapW( &s.return_list, (intptr_t)&t, (intptr_t)old )==(intptr_t)old ) |
| return; |
| } |
| deallocate_task(t); |
| if( __TBB_FetchAndDecrementWrelease( &s.small_task_count )==1 ) { |
| // We freed the last task allocated by scheduler s, so it's our responsibility |
| // to free the scheduler. |
| NFS_Free( &s ); |
| } |
| } |
| |
| task** generic_scheduler::allocate_task_pool( size_t n ) { |
| __TBB_ASSERT( n > task_pool_size, "Cannot shrink the task pool" ); |
| size_t byte_size = ((n * sizeof(task*) + NFS_MaxLineSize - 1) / NFS_MaxLineSize) * NFS_MaxLineSize; |
| task_pool_size = byte_size / sizeof(task*); |
| task** new_pool = (task**)NFS_Allocate( byte_size, 1, NULL ); |
| // No need to clear the fresh deque since valid items are designated by the head and tail members. |
| #if TBB_USE_ASSERT>=2 |
| // But clear it in the high vigilance debug mode |
| memset( new_pool, -1, n ); |
| #endif /* TBB_USE_ASSERT>=2 */ |
| return new_pool; |
| } |
| |
| void generic_scheduler::grow_task_pool( size_t new_size ) { |
| __TBB_ASSERT( assert_okay(), NULL ); |
| if ( new_size < 2 * task_pool_size ) |
| new_size = 2 * task_pool_size; |
| task** new_pool = allocate_task_pool( new_size ); // updates task_pool_size |
| task** old_pool = dummy_slot.task_pool; |
| acquire_task_pool(); // requires the old dummy_slot.task_pool value |
| // my_arena_slot->tail should not be updated before my_arena_slot->head because their |
| // values are used by other threads to check if this task pool is empty. |
| size_t new_tail = my_arena_slot->tail - my_arena_slot->head; |
| __TBB_ASSERT( new_tail <= task_pool_size, "new task pool is too short" ); |
| memcpy( new_pool, old_pool + my_arena_slot->head, new_tail * sizeof(task*) ); |
| my_arena_slot->head = 0; |
| my_arena_slot->tail = new_tail; |
| dummy_slot.task_pool = new_pool; |
| release_task_pool(); // updates the task pool pointer in our arena slot |
| free_task_pool( old_pool ); |
| __TBB_ASSERT( assert_okay(), NULL ); |
| } |
| |
| /** ATTENTION: |
| This method is mostly the same as generic_scheduler::lock_task_pool(), with |
| a little different logic of slot state checks (slot is either locked or points |
| to our task pool). |
| Thus if either of them is changed, consider changing the counterpart as well. **/ |
| inline void generic_scheduler::acquire_task_pool() const { |
| if ( !in_arena() ) |
| return; // we are not in arena - nothing to lock |
| atomic_backoff backoff; |
| bool sync_prepare_done = false; |
| for(;;) { |
| #if TBB_USE_ASSERT |
| __TBB_ASSERT( my_arena_slot == my_arena->slot + arena_index, "invalid arena slot index" ); |
| // Local copy of the arena slot task pool pointer is necessary for the next |
| // assertion to work correctly to exclude asynchronous state transition effect. |
| task** tp = my_arena_slot->task_pool; |
| __TBB_ASSERT( tp == LockedTaskPool || tp == dummy_slot.task_pool, "slot ownership corrupt?" ); |
| #endif |
| if( my_arena_slot->task_pool != LockedTaskPool && |
| __TBB_CompareAndSwapW( &my_arena_slot->task_pool, (intptr_t)LockedTaskPool, |
| (intptr_t)dummy_slot.task_pool ) == (intptr_t)dummy_slot.task_pool ) |
| { |
| // We acquired our own slot |
| ITT_NOTIFY(sync_acquired, my_arena_slot); |
| break; |
| } |
| else if( !sync_prepare_done ) { |
| // Start waiting |
| ITT_NOTIFY(sync_prepare, my_arena_slot); |
| sync_prepare_done = true; |
| } |
| // Someone else acquired a lock, so pause and do exponential backoff. |
| backoff.pause(); |
| } |
| __TBB_ASSERT( my_arena_slot->task_pool == LockedTaskPool, "not really acquired task pool" ); |
| } // generic_scheduler::acquire_task_pool |
| |
| inline void generic_scheduler::release_task_pool() const { |
| if ( !in_arena() ) |
| return; // we are not in arena - nothing to unlock |
| __TBB_ASSERT( my_arena_slot, "we are not in arena" ); |
| __TBB_ASSERT( my_arena_slot->task_pool == LockedTaskPool, "arena slot is not locked" ); |
| ITT_NOTIFY(sync_releasing, my_arena_slot); |
| __TBB_store_with_release( my_arena_slot->task_pool, dummy_slot.task_pool ); |
| } |
| |
| /** ATTENTION: |
| This method is mostly the same as generic_scheduler::acquire_task_pool(), |
| with a little different logic of slot state checks (slot can be empty, locked |
| or point to any task pool other than ours, and asynchronous transitions between |
| all these states are possible). |
| Thus if any of them is changed, consider changing the counterpart as well **/ |
| inline task** generic_scheduler::lock_task_pool( arena_slot* victim_arena_slot ) const { |
| task** victim_task_pool; |
| atomic_backoff backoff; |
| bool sync_prepare_done = false; |
| for(;;) { |
| victim_task_pool = victim_arena_slot->task_pool; |
| // TODO: Investigate the effect of bailing out on the locked pool without trying to lock it. |
| // When doing this update assertion in the end of the method. |
| if ( victim_task_pool == EmptyTaskPool ) { |
| // The victim thread emptied its task pool - nothing to lock |
| if( sync_prepare_done ) |
| ITT_NOTIFY(sync_cancel, victim_arena_slot); |
| break; |
| } |
| if( victim_task_pool != LockedTaskPool && |
| __TBB_CompareAndSwapW( &victim_arena_slot->task_pool, |
| (intptr_t)LockedTaskPool, (intptr_t)victim_task_pool ) == (intptr_t)victim_task_pool ) |
| { |
| // We've locked victim's task pool |
| ITT_NOTIFY(sync_acquired, victim_arena_slot); |
| break; |
| } |
| else if( !sync_prepare_done ) { |
| // Start waiting |
| ITT_NOTIFY(sync_prepare, victim_arena_slot); |
| sync_prepare_done = true; |
| } |
| // Someone else acquired a lock, so pause and do exponential backoff. |
| backoff.pause(); |
| } |
| __TBB_ASSERT( victim_task_pool == EmptyTaskPool || |
| (victim_arena_slot->task_pool == LockedTaskPool && victim_task_pool != LockedTaskPool), |
| "not really locked victim's task pool?" ); |
| return victim_task_pool; |
| } // generic_scheduler::lock_task_pool |
| |
| inline void generic_scheduler::unlock_task_pool( arena_slot* victim_arena_slot, |
| task** victim_task_pool ) const { |
| __TBB_ASSERT( victim_arena_slot, "empty victim arena slot pointer" ); |
| __TBB_ASSERT( victim_arena_slot->task_pool == LockedTaskPool, "victim arena slot is not locked" ); |
| ITT_NOTIFY(sync_releasing, victim_arena_slot); |
| __TBB_store_with_release( victim_arena_slot->task_pool, victim_task_pool ); |
| } |
| |
| |
| inline task* generic_scheduler::prepare_for_spawning( task* t ) { |
| __TBB_ASSERT( t->state()==task::allocated, "attempt to spawn task that is not in 'allocated' state" ); |
| t->prefix().owner = this; |
| t->prefix().state = task::ready; |
| #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 spawn task whose parent has a ref_count<0" ); |
| __TBB_ASSERT( ref_count!=0, "attempt to spawn task whose parent has a ref_count==0 (forgot to set_ref_count?)" ); |
| parent->prefix().extra_state |= es_ref_count_active; |
| } |
| #endif /* TBB_USE_ASSERT */ |
| affinity_id dst_thread = t->prefix().affinity; |
| __TBB_ASSERT( dst_thread == 0 || is_version_3_task(*t), "backwards compatibility to TBB 2.0 tasks is broken" ); |
| if( dst_thread != 0 && dst_thread != my_affinity_id ) { |
| task_proxy& proxy = (task_proxy&)allocate_task( sizeof(task_proxy), |
| __TBB_CONTEXT_ARG(NULL, NULL) ); |
| // Mark as a proxy |
| proxy.prefix().extra_state = es_task_proxy; |
| proxy.outbox = &my_arena->mailbox(dst_thread); |
| proxy.task_and_tag = intptr_t(t)|3; |
| ITT_NOTIFY( sync_releasing, proxy.outbox ); |
| // Mail the proxy - after this point t may be destroyed by another thread at any moment. |
| proxy.outbox->push(proxy); |
| return &proxy; |
| } |
| return t; |
| } |
| |
| /** Conceptually, this method should be a member of class scheduler. |
| But doing so would force us to publish class scheduler in the headers. */ |
| void generic_scheduler::local_spawn( task& first, task*& next ) { |
| __TBB_ASSERT( governor::is_set(this), NULL ); |
| __TBB_ASSERT( assert_okay(), NULL ); |
| if ( &first.prefix().next == &next ) { |
| // Single task is being spawned |
| if ( my_arena_slot->tail == task_pool_size ) { |
| // 1 compensates for head possibly temporarily incremented by a thief |
| if ( my_arena_slot->head > 1 ) { |
| // Move the busy part of the deque to the beginning of the allocated space |
| acquire_task_pool(); |
| my_arena_slot->tail -= my_arena_slot->head; |
| memmove( dummy_slot.task_pool, dummy_slot.task_pool + my_arena_slot->head, my_arena_slot->tail * sizeof(task*) ); |
| my_arena_slot->head = 0; |
| release_task_pool(); |
| } |
| else { |
| grow_task_pool( task_pool_size + 1 ); |
| } |
| } |
| dummy_slot.task_pool[my_arena_slot->tail] = prepare_for_spawning( &first ); |
| ITT_NOTIFY(sync_releasing, my_arena_slot); |
| // The following store with release is required on ia64 only |
| size_t new_tail = my_arena_slot->tail + 1; |
| __TBB_store_with_release( my_arena_slot->tail, new_tail ); |
| __TBB_ASSERT ( my_arena_slot->tail <= task_pool_size, "task deque end was overwritten" ); |
| } |
| else { |
| // Task list is being spawned |
| const size_t initial_capacity = 64; |
| task *arr[initial_capacity]; |
| fast_reverse_vector<task*> tasks(arr, initial_capacity); |
| task *t_next = NULL; |
| for( task* t = &first; ; t = t_next ) { |
| // After prepare_for_spawning returns t may already have been destroyed. |
| // So milk it while it is alive. |
| bool end = &t->prefix().next == &next; |
| t_next = t->prefix().next; |
| tasks.push_back( prepare_for_spawning(t) ); |
| if( end ) |
| break; |
| } |
| size_t num_tasks = tasks.size(); |
| __TBB_ASSERT ( arena_index != null_arena_index, "invalid arena slot index" ); |
| if ( my_arena_slot->tail + num_tasks > task_pool_size ) { |
| // 1 compensates for head possibly temporarily incremented by a thief |
| size_t new_size = my_arena_slot->tail - my_arena_slot->head + num_tasks + 1; |
| if ( new_size <= task_pool_size ) { |
| // Move the busy part of the deque to the beginning of the allocated space |
| acquire_task_pool(); |
| my_arena_slot->tail -= my_arena_slot->head; |
| memmove( dummy_slot.task_pool, dummy_slot.task_pool + my_arena_slot->head, my_arena_slot->tail * sizeof(task*) ); |
| my_arena_slot->head = 0; |
| release_task_pool(); |
| } |
| else { |
| grow_task_pool( new_size ); |
| } |
| } |
| #if DO_ITT_NOTIFY |
| else { |
| // The preceding if-branch issues the same ittnotify inside release_task_pool() or grow_task_pool() methods |
| ITT_NOTIFY(sync_releasing, my_arena_slot); |
| } |
| #endif /* DO_ITT_NOTIFY */ |
| tasks.copy_memory( dummy_slot.task_pool + my_arena_slot->tail ); |
| // The following store with release is required on ia64 only |
| size_t new_tail = my_arena_slot->tail + num_tasks; |
| __TBB_store_with_release( my_arena_slot->tail, new_tail ); |
| __TBB_ASSERT ( my_arena_slot->tail <= task_pool_size, "task deque end was overwritten" ); |
| } |
| #if __TBB_ARENA_PER_MASTER |
| if ( !in_arena() ) |
| enter_arena(); |
| my_arena->advertise_new_work</*Spawned=*/true>(); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| if ( !in_arena() ) { |
| if ( is_worker() ) |
| enter_arena(); |
| else |
| try_enter_arena(); |
| } |
| my_arena->mark_pool_full(); |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| __TBB_ASSERT( assert_okay(), NULL ); |
| } |
| |
| void generic_scheduler::local_spawn_root_and_wait( task& first, task*& next ) { |
| __TBB_ASSERT( governor::is_set(this), NULL ); |
| __TBB_ASSERT( &first, NULL ); |
| auto_empty_task dummy( __TBB_CONTEXT_ARG(this, first.prefix().context) ); |
| internal::reference_count n = 0; |
| for( task* t=&first; ; t=t->prefix().next ) { |
| ++n; |
| __TBB_ASSERT( !t->prefix().parent, "not a root task, or already running" ); |
| t->prefix().parent = &dummy; |
| if( &t->prefix().next==&next ) break; |
| #if __TBB_TASK_GROUP_CONTEXT |
| __TBB_ASSERT( t->prefix().context == t->prefix().next->prefix().context, |
| "all the root tasks in list must share the same context"); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| } |
| dummy.prefix().ref_count = n+1; |
| if( n>1 ) |
| local_spawn( *first.prefix().next, next ); |
| local_wait_for_all( dummy, &first ); |
| } |
| |
| inline task* generic_scheduler::get_mailbox_task() { |
| __TBB_ASSERT( my_affinity_id>0, "not in arena" ); |
| task* result = NULL; |
| while( task_proxy* t = inbox.pop() ) { |
| intptr_t tat = __TBB_load_with_acquire(t->task_and_tag); |
| __TBB_ASSERT( tat==task_proxy::mailbox_bit || (tat==(tat|3)&&tat!=3), NULL ); |
| if( tat!=task_proxy::mailbox_bit && __TBB_CompareAndSwapW( &t->task_and_tag, task_proxy::pool_bit, tat )==tat ) { |
| // Successfully grabbed the task, and left pool seeker with job of freeing the proxy. |
| ITT_NOTIFY( sync_acquired, inbox.outbox() ); |
| result = (task*)(tat & ~3); |
| result->prefix().owner = this; |
| break; |
| } |
| free_task_proxy( *t ); |
| } |
| return result; |
| } |
| |
| inline task* generic_scheduler::strip_proxy( task_proxy* tp ) { |
| __TBB_ASSERT( tp->prefix().extra_state==es_task_proxy, NULL ); |
| intptr_t tat = __TBB_load_with_acquire(tp->task_and_tag); |
| if( (tat&3)==3 ) { |
| // proxy is shared by a pool and a mailbox. |
| // Attempt to transition it to "empty proxy in mailbox" state. |
| if( __TBB_CompareAndSwapW( &tp->task_and_tag, task_proxy::mailbox_bit, tat )==tat ) { |
| // Successfully grabbed the task, and left the mailbox with the job of freeing the proxy. |
| return (task*)(tat&~3); |
| } |
| __TBB_ASSERT( tp->task_and_tag==task_proxy::pool_bit, NULL ); |
| } else { |
| // We have exclusive access to the proxy |
| __TBB_ASSERT( (tat&3)==task_proxy::pool_bit, "task did not come from pool?" ); |
| __TBB_ASSERT ( !(tat&~3), "Empty proxy in the pool contains non-zero task pointer" ); |
| } |
| #if TBB_USE_ASSERT |
| tp->prefix().state = task::allocated; |
| #endif |
| free_task_proxy( *tp ); |
| // Another thread grabbed the underlying task via their mailbox |
| return NULL; |
| } |
| |
| #if __TBB_ARENA_PER_MASTER |
| void generic_scheduler::local_enqueue( task& t ) { |
| __TBB_ASSERT( governor::is_set(this), NULL ); |
| __TBB_ASSERT( t.state()==task::allocated, "attempt to enqueue task that is not in 'allocated' state" ); |
| t.prefix().owner = this; |
| t.prefix().state = task::ready; |
| |
| #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" ); |
| __TBB_ASSERT( ref_count!=0, "attempt to enqueue task whose parent has a ref_count==0 (forgot to set_ref_count?)" ); |
| 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 */ |
| |
| __TBB_ASSERT( my_arena, "thread is not in any arena" ); |
| ITT_NOTIFY(sync_releasing, &my_arena->my_task_stream); |
| my_arena->my_task_stream.push( &t, my_arena_slot->hint_for_push ); |
| my_arena->advertise_new_work< /*Spawned=*/ false >(); |
| __TBB_ASSERT( assert_okay(), NULL ); |
| } |
| |
| inline task* generic_scheduler::dequeue_task() { |
| task* result = NULL; |
| my_arena->my_task_stream.pop(result, my_arena_slot->hint_for_pop); |
| if (result) ITT_NOTIFY(sync_acquired, &my_arena->my_task_stream); |
| return result; |
| } |
| #endif /* __TBB_ARENA_PER_MASTER */ |
| |
| inline task* generic_scheduler::get_task() { |
| task* result = NULL; |
| retry: |
| --my_arena_slot->tail; |
| __TBB_rel_acq_fence(); |
| if ( (intptr_t)my_arena_slot->head > (intptr_t)my_arena_slot->tail ) { |
| acquire_task_pool(); |
| if ( (intptr_t)my_arena_slot->head <= (intptr_t)my_arena_slot->tail ) { |
| // The thief backed off - grab the task |
| __TBB_ASSERT_VALID_TASK_PTR( dummy_slot.task_pool[my_arena_slot->tail] ); |
| result = dummy_slot.task_pool[my_arena_slot->tail]; |
| __TBB_POISON_TASK_PTR( dummy_slot.task_pool[my_arena_slot->tail] ); |
| } |
| else { |
| __TBB_ASSERT ( my_arena_slot->head == my_arena_slot->tail + 1, "victim/thief arbitration algorithm failure" ); |
| } |
| if ( (intptr_t)my_arena_slot->head < (intptr_t)my_arena_slot->tail ) { |
| release_task_pool(); |
| } |
| else { |
| // In any case the deque is empty now, so compact it |
| my_arena_slot->head = my_arena_slot->tail = 0; |
| if ( in_arena() ) |
| leave_arena(); |
| } |
| } |
| else { |
| __TBB_ASSERT_VALID_TASK_PTR( dummy_slot.task_pool[my_arena_slot->tail] ); |
| result = dummy_slot.task_pool[my_arena_slot->tail]; |
| __TBB_POISON_TASK_PTR( dummy_slot.task_pool[my_arena_slot->tail] ); |
| } |
| if( result && is_proxy(*result) ) { |
| result = strip_proxy((task_proxy*)result); |
| if( !result ) { |
| goto retry; |
| } |
| GATHER_STATISTIC( ++proxy_execute_count ); |
| // Following assertion should be true because TBB 2.0 tasks never specify affinity, and hence are not proxied. |
| __TBB_ASSERT( is_version_3_task(*result), "backwards compatibility with TBB 2.0 broken" ); |
| // Task affinity has changed. |
| innermost_running_task = result; |
| result->note_affinity(my_affinity_id); |
| } |
| return result; |
| } // generic_scheduler::get_task |
| |
| task* generic_scheduler::steal_task( arena_slot& victim_slot ) { |
| task** victim_pool = lock_task_pool( &victim_slot ); |
| if ( !victim_pool ) |
| return NULL; |
| const size_t none = ~size_t(0); |
| size_t first_skipped_proxy = none; |
| task* result = NULL; |
| retry: |
| ++victim_slot.head; |
| __TBB_rel_acq_fence(); |
| if ( (intptr_t)victim_slot.head > (intptr_t)victim_slot.tail ) { |
| --victim_slot.head; |
| } |
| else { |
| __TBB_ASSERT_VALID_TASK_PTR( victim_pool[victim_slot.head - 1]); |
| result = victim_pool[victim_slot.head - 1]; |
| if( is_proxy(*result) ) { |
| task_proxy& tp = *static_cast<task_proxy*>(result); |
| // If task will likely be grabbed by whom it was mailed to, skip it. |
| if( (tp.task_and_tag & 3) == 3 && tp.outbox->recipient_is_idle() ) { |
| if ( first_skipped_proxy == none ) |
| first_skipped_proxy = victim_slot.head - 1; |
| result = NULL; |
| goto retry; |
| } |
| } |
| __TBB_POISON_TASK_PTR(victim_pool[victim_slot.head - 1]); |
| } |
| if ( first_skipped_proxy != none ) { |
| if ( result ) { |
| victim_pool[victim_slot.head - 1] = victim_pool[first_skipped_proxy]; |
| __TBB_POISON_TASK_PTR( victim_pool[first_skipped_proxy] ); |
| __TBB_store_with_release( victim_slot.head, first_skipped_proxy + 1 ); |
| } |
| else |
| __TBB_store_with_release( victim_slot.head, first_skipped_proxy ); |
| } |
| unlock_task_pool( &victim_slot, victim_pool ); |
| return result; |
| } |
| |
| inline void generic_scheduler::do_enter_arena() { |
| my_arena_slot = &my_arena->slot[arena_index]; |
| __TBB_ASSERT ( my_arena_slot->head == my_arena_slot->tail, "task deque of a free slot must be empty" ); |
| __TBB_ASSERT ( dummy_slot.head < dummy_slot.tail, "entering arena without tasks to share" ); |
| my_arena_slot->head = dummy_slot.head; |
| my_arena_slot->tail = dummy_slot.tail; |
| // Release signal on behalf of previously spawned tasks (when this thread was not in arena yet) |
| ITT_NOTIFY(sync_releasing, my_arena_slot); |
| __TBB_store_with_release( my_arena_slot->task_pool, dummy_slot.task_pool ); |
| // We'll leave arena only when it's empty, so clean up local instances of indices. |
| dummy_slot.head = dummy_slot.tail = 0; |
| } |
| |
| void generic_scheduler::enter_arena() { |
| __TBB_ASSERT ( my_arena, "no arena: initialization not completed?" ); |
| #if __TBB_ARENA_PER_MASTER |
| __TBB_ASSERT ( !in_arena(), "thread is already in arena?" ); |
| __TBB_ASSERT ( arena_index < my_arena->my_num_slots, "arena slot index is out-of-bound" ); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| __TBB_ASSERT ( is_worker(), "only workers should use enter_arena()" ); |
| __TBB_ASSERT ( !in_arena(), "worker already in arena?" ); |
| __TBB_ASSERT ( arena_index < my_arena->prefix().number_of_workers, "invalid worker arena slot index" ); |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| __TBB_ASSERT ( my_arena->slot[arena_index].task_pool == EmptyTaskPool, "someone else grabbed my arena slot?" ); |
| do_enter_arena(); |
| } |
| |
| #if !__TBB_ARENA_PER_MASTER |
| void generic_scheduler::try_enter_arena() { |
| __TBB_ASSERT ( !is_worker(), "only masters should use try_enter_arena()" ); |
| __TBB_ASSERT ( my_arena, "no arena: initialization not completed?" ); |
| __TBB_ASSERT ( !in_arena(), "master already in arena?" ); |
| __TBB_ASSERT ( arena_index >= my_arena->prefix().number_of_workers && |
| arena_index < my_arena->prefix().number_of_slots, "invalid arena slot hint value" ); |
| |
| size_t h = arena_index; |
| // We do not lock task pool upon successful entering arena |
| if( my_arena->slot[h].task_pool != EmptyTaskPool || |
| __TBB_CompareAndSwapW( &my_arena->slot[h].task_pool, (intptr_t)LockedTaskPool, |
| (intptr_t)EmptyTaskPool ) != (intptr_t)EmptyTaskPool ) |
| { |
| // Hinted arena slot is already busy, try some of the others at random |
| unsigned first = my_arena->prefix().number_of_workers, |
| last = my_arena->prefix().number_of_slots; |
| unsigned n = last - first - 1; |
| /// \todo Is this limit reasonable? |
| size_t max_attempts = last - first; |
| for (;;) { |
| size_t k = first + random.get() % n; |
| if( k >= h ) |
| ++k; // Adjusts random distribution to exclude previously tried slot |
| h = k; |
| if( my_arena->slot[h].task_pool == EmptyTaskPool && |
| __TBB_CompareAndSwapW( &my_arena->slot[h].task_pool, (intptr_t)LockedTaskPool, |
| (intptr_t)EmptyTaskPool ) == (intptr_t)EmptyTaskPool ) |
| { |
| break; |
| } |
| if ( --max_attempts == 0 ) { |
| // After so many attempts we are still unable to find a vacant arena slot. |
| // Cease the vain effort and work outside of arena for a while. |
| return; |
| } |
| } |
| } |
| // Successfully claimed a slot in the arena. |
| ITT_NOTIFY(sync_acquired, &my_arena->slot[h]); |
| __TBB_ASSERT ( my_arena->slot[h].task_pool == LockedTaskPool, "arena slot is not actually acquired" ); |
| arena_index = h; |
| do_enter_arena(); |
| attach_mailbox( affinity_id(h+1) ); |
| } |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| |
| void generic_scheduler::leave_arena() { |
| __TBB_ASSERT( in_arena(), "Not in arena" ); |
| // Do not reset arena_index. It will be used to (attempt to) re-acquire the slot next time |
| __TBB_ASSERT( &my_arena->slot[arena_index] == my_arena_slot, "arena slot and slot index mismatch" ); |
| __TBB_ASSERT ( my_arena_slot->task_pool == LockedTaskPool, "Task pool must be locked when leaving arena" ); |
| __TBB_ASSERT ( my_arena_slot->head == my_arena_slot->tail, "Cannot leave arena when the task pool is not empty" ); |
| #if !__TBB_ARENA_PER_MASTER |
| if ( !is_worker() ) { |
| my_affinity_id = 0; |
| inbox.detach(); |
| } |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| ITT_NOTIFY(sync_releasing, &my_arena->slot[arena_index]); |
| __TBB_store_with_release( my_arena_slot->task_pool, EmptyTaskPool ); |
| my_arena_slot = &dummy_slot; |
| } |
| |
| #if __TBB_ARENA_PER_MASTER |
| generic_scheduler* generic_scheduler::create_worker( market& m, size_t index ) { |
| generic_scheduler* s = allocate_scheduler( NULL, index ); |
| #if __TBB_TASK_GROUP_CONTEXT |
| s->dummy_task->prefix().context = &dummy_context; |
| // Sync up the local cancellation state with the global one. No need for fence here. |
| s->local_cancel_count = global_cancel_count; |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| s->my_market = &m; |
| s->init_stack_info(); |
| return s; |
| } |
| |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| |
| generic_scheduler* generic_scheduler::create_worker( arena& a, size_t index ) { |
| generic_scheduler* s = allocate_scheduler( &a, index ); |
| |
| // Put myself into the arena |
| #if __TBB_TASK_GROUP_CONTEXT |
| s->dummy_task->prefix().context = &dummy_context; |
| // Sync up the local cancellation state with the global one. No need for fence here. |
| s->local_cancel_count = global_cancel_count; |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| s->attach_mailbox( index+1 ); |
| s->init_stack_info(); |
| |
| __TBB_store_with_release( a.prefix().worker_list[index].scheduler, s ); |
| return s; |
| } |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| |
| generic_scheduler* generic_scheduler::create_master( arena& a ) { |
| generic_scheduler* s = allocate_scheduler( &a, |
| #if __TBB_ARENA_PER_MASTER |
| 0 // Master thread always occupies the first slot |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| null_arena_index // Master thread will have to search for a vacant slot |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| ); |
| task& t = *s->dummy_task; |
| s->innermost_running_task = &t; |
| t.prefix().ref_count = 1; |
| governor::sign_on(s); |
| __TBB_ASSERT( &task::self()==&t, "governor::sign_on failed?" ); |
| #if __TBB_ARENA_PER_MASTER |
| #if __TBB_TASK_GROUP_CONTEXT |
| // Context to be used by root tasks by default (if the user has not specified one). |
| // Allocation is done by NFS allocator because we cannot reuse memory allocated |
| // for task objects since the free list is empty at the moment. |
| t.prefix().context = a.my_master_default_ctx = |
| new ( NFS_Allocate(sizeof(task_group_context), 1, NULL) ) task_group_context(task_group_context::isolated); |
| #endif |
| s->my_market = a.my_market; |
| __TBB_ASSERT( s->arena_index == 0, "Master thread must occupy the first slot in its arena" ); |
| s->attach_mailbox(1); |
| a.slot[0].my_scheduler = s; |
| #if _WIN32|_WIN64 |
| __TBB_ASSERT( s->my_market, NULL ); |
| s->my_market->register_master( s->master_exec_resource ); |
| #endif /* _WIN32|_WIN64 */ |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| #if _WIN32|_WIN64 |
| s->register_master(); |
| #endif |
| #if __TBB_TASK_GROUP_CONTEXT |
| // Context to be used by root tasks by default (if the user has not specified one). |
| // Allocation is done by NFS allocator because we cannot reuse memory allocated |
| // for task objects since the free list is empty at the moment. |
| t.prefix().context = new ( NFS_Allocate(sizeof(task_group_context), 1, NULL) ) task_group_context(task_group_context::isolated); |
| scheduler_list_node_t &node = s->my_node; |
| { |
| mutex::scoped_lock lock(the_scheduler_list_mutex); |
| node.my_next = the_scheduler_list_head.my_next; |
| node.my_prev = &the_scheduler_list_head; |
| the_scheduler_list_head.my_next->my_prev = &node; |
| the_scheduler_list_head.my_next = &node; |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| unsigned last = a.prefix().number_of_slots, |
| cur_limit = a.prefix().limit; |
| // This slot index assignment is just a hint to ... |
| if ( cur_limit < last ) { |
| // ... to prevent competition between the first few masters. |
| s->arena_index = cur_limit++; |
| // In the absence of exception handling this code is a subject to data |
| // race in case of multiple masters concurrently entering empty arena. |
| // But it does not affect correctness, and can only result in a few |
| // masters competing for the same arena slot during the first acquisition. |
| // The cost of competition is low in comparison to that of oversubscription. |
| a.prefix().limit = cur_limit; |
| } |
| else { |
| // ... to minimize the probability of competition between multiple masters. |
| unsigned first = a.prefix().number_of_workers; |
| s->arena_index = first + s->random.get() % (last - first); |
| } |
| #if __TBB_TASK_GROUP_CONTEXT |
| } |
| #endif |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| s->init_stack_info(); |
| #if __TBB_TASK_GROUP_CONTEXT |
| // Sync up the local cancellation state with the global one. No need for fence here. |
| s->local_cancel_count = global_cancel_count; |
| #endif |
| #if __TBB_SCHEDULER_OBSERVER |
| // Process any existing observers. |
| s->notify_entry_observers(); |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| return s; |
| } |
| |
| void generic_scheduler::cleanup_worker( void* arg, bool is_worker ) { |
| generic_scheduler& s = *(generic_scheduler*)arg; |
| __TBB_ASSERT( s.dummy_slot.task_pool, "cleaning up worker with missing task pool" ); |
| // APM TODO: Decide how observers should react to each entry/leave to/from arena |
| #if __TBB_SCHEDULER_OBSERVER |
| s.notify_exit_observers( is_worker ); |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| // When comparing "head" and "tail" indices ">=" is used because this worker's |
| // task pool may still be published in the arena, and thieves can optimistically |
| // bump "head" (and then roll back). |
| __TBB_ASSERT( s.my_arena_slot->task_pool == EmptyTaskPool || s.my_arena_slot->head >= s.my_arena_slot->tail, |
| "worker has unfinished work at run down" ); |
| s.free_scheduler(); |
| } |
| |
| void generic_scheduler::cleanup_master() { |
| generic_scheduler& s = *this; // for similarity with cleanup_worker |
| __TBB_ASSERT( s.dummy_slot.task_pool, "cleaning up master with missing task pool" ); |
| #if __TBB_SCHEDULER_OBSERVER |
| s.notify_exit_observers(/*is_worker=*/false); |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| if ( !local_task_pool_empty() ) { |
| __TBB_ASSERT ( governor::is_set(this), "TLS slot is cleared before the task pool cleanup" ); |
| s.local_wait_for_all( *s.dummy_task, NULL ); |
| __TBB_ASSERT ( governor::is_set(this), "Other thread reused our TLS key during the task pool cleanup" ); |
| } |
| #if __TBB_ARENA_PER_MASTER |
| #if _WIN32|_WIN64 |
| __TBB_ASSERT( s.my_market, NULL ); |
| s.my_market->unregister_master( s.master_exec_resource ); |
| #endif /* _WIN32|_WIN64 */ |
| arena* a = s.my_arena; |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| #if _WIN32|_WIN64 |
| s.unregister_master(); |
| #endif /* _WIN32|_WIN64 */ |
| #endif /* __TBB_ARENA_PER_MASTER */ |
| s.free_scheduler(); |
| #if __TBB_ARENA_PER_MASTER |
| a->slot[0].my_scheduler = NULL; |
| // Do not close arena if some fire-and-forget tasks remain; workers should care of it. |
| if( a->my_task_stream.empty() && a->pool_state.fetch_and_store(arena::SNAPSHOT_EMPTY)!=arena::SNAPSHOT_EMPTY ) |
| a->my_market->adjust_demand( *a, -(int)a->my_max_num_workers ); |
| if ( --a->my_num_threads_active==0 && a->pool_state==arena::SNAPSHOT_EMPTY ) |
| a->close_arena(); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| governor::finish_with_arena(); |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| } |
| |
| #if __TBB_SCHEDULER_OBSERVER |
| void generic_scheduler::notify_entry_observers() { |
| local_last_observer_proxy = observer_proxy::process_list(local_last_observer_proxy,is_worker(),/*is_entry=*/true); |
| } |
| |
| void generic_scheduler::notify_exit_observers( bool is_worker ) { |
| observer_proxy::process_list(local_last_observer_proxy,is_worker,/*is_entry=*/false); |
| } |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| |
| } // namespace internal |
| } // namespace tbb |
| |
