| /* |
| 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. |
| */ |
| |
| #ifndef _TBB_custom_scheduler_H |
| #define _TBB_custom_scheduler_H |
| |
| #include "scheduler.h" |
| #include "observer_proxy.h" |
| #include "itt_notify.h" |
| |
| namespace tbb { |
| namespace internal { |
| |
| //! Amount of time to pause between steals. |
| /** The default values below were found to be best empirically for K-Means |
| on the 32-way Altix and 4-way (*2 for HT) fxqlin04. */ |
| #if __TBB_ipf |
| static const long PauseTime = 1500; |
| #else |
| static const long PauseTime = 80; |
| #endif |
| |
| //------------------------------------------------------------------------ |
| //! Traits classes for scheduler |
| //------------------------------------------------------------------------ |
| |
| struct DefaultSchedulerTraits { |
| static const bool itt_possible = true; |
| static const bool has_slow_atomic = false; |
| }; |
| |
| struct IntelSchedulerTraits { |
| static const bool itt_possible = false; |
| #if __TBB_x86_32||__TBB_x86_64 |
| static const bool has_slow_atomic = true; |
| #else |
| static const bool has_slow_atomic = false; |
| #endif /* __TBB_x86_32||__TBB_x86_64 */ |
| }; |
| |
| //------------------------------------------------------------------------ |
| // custom_scheduler |
| //------------------------------------------------------------------------ |
| |
| //! A scheduler with a customized evaluation loop. |
| /** The customization can use SchedulerTraits to make decisions without needing a run-time check. */ |
| template<typename SchedulerTraits> |
| class custom_scheduler: private generic_scheduler { |
| typedef custom_scheduler<SchedulerTraits> scheduler_type; |
| |
| //! Scheduler loop that dispatches tasks. |
| /** If child is non-NULL, it is dispatched first. |
| Then, until "parent" has a reference count of 1, other task are dispatched or stolen. */ |
| /*override*/ |
| void local_wait_for_all( task& parent, task* child ); |
| |
| //! Entry point from client code to the scheduler loop that dispatches tasks. |
| /** The method is virtual, but the *this object is used only for sake of dispatching on the correct vtable, |
| not necessarily the correct *this object. The correct *this object is looked up in TLS. */ |
| /*override*/ |
| void wait_for_all( task& parent, task* child ) { |
| static_cast<custom_scheduler*>(governor::local_scheduler())->scheduler_type::local_wait_for_all( parent, child ); |
| } |
| |
| //! Construct a custom_scheduler |
| custom_scheduler( arena* a, size_t index ) : generic_scheduler(a, index) {} |
| |
| //! Decrements ref_count of a predecessor. |
| /** If it achieves 0, the predecessor is scheduled for execution. |
| When changing, remember that this is a hot path function. */ |
| void tally_completion_of_predecessor( task& s, task*& bypass_slot ) { |
| task_prefix& p = s.prefix(); |
| if( SchedulerTraits::itt_possible ) |
| ITT_NOTIFY(sync_releasing, &p.ref_count); |
| if( SchedulerTraits::has_slow_atomic && p.ref_count==1 ) { |
| p.ref_count=0; |
| } else { |
| if( __TBB_FetchAndDecrementWrelease(&p.ref_count) > 1 ) // more references exist |
| return; |
| } |
| __TBB_ASSERT(p.ref_count==0, "completion of task caused predecessor's reference count to underflow"); |
| if( SchedulerTraits::itt_possible ) |
| ITT_NOTIFY(sync_acquired, &p.ref_count); |
| #if TBB_USE_ASSERT |
| p.extra_state &= ~es_ref_count_active; |
| #endif /* TBB_USE_ASSERT */ |
| |
| if( bypass_slot==NULL ) |
| bypass_slot = &s; |
| else |
| local_spawn( s, s.prefix().next ); |
| } |
| |
| public: |
| static generic_scheduler* allocate_scheduler( arena* a, size_t index ) { |
| #if !__TBB_ARENA_PER_MASTER |
| __TBB_ASSERT( a, "missing arena" ); |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| scheduler_type* s = (scheduler_type*)NFS_Allocate(sizeof(scheduler_type),1,NULL); |
| new( s ) scheduler_type( a, index ); |
| __TBB_ASSERT( s->assert_okay(), NULL ); |
| ITT_SYNC_CREATE(s, SyncType_Scheduler, SyncObj_TaskPoolSpinning); |
| return s; |
| } |
| |
| //! Try getting a task from the mailbox or stealing from another scheduler. |
| /** Returns the stolen task or NULL if all attempts fail. */ |
| /* override */ task* receive_or_steal_task( reference_count&, bool ); |
| |
| }; // class custom_scheduler<> |
| |
| //------------------------------------------------------------------------ |
| // custom_scheduler methods |
| //------------------------------------------------------------------------ |
| |
| template<typename SchedulerTraits> |
| task* custom_scheduler<SchedulerTraits>::receive_or_steal_task( reference_count& completion_ref_count, |
| bool return_if_no_work ) { |
| task* t = NULL; |
| inbox.set_is_idle( true ); |
| // The state "failure_count==-1" is used only when itt_possible is true, |
| // and denotes that a sync_prepare has not yet been issued. |
| for( int failure_count = -static_cast<int>(SchedulerTraits::itt_possible);; ++failure_count) { |
| if( completion_ref_count==1 ) { |
| if( SchedulerTraits::itt_possible ) { |
| if( failure_count!=-1 ) { |
| ITT_NOTIFY(sync_prepare, &completion_ref_count); |
| // Notify Intel(R) Thread Profiler that thread has stopped spinning. |
| ITT_NOTIFY(sync_acquired, this); |
| } |
| ITT_NOTIFY(sync_acquired, &completion_ref_count); |
| } |
| inbox.set_is_idle( false ); |
| return NULL; |
| } |
| #if __TBB_ARENA_PER_MASTER |
| size_t n = my_arena->my_limit; |
| __TBB_ASSERT( arena_index < n, NULL ); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| size_t n = my_arena->prefix().limit; |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| if( n>1 ) { |
| if( my_affinity_id && (t=get_mailbox_task()) ) { |
| GATHER_STATISTIC( ++mail_received_count ); |
| } |
| #if __TBB_ARENA_PER_MASTER |
| // Check if there are tasks in starvation-resistant stream. |
| // Only allowed for workers with empty stack, which is identified by return_if_no_work. |
| else if ( return_if_no_work && (t=dequeue_task()) ) { |
| // just proceed with the obtained task |
| } |
| // Check if the resource manager requires our arena to relinquish some threads |
| else if ( return_if_no_work && (my_arena->my_num_workers_allotted < my_arena->num_workers_active()) ) |
| return NULL; |
| #endif /* __TBB_ARENA_PER_MASTER */ |
| else { |
| // Try to steal a task from a random victim. |
| if ( !can_steal() ) |
| goto fail; |
| size_t k = random.get() % (n-1); |
| arena_slot* victim = &my_arena->slot[k]; |
| // The following condition excludes the master that might have |
| // already taken our previous place in the arena from the list . |
| // of potential victims. But since such a situation can take |
| // place only in case of significant oversubscription, keeping |
| // the checks simple seems to be preferable to complicating the code. |
| if( k >= arena_index ) |
| ++victim; // Adjusts random distribution to exclude self |
| t = steal_task( *victim ); |
| if( !t ) goto fail; |
| if( is_proxy(*t) ) { |
| t = strip_proxy((task_proxy*)t); |
| if( !t ) goto fail; |
| GATHER_STATISTIC( ++proxy_steal_count ); |
| } |
| t->prefix().extra_state |= es_task_is_stolen; |
| if( is_version_3_task(*t) ) { |
| innermost_running_task = t; |
| t->note_affinity( my_affinity_id ); |
| } |
| GATHER_STATISTIC( ++steal_count ); |
| } |
| __TBB_ASSERT(t,NULL); |
| #if __TBB_SCHEDULER_OBSERVER |
| // No memory fence required for read of global_last_observer_proxy, because prior fence on steal/mailbox suffices. |
| if( local_last_observer_proxy!=global_last_observer_proxy ) { |
| notify_entry_observers(); |
| } |
| #endif /* __TBB_SCHEDULER_OBSERVER */ |
| if( SchedulerTraits::itt_possible ) { |
| if( failure_count!=-1 ) { |
| // FIXME - might be victim, or might be selected from a mailbox |
| // Notify Intel(R) Thread Profiler that thread has stopped spinning. |
| ITT_NOTIFY(sync_acquired, this); |
| } |
| } |
| inbox.set_is_idle( false ); |
| break; // jumps to: return t; |
| } |
| fail: |
| if( SchedulerTraits::itt_possible && failure_count==-1 ) { |
| // The first attempt to steal work failed, so notify Intel(R) Thread Profiler that |
| // the thread has started spinning. Ideally, we would do this notification |
| // *before* the first failed attempt to steal, but at that point we do not |
| // know that the steal will fail. |
| ITT_NOTIFY(sync_prepare, this); |
| failure_count = 0; |
| } |
| // Pause, even if we are going to yield, because the yield might return immediately. |
| __TBB_Pause(PauseTime); |
| int yield_threshold = 2*int(n); |
| if( failure_count>=yield_threshold ) { |
| __TBB_Yield(); |
| if( failure_count>=yield_threshold+100 ) { |
| // When a worker thread has nothing to do, return it to RML. |
| // For purposes of affinity support, the thread is considered idle while in RML. |
| if( return_if_no_work && my_arena->is_out_of_work() ) |
| return NULL; |
| failure_count = yield_threshold; |
| } |
| } |
| } |
| return t; |
| } |
| |
| template<typename SchedulerTraits> |
| void custom_scheduler<SchedulerTraits>::local_wait_for_all( task& parent, task* child ) { |
| __TBB_ASSERT( governor::is_set(this), NULL ); |
| if( child ) { |
| child->prefix().owner = this; |
| } |
| __TBB_ASSERT( parent.ref_count() >= (child && child->parent() == &parent ? 2 : 1), "ref_count is too small" ); |
| __TBB_ASSERT( assert_okay(), NULL ); |
| // Using parent's refcount in sync_prepare (in the stealing loop below) is |
| // a workaround for TP. We need to name it here to display correctly in Ampl. |
| if( SchedulerTraits::itt_possible ) |
| ITT_SYNC_CREATE(&parent.prefix().ref_count, SyncType_Scheduler, SyncObj_TaskStealingLoop); |
| #if __TBB_TASK_GROUP_CONTEXT |
| __TBB_ASSERT( parent.prefix().context || (is_worker() && &parent == dummy_task), "parent task does not have context" ); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| task* t = child; |
| // Constant all_local_work_done is an unreacheable refcount value that prevents |
| // early quitting the dispatch loop. It is defined to be in the middle of the range |
| // of negative values representable by the reference_count type. |
| static const reference_count |
| // For normal dispatch loops |
| parents_work_done = 1, |
| // For termination dispatch loops in masters |
| all_local_work_done = (reference_count)3 << (sizeof(reference_count) * 8 - 2); |
| reference_count quit_point; |
| if( innermost_running_task == dummy_task ) { |
| // We are in the outermost task dispatch loop of a master thread, |
| __TBB_ASSERT( !is_worker(), NULL ); |
| quit_point = &parent == dummy_task ? all_local_work_done : parents_work_done; |
| } else { |
| quit_point = parents_work_done; |
| } |
| task* old_innermost_running_task = innermost_running_task; |
| #if __TBB_TASK_GROUP_CONTEXT && TBB_USE_EXCEPTIONS |
| exception_was_caught: |
| try { |
| #endif /* __TBB_TASK_GROUP_CONTEXT && TBB_USE_EXCEPTIONS */ |
| // Outer loop steals tasks when necessary. |
| for(;;) { |
| // Middle loop evaluates tasks that are pulled off "array". |
| do { |
| // Inner loop evaluates tasks that are handed directly to us by other tasks. |
| while(t) { |
| __TBB_ASSERT( inbox.assert_is_idle(false), NULL ); |
| #if TBB_USE_ASSERT |
| __TBB_ASSERT(!is_proxy(*t),"unexpected proxy"); |
| __TBB_ASSERT( t->prefix().owner==this, NULL ); |
| #if __TBB_TASK_GROUP_CONTEXT |
| if ( !t->prefix().context->my_cancellation_requested ) |
| #endif |
| __TBB_ASSERT( 1L<<t->state() & (1L<<task::allocated|1L<<task::ready|1L<<task::reexecute), NULL ); |
| __TBB_ASSERT(assert_okay(),NULL); |
| #endif /* TBB_USE_ASSERT */ |
| task* t_next = NULL; |
| innermost_running_task = t; |
| t->prefix().state = task::executing; |
| #if __TBB_TASK_GROUP_CONTEXT |
| if ( !t->prefix().context->my_cancellation_requested ) |
| #endif |
| { |
| GATHER_STATISTIC( ++execute_count ); |
| #if __TBB_TASK_GROUP_CONTEXT |
| if( SchedulerTraits::itt_possible ) |
| ITT_STACK(callee_enter, t->prefix().context->itt_caller); |
| #endif |
| t_next = t->execute(); |
| #if __TBB_TASK_GROUP_CONTEXT |
| if( SchedulerTraits::itt_possible ) |
| ITT_STACK(callee_leave, t->prefix().context->itt_caller); |
| #endif |
| if (t_next) { |
| __TBB_ASSERT( t_next->state()==task::allocated, |
| "if task::execute() returns task, it must be marked as allocated" ); |
| t_next->prefix().extra_state &= ~es_task_is_stolen; |
| #if STATISTICS |
| affinity_id next_affinity=t_next->prefix().affinity; |
| if (next_affinity != 0 && next_affinity != my_affinity_id) |
| GATHER_STATISTIC( ++proxy_bypass_count ); |
| #endif |
| } |
| } |
| __TBB_ASSERT(assert_okay(),NULL); |
| switch( task::state_type(t->prefix().state) ) { |
| case task::executing: { |
| task* s = t->parent(); |
| __TBB_ASSERT( innermost_running_task==t, NULL ); |
| __TBB_ASSERT( t->prefix().ref_count==0, "Task still has children after it has been executed" ); |
| t->~task(); |
| if( s ) |
| tally_completion_of_predecessor(*s, t_next); |
| free_task<no_hint>( *t ); |
| __TBB_ASSERT(assert_okay(),NULL); |
| break; |
| } |
| |
| case task::recycle: // set by recycle_as_safe_continuation() |
| t->prefix().state = task::allocated; |
| __TBB_ASSERT( t_next != t, "a task returned from method execute() can not be recycled in another way" ); |
| t->prefix().extra_state &= ~es_task_is_stolen; |
| // for safe continuation, need atomically decrement ref_count; |
| tally_completion_of_predecessor(*t, t_next); |
| __TBB_ASSERT(assert_okay(),NULL); |
| break; |
| |
| case task::reexecute: // set by recycle_to_reexecute() |
| __TBB_ASSERT( t_next, "reexecution requires that method execute() return another task" ); |
| __TBB_ASSERT( t_next != t, "a task returned from method execute() can not be recycled in another way" ); |
| t->prefix().state = task::allocated; |
| t->prefix().extra_state &= ~es_task_is_stolen; |
| local_spawn( *t, t->prefix().next ); |
| __TBB_ASSERT(assert_okay(),NULL); |
| break; |
| case task::allocated: |
| t->prefix().extra_state &= ~es_task_is_stolen; |
| break; |
| #if TBB_USE_ASSERT |
| case task::ready: |
| __TBB_ASSERT( false, "task is in READY state upon return from method execute()" ); |
| break; |
| default: |
| __TBB_ASSERT( false, "illegal state" ); |
| #else |
| default: // just to shut up some compilation warnings |
| break; |
| #endif /* TBB_USE_ASSERT */ |
| } |
| |
| if( t_next ) { |
| // The store here has a subtle secondary effect - it fetches *t_next into cache. |
| t_next->prefix().owner = this; |
| } |
| t = t_next; |
| } // end of scheduler bypass loop |
| __TBB_ASSERT(assert_okay(),NULL); |
| |
| if ( parent.prefix().ref_count == quit_point ) |
| break; |
| t = get_task(); |
| __TBB_ASSERT(!t || !is_proxy(*t),"unexpected proxy"); |
| #if TBB_USE_ASSERT |
| __TBB_ASSERT(assert_okay(),NULL); |
| if(t) { |
| AssertOkay(*t); |
| __TBB_ASSERT( t->prefix().owner==this, "thread got task that it does not own" ); |
| } |
| #endif /* TBB_USE_ASSERT */ |
| } while( t ); // end of local task array processing loop |
| |
| if ( quit_point == all_local_work_done ) { |
| __TBB_ASSERT( my_arena_slot == &dummy_slot && my_arena_slot->head == 0 && my_arena_slot->tail == 0, NULL ); |
| innermost_running_task = old_innermost_running_task; |
| return; |
| } |
| #if __TBB_ARENA_PER_MASTER |
| __TBB_ASSERT( my_arena->my_max_num_workers > 0 || parent.prefix().ref_count == 1, "deadlock detected" ); |
| #else /* !__TBB_ARENA_PER_MASTER */ |
| __TBB_ASSERT( my_arena->prefix().number_of_workers>0||parent.prefix().ref_count==1, "deadlock detected" ); |
| #endif /* !__TBB_ARENA_PER_MASTER */ |
| // old_innermost_running_task is NULL *iff* a worker thread is in its "inborn" dispath loop |
| // (i.e. its execution stack is empty), and it should return from there if no work is available. |
| t = receive_or_steal_task( parent.prefix().ref_count, !old_innermost_running_task ); |
| if (!t) { |
| if( parent.prefix().ref_count==1 ) goto done; |
| __TBB_ASSERT( is_worker() && !old_innermost_running_task, "a thread exits dispatch loop prematurely" ); |
| innermost_running_task = NULL; |
| return; |
| } |
| __TBB_ASSERT(t,NULL); |
| __TBB_ASSERT(!is_proxy(*t),"unexpected proxy"); |
| t->prefix().owner = this; |
| } // end of stealing loop |
| #if __TBB_TASK_GROUP_CONTEXT && TBB_USE_EXCEPTIONS |
| } TbbCatchAll( t->prefix().context ); |
| |
| if( task::state_type(t->prefix().state) == task::recycle ) { // state set by recycle_as_safe_continuation() |
| t->prefix().state = task::allocated; |
| // for safe continuation, need to atomically decrement ref_count; |
| if( SchedulerTraits::itt_possible ) |
| ITT_NOTIFY(sync_releasing, &t->prefix().ref_count); |
| if( __TBB_FetchAndDecrementWrelease(&t->prefix().ref_count)==1 ) { |
| if( SchedulerTraits::itt_possible ) |
| ITT_NOTIFY(sync_acquired, &t->prefix().ref_count); |
| }else{ |
| t = NULL; |
| } |
| } |
| goto exception_was_caught; |
| #endif /* __TBB_TASK_GROUP_CONTEXT && TBB_USE_EXCEPTIONS */ |
| done: |
| if ( !ConcurrentWaitsEnabled(parent) ) |
| parent.prefix().ref_count = 0; |
| #if TBB_USE_ASSERT |
| parent.prefix().extra_state &= ~es_ref_count_active; |
| #endif /* TBB_USE_ASSERT */ |
| innermost_running_task = old_innermost_running_task; |
| #if __TBB_TASK_GROUP_CONTEXT |
| __TBB_ASSERT(parent.prefix().context && dummy_task->prefix().context, NULL); |
| task_group_context* parent_ctx = parent.prefix().context; |
| if ( parent_ctx->my_cancellation_requested ) { |
| task_group_context::exception_container_type *pe = parent_ctx->my_exception; |
| if ( innermost_running_task == dummy_task && parent_ctx == dummy_task->prefix().context ) { |
| // We are in the outermost task dispatch loop of a master thread, and |
| // the whole task tree has been collapsed. So we may clear cancellation data. |
| parent_ctx->my_cancellation_requested = 0; |
| __TBB_ASSERT(dummy_task->prefix().context == parent_ctx || !CancellationInfoPresent(*dummy_task), |
| "Unexpected exception or cancellation data in the dummy task"); |
| // If possible, add assertion that master's dummy task context does not have children |
| } |
| if ( pe ) |
| pe->throw_self(); |
| } |
| __TBB_ASSERT(!is_worker() || !CancellationInfoPresent(*dummy_task), |
| "Worker's dummy task context modified"); |
| __TBB_ASSERT(innermost_running_task != dummy_task || !CancellationInfoPresent(*dummy_task), |
| "Unexpected exception or cancellation data in the master's dummy task"); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| __TBB_ASSERT( assert_okay(), NULL ); |
| } |
| |
| } // namespace internal |
| } // namespace tbb |
| |
| #endif /* _TBB_custom_scheduler_H */ |
