tbb::internal::custom_scheduler< SchedulerTraits > Class Template Reference

A scheduler with a customized evaluation loop. More...

#include <custom_scheduler.h>

Inheritance diagram for tbb::internal::custom_scheduler< SchedulerTraits >:

Collaboration diagram for tbb::internal::custom_scheduler< SchedulerTraits >:

Public Member Functions
task *	receive_or_steal_task (__TBB_ISOLATION_ARG(__TBB_atomic reference_count &completion_ref_count, isolation_tag isolation)) __TBB_override
	Try getting a task from the mailbox or stealing from another scheduler. More...

Static Public Member Functions
static generic_scheduler *	allocate_scheduler (market &m, bool genuine)

Private Types
typedef custom_scheduler< SchedulerTraits >	scheduler_type

Private Member Functions
	custom_scheduler (market &m, bool genuine)
void	local_wait_for_all (task &parent, task *child) __TBB_override
	Scheduler loop that dispatches tasks. More...
void	wait_for_all (task &parent, task *child) __TBB_override
	Entry point from client code to the scheduler loop that dispatches tasks. More...
void	tally_completion_of_predecessor (task &s, __TBB_ISOLATION_ARG(task *&bypass_slot, isolation_tag isolation))
	Decrements ref_count of a predecessor. More...
bool	process_bypass_loop (context_guard_helper< SchedulerTraits::itt_possible > &context_guard, __TBB_ISOLATION_ARG(task *t, isolation_tag isolation))
	Implements the bypass loop of the dispatch loop (local_wait_for_all). More...
Private Member Functions inherited from tbb::internal::generic_scheduler
bool	is_task_pool_published () const
bool	is_local_task_pool_quiescent () const
bool	is_quiescent_local_task_pool_empty () const
bool	is_quiescent_local_task_pool_reset () const
void	attach_mailbox (affinity_id id)
void	init_stack_info ()
	Sets up the data necessary for the stealing limiting heuristics. More...
bool	can_steal ()
	Returns true if stealing is allowed. More...
void	publish_task_pool ()
	Used by workers to enter the task pool. More...
void	leave_task_pool ()
	Leave the task pool. More...
void	reset_task_pool_and_leave ()
	Resets head and tail indices to 0, and leaves task pool. More...
task **	lock_task_pool (arena_slot *victim_arena_slot) const
	Locks victim's task pool, and returns pointer to it. The pointer can be NULL. More...
void	unlock_task_pool (arena_slot *victim_arena_slot, task **victim_task_pool) const
	Unlocks victim's task pool. More...
void	acquire_task_pool () const
	Locks the local task pool. More...
void	release_task_pool () const
	Unlocks the local task pool. More...
task *	prepare_for_spawning (task *t)
	Checks if t is affinitized to another thread, and if so, bundles it as proxy. More...
void	commit_spawned_tasks (size_t new_tail)
	Makes newly spawned tasks visible to thieves. More...
void	commit_relocated_tasks (size_t new_tail)
	Makes relocated tasks visible to thieves and releases the local task pool. More...
task *	get_task (__TBB_ISOLATION_EXPR(isolation_tag isolation))
	Get a task from the local pool. More...
task *	get_task (size_t T)
	Get a task from the local pool at specified location T. More...
task *	get_mailbox_task (__TBB_ISOLATION_EXPR(isolation_tag isolation))
	Attempt to get a task from the mailbox. More...
task *	steal_task (__TBB_ISOLATION_EXPR(isolation_tag isolation))
	Attempts to steal a task from a randomly chosen thread/scheduler. More...
task *	steal_task_from (__TBB_ISOLATION_ARG(arena_slot &victim_arena_slot, isolation_tag isolation))
	Steal task from another scheduler's ready pool. More...
size_t	prepare_task_pool (size_t n)
	Makes sure that the task pool can accommodate at least n more elements. More...
bool	cleanup_master (bool blocking_terminate)
	Perform necessary cleanup when a master thread stops using TBB. More...
void	assert_task_pool_valid () const
void	attach_arena (arena *, size_t index, bool is_master)
void	nested_arena_entry (arena *, size_t)
void	nested_arena_exit ()
void	wait_until_empty ()
void	spawn (task &first, task *&next) __TBB_override
	For internal use only. More...
void	spawn_root_and_wait (task &first, task *&next) __TBB_override
	For internal use only. More...
void	enqueue (task &, void *reserved) __TBB_override
	For internal use only. More...
void	local_spawn (task *first, task *&next)
void	local_spawn_root_and_wait (task *first, task *&next)
void	destroy ()
	Destroy and deallocate this scheduler object. More...
void	cleanup_scheduler ()
	Cleans up this scheduler (the scheduler might be destroyed). More...
task &	allocate_task (size_t number_of_bytes, __TBB_CONTEXT_ARG(task *parent, task_group_context *context))
	Allocate task object, either from the heap or a free list. More...
template<free_task_hint h>
void	free_task (task &t)
	Put task on free list. More...
void	deallocate_task (task &t)
	Return task object to the memory allocator. More...
bool	is_worker () const
	True if running on a worker thread, false otherwise. More...
bool	outermost_level () const
	True if the scheduler is on the outermost dispatch level. More...
bool	master_outermost_level () const
	True if the scheduler is on the outermost dispatch level in a master thread. More...
bool	worker_outermost_level () const
	True if the scheduler is on the outermost dispatch level in a worker thread. More...
unsigned	max_threads_in_arena ()
	Returns the concurrency limit of the current arena. More...
void	free_nonlocal_small_task (task &t)
	Free a small task t that that was allocated by a different scheduler. More...
	generic_scheduler (market &, bool)
Private Member Functions inherited from tbb::internal::scheduler
virtual	~scheduler ()=0
	Pure virtual destructor;. More...

Additional Inherited Members
Static Private Member Functions inherited from tbb::internal::generic_scheduler
static bool	is_version_3_task (task &t)
static bool	is_proxy (const task &t)
	True if t is a task_proxy. More...
static generic_scheduler *	create_master (arena *a)
	Initialize a scheduler for a master thread. More...
static generic_scheduler *	create_worker (market &m, size_t index, bool geniune)
	Initialize a scheduler for a worker thread. More...
static void	cleanup_worker (void *arg, bool worker)
	Perform necessary cleanup when a worker thread finishes. More...
static task *	plugged_return_list ()
	Special value used to mark my_return_list as not taking any more entries. More...
Private Attributes inherited from tbb::internal::generic_scheduler
uintptr_t	my_stealing_threshold
	Position in the call stack specifying its maximal filling when stealing is still allowed. More...
market *	my_market
	The market I am in. More...
FastRandom	my_random
	Random number generator used for picking a random victim from which to steal. More...
task *	my_free_list
	Free list of small tasks that can be reused. More...
task *	my_dummy_task
	Fake root task created by slave threads. More...
long	my_ref_count
	Reference count for scheduler. More...
bool	my_auto_initialized
	True if *this was created by automatic TBB initialization. More...
__TBB_atomic intptr_t	my_small_task_count
	Number of small tasks that have been allocated by this scheduler. More...
task *	my_return_list
	List of small tasks that have been returned to this scheduler by other schedulers. More...
Private Attributes inherited from tbb::internal::intrusive_list_node
intrusive_list_node *	my_prev_node
intrusive_list_node *	my_next_node
Private Attributes inherited from tbb::internal::scheduler_state
size_t	my_arena_index
	Index of the arena slot the scheduler occupies now, or occupied last time. More...
arena_slot *	my_arena_slot
	Pointer to the slot in the arena we own at the moment. More...
arena *	my_arena
	The arena that I own (if master) or am servicing at the moment (if worker) More...
task *	my_innermost_running_task
	Innermost task whose task::execute() is running. A dummy task on the outermost level. More...
mail_inbox	my_inbox
affinity_id	my_affinity_id
	The mailbox id assigned to this scheduler. More...
scheduler_properties	my_properties
Static Private Attributes inherited from tbb::internal::generic_scheduler
static const size_t	quick_task_size = 256-task_prefix_reservation_size
	If sizeof(task) is <=quick_task_size, it is handled on a free list instead of malloc'd. More...
static const size_t	null_arena_index = ~size_t(0)
static const size_t	min_task_pool_size = 64

Detailed Description

template<typename SchedulerTraits>
class tbb::internal::custom_scheduler< SchedulerTraits >

A scheduler with a customized evaluation loop.

The customization can use SchedulerTraits to make decisions without needing a run-time check.

Definition at line 52 of file custom_scheduler.h.

Member Typedef Documentation

scheduler_type

template<typename SchedulerTraits >

typedef custom_scheduler<SchedulerTraits> tbb::internal::custom_scheduler< SchedulerTraits >::scheduler_type

private

Definition at line 53 of file custom_scheduler.h.

Constructor & Destructor Documentation

custom_scheduler()

template<typename SchedulerTraits >

tbb::internal::custom_scheduler< SchedulerTraits >::custom_scheduler ( market &  m, bool  genuine  )

inlineprivate

Definition at line 55 of file custom_scheduler.h.

: generic_scheduler(m, genuine) {}

Member Function Documentation

allocate_scheduler()

template<typename SchedulerTraits >

static generic_scheduler* tbb::internal::custom_scheduler< SchedulerTraits >::allocate_scheduler ( market &  m, bool  genuine  )

inlinestatic

Definition at line 137 of file custom_scheduler.h.

                                                                            {
        void* p = NFS_Allocate(1, sizeof(scheduler_type), NULL);
        std::memset(p, 0, sizeof(scheduler_type));
        scheduler_type* s = new( p ) scheduler_type( m, genuine );
        s->assert_task_pool_valid();
        ITT_SYNC_CREATE(s, SyncType_Scheduler, SyncObj_TaskPoolSpinning);
        return s;
    }

References ITT_SYNC_CREATE, tbb::internal::NFS_Allocate(), p, and s.

Here is the call graph for this function:

local_wait_for_all()

template<typename SchedulerTraits >

void tbb::internal::custom_scheduler< SchedulerTraits >::local_wait_for_all ( task &  parent, task *  child  )

privatevirtual

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.

Implements tbb::internal::generic_scheduler.

Definition at line 557 of file custom_scheduler.h.

                                                                                      {
    __TBB_ASSERT( governor::is_set(this), NULL );
    __TBB_ASSERT( parent.ref_count() >= (child && child->parent() == &parent ? 2 : 1), "ref_count is too small" );
    __TBB_ASSERT( my_innermost_running_task, NULL );
#if __TBB_TASK_GROUP_CONTEXT
    __TBB_ASSERT( parent.prefix().context, "parent task does not have context" );
#endif /* __TBB_TASK_GROUP_CONTEXT */
    assert_task_pool_valid();
    // 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);

    // TODO: consider extending the "context" guard to a "dispatch loop" guard to additionally
    // guard old_innermost_running_task and old_properties states.
    context_guard_helper</*report_tasks=*/SchedulerTraits::itt_possible> context_guard;
    task* old_innermost_running_task = my_innermost_running_task;
    scheduler_properties old_properties = my_properties;

    task* t = child;
    bool cleanup = !is_worker() && &parent==my_dummy_task;
    // Remove outermost property to indicate nested level.
    __TBB_ASSERT(my_properties.outermost || my_innermost_running_task!=my_dummy_task, "The outermost property should be set out of a dispatch loop");
    my_properties.outermost &= my_innermost_running_task==my_dummy_task;
#if __TBB_PREVIEW_CRITICAL_TASKS
    my_properties.has_taken_critical_task |= is_critical(*my_innermost_running_task);
#endif
#if __TBB_TASK_PRIORITY
    __TBB_ASSERT( (uintptr_t)*my_ref_top_priority < (uintptr_t)num_priority_levels, NULL );
    volatile intptr_t *old_ref_top_priority = my_ref_top_priority;
    // When entering nested parallelism level market level counter
    // must be replaced with the one local to this arena.
    volatile uintptr_t *old_ref_reload_epoch = my_ref_reload_epoch;
    if ( !outermost_level() ) {
        // We are in a nested dispatch loop.
        // Market or arena priority must not prevent child tasks from being
        // executed so that dynamic priority changes did not cause deadlock.
        my_ref_top_priority = &parent.prefix().context->my_priority;
        my_ref_reload_epoch = &my_arena->my_reload_epoch;
        if (my_ref_reload_epoch != old_ref_reload_epoch)
            my_local_reload_epoch = *my_ref_reload_epoch - 1;
    }
#endif /* __TBB_TASK_PRIORITY */
#if __TBB_TASK_ISOLATION
    isolation_tag isolation = my_innermost_running_task->prefix().isolation;
    if (t && isolation != no_isolation) {
        __TBB_ASSERT(t->prefix().isolation == no_isolation, NULL);
        // Propagate the isolation to the task executed without spawn.
        t->prefix().isolation = isolation;
    }
#endif /* __TBB_TASK_ISOLATION */
#if __TBB_PREVIEW_RESUMABLE_TASKS
    // The recall flag for the original owner of this scheduler.
    // It is used only on outermost level of currently attached arena slot.
    tbb::atomic<bool> recall_flag;
    recall_flag = false;
    if (outermost_level() && my_wait_task == NULL && my_properties.genuine) {
        __TBB_ASSERT(my_arena_slot->my_scheduler == this, NULL);
        __TBB_ASSERT(my_arena_slot->my_scheduler_is_recalled == NULL, NULL);
        my_arena_slot->my_scheduler_is_recalled = &recall_flag;
        my_current_is_recalled = &recall_flag;
    }
    __TBB_ASSERT(my_arena_slot->my_scheduler_is_recalled != NULL, NULL);
    task* old_wait_task = my_wait_task;
    my_wait_task = &parent;
#endif
#if TBB_USE_EXCEPTIONS
    // Infinite safeguard EH loop
    for (;;) {
    try {
#endif /* TBB_USE_EXCEPTIONS */
    // Outer loop receives tasks from global environment (via mailbox, FIFO queue(s),
    // and by  stealing from other threads' task pools).
    // All exit points from the dispatch loop are located in its immediate scope.
    for(;;) {
        // Middle loop retrieves tasks from the local task pool.
        for(;;) {
            // Inner loop evaluates tasks coming from nesting loops and those returned
            // by just executed tasks (bypassing spawn or enqueue calls).
            if ( !process_bypass_loop( context_guard, __TBB_ISOLATION_ARG(t, isolation) ) ) {
#if __TBB_PREVIEW_RESUMABLE_TASKS
                // Restore the old properties for the coroutine reusage (leave in a valid state)
                my_innermost_running_task = old_innermost_running_task;
                my_properties = old_properties;
                my_wait_task = old_wait_task;
#endif
                return;
            }

            // Check "normal" exit condition when parent's work is done.
            if ( parent.prefix().ref_count == 1 ) {
                __TBB_ASSERT( !cleanup, NULL );
                __TBB_control_consistency_helper(); // on ref_count
                ITT_NOTIFY( sync_acquired, &parent.prefix().ref_count );
                goto done;
            }
#if __TBB_PREVIEW_RESUMABLE_TASKS
            // The thread may be otside of its original scheduler. Check the recall request.
            if ( &recall_flag != my_arena_slot->my_scheduler_is_recalled ) {
                __TBB_ASSERT( my_arena_slot->my_scheduler_is_recalled != NULL, "A broken recall flag" );
                if ( *my_arena_slot->my_scheduler_is_recalled ) {
                    if ( !resume_original_scheduler() ) {
                        // We are requested to finish the current coroutine before the resume.
                        __TBB_ASSERT( !my_properties.genuine && my_properties.outermost,
                            "Only a coroutine on outermost level can be left." );
                        // Restore the old properties for the coroutine reusage (leave in a valid state)
                        my_innermost_running_task = old_innermost_running_task;
                        my_properties = old_properties;
                        my_wait_task = old_wait_task;
                        return;
                    }
                }
            }
#endif
            // Retrieve the task from local task pool.
            __TBB_ASSERT( is_task_pool_published() || is_quiescent_local_task_pool_reset(), NULL );
            t = is_task_pool_published() ? get_task( __TBB_ISOLATION_EXPR( isolation ) ) : NULL;
            assert_task_pool_valid();

            if ( !t ) // No tasks in the local task pool. Go to stealing loop.
                break;
        }; // end of local task pool retrieval loop

#if __TBB_HOARD_NONLOCAL_TASKS
        // before stealing, previously stolen task objects are returned
        for (; my_nonlocal_free_list; my_nonlocal_free_list = t ) {
            t = my_nonlocal_free_list->prefix().next;
            free_nonlocal_small_task( *my_nonlocal_free_list );
        }
#endif
        if ( cleanup ) {
            __TBB_ASSERT( !is_task_pool_published() && is_quiescent_local_task_pool_reset(), NULL );
            __TBB_ASSERT( !worker_outermost_level(), NULL );
            my_innermost_running_task = old_innermost_running_task;
            my_properties = old_properties;
#if __TBB_TASK_PRIORITY
            my_ref_top_priority = old_ref_top_priority;
            if(my_ref_reload_epoch != old_ref_reload_epoch)
                my_local_reload_epoch = *old_ref_reload_epoch-1;
            my_ref_reload_epoch = old_ref_reload_epoch;
#endif /* __TBB_TASK_PRIORITY */
#if __TBB_PREVIEW_RESUMABLE_TASKS
            if (&recall_flag != my_arena_slot->my_scheduler_is_recalled) {
                // The recall point
                __TBB_ASSERT(!recall_flag, NULL);
                tbb::task::suspend(recall_functor(&recall_flag));
                if (my_inbox.is_idle_state(true))
                    my_inbox.set_is_idle(false);
                continue;
            }
            __TBB_ASSERT(&recall_flag == my_arena_slot->my_scheduler_is_recalled, NULL);
            __TBB_ASSERT(!(my_wait_task->prefix().ref_count & internal::abandon_flag), NULL);
            my_wait_task = old_wait_task;
#endif
            return;
        }
        t = receive_or_steal_task( __TBB_ISOLATION_ARG( parent.prefix().ref_count, isolation ) );
        if ( !t ) {
#if __TBB_PREVIEW_RESUMABLE_TASKS
            if ( *my_arena_slot->my_scheduler_is_recalled )
                continue;
            // Done if either original thread enters or we are on the nested level or attached the same arena
            if ( &recall_flag == my_arena_slot->my_scheduler_is_recalled || old_wait_task != NULL )
                goto done;
            // The recall point. Continue dispatch loop because recalled thread may have tasks in it's task pool.
            __TBB_ASSERT(!recall_flag, NULL);
            tbb::task::suspend( recall_functor(&recall_flag) );
            if ( my_inbox.is_idle_state(true) )
                my_inbox.set_is_idle(false);
#else
            // Just exit dispatch loop
            goto done;
#endif
        }
    } // end of infinite stealing loop
#if TBB_USE_EXCEPTIONS
    __TBB_ASSERT( false, "Must never get here" );
    } // end of try-block
    TbbCatchAll( my_innermost_running_task->prefix().context );
    t = my_innermost_running_task;
    // Complete post-processing ...
    if( t->state() == task::recycle
#if __TBB_RECYCLE_TO_ENQUEUE
        // TODO: the enqueue semantics gets lost below, consider reimplementing
        ||  t->state() == task::to_enqueue
#endif
      ) {
        // ... for recycled tasks to atomically decrement ref_count
        t->prefix().state = task::allocated;
        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;
        }
    }
    } // end of infinite EH loop
    __TBB_ASSERT( false, "Must never get here too" );
#endif /* TBB_USE_EXCEPTIONS */
done:
#if __TBB_PREVIEW_RESUMABLE_TASKS
    __TBB_ASSERT(!(parent.prefix().ref_count & internal::abandon_flag), NULL);
    my_wait_task = old_wait_task;
    if (my_wait_task == NULL) {
        __TBB_ASSERT(outermost_level(), "my_wait_task could be NULL only on outermost level");
        if (&recall_flag != my_arena_slot->my_scheduler_is_recalled) {
            // The recall point.
            __TBB_ASSERT(my_properties.genuine, NULL);
            __TBB_ASSERT(!recall_flag, NULL);
            tbb::task::suspend(recall_functor(&recall_flag));
            if (my_inbox.is_idle_state(true))
                my_inbox.set_is_idle(false);
        }
        __TBB_ASSERT(my_arena_slot->my_scheduler == this, NULL);
        my_arena_slot->my_scheduler_is_recalled = NULL;
        my_current_is_recalled = NULL;
    }

#endif /* __TBB_PREVIEW_RESUMABLE_TASKS */
    my_innermost_running_task = old_innermost_running_task;
    my_properties = old_properties;
#if __TBB_TASK_PRIORITY
    my_ref_top_priority = old_ref_top_priority;
    if(my_ref_reload_epoch != old_ref_reload_epoch)
        my_local_reload_epoch = *old_ref_reload_epoch-1;
    my_ref_reload_epoch = old_ref_reload_epoch;
#endif /* __TBB_TASK_PRIORITY */
    if ( !ConcurrentWaitsEnabled(parent) ) {
        if ( parent.prefix().ref_count != 1) {
            // This is a worker that was revoked by the market.
            __TBB_ASSERT( worker_outermost_level(),
                "Worker thread exits nested dispatch loop prematurely" );
            return;
        }
        parent.prefix().ref_count = 0;
    }
#if TBB_USE_ASSERT
    parent.prefix().extra_state &= ~es_ref_count_active;
#endif /* TBB_USE_ASSERT */
#if __TBB_TASK_GROUP_CONTEXT
    __TBB_ASSERT(parent.prefix().context && default_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 ( master_outermost_level() && parent_ctx == default_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;
            // TODO: Add assertion that master's dummy task context does not have children
            parent_ctx->my_state &= ~(uintptr_t)task_group_context::may_have_children;
        }
        if ( pe ) {
            // On Windows, FPU control settings changed in the helper destructor are not visible
            // outside a catch block. So restore the default settings manually before rethrowing
            // the exception.
            context_guard.restore_default();
            TbbRethrowException( pe );
        }
    }
    __TBB_ASSERT(!is_worker() || !CancellationInfoPresent(*my_dummy_task),
        "Worker's dummy task context modified");
    __TBB_ASSERT(!master_outermost_level() || !CancellationInfoPresent(*my_dummy_task),
        "Unexpected exception or cancellation data in the master's dummy task");
#endif /* __TBB_TASK_GROUP_CONTEXT */
    assert_task_pool_valid();
}

References __TBB_ASSERT, __TBB_control_consistency_helper, __TBB_FetchAndDecrementWrelease, __TBB_ISOLATION_ARG, __TBB_ISOLATION_EXPR, tbb::task::allocated, tbb::internal::ConcurrentWaitsEnabled(), tbb::internal::es_ref_count_active, tbb::internal::is_critical(), tbb::internal::governor::is_set(), tbb::internal::task_prefix::isolation, ITT_NOTIFY, ITT_SYNC_CREATE, tbb::task_group_context::may_have_children, tbb::internal::task_prefix::next, tbb::internal::no_isolation, tbb::internal::num_priority_levels, parent, tbb::task::parent(), tbb::task::prefix(), tbb::task::recycle, tbb::internal::task_prefix::state, and sync_releasing.

Here is the call graph for this function:

process_bypass_loop()

template<typename SchedulerTraits >

bool tbb::internal::custom_scheduler< SchedulerTraits >::process_bypass_loop ( context_guard_helper< SchedulerTraits::itt_possible > &  context_guard, __TBB_ISOLATION_ARG(task *t, isolation_tag isolation)    )

private

Implements the bypass loop of the dispatch loop (local_wait_for_all).

Definition at line 389 of file custom_scheduler.h.

{
    while ( t ) {
        __TBB_ASSERT( my_inbox.is_idle_state(false), NULL );
        __TBB_ASSERT(!is_proxy(*t),"unexpected proxy");
        __TBB_ASSERT( t->prefix().owner, NULL );
#if __TBB_TASK_ISOLATION
        __TBB_ASSERT_EX( isolation == no_isolation || isolation == t->prefix().isolation,
            "A task from another isolated region is going to be executed" );
#endif /* __TBB_TASK_ISOLATION */
        assert_task_valid(t);
#if __TBB_TASK_GROUP_CONTEXT && TBB_USE_ASSERT
        assert_context_valid(t->prefix().context);
        if ( !t->prefix().context->my_cancellation_requested )
#endif
        // TODO: make the assert stronger by prohibiting allocated state.
        __TBB_ASSERT( 1L<<t->state() & (1L<<task::allocated|1L<<task::ready|1L<<task::reexecute), NULL );
        assert_task_pool_valid();
#if __TBB_PREVIEW_CRITICAL_TASKS
        // TODO: check performance and optimize if needed for added conditions on the
        // hot-path.
        if( !internal::is_critical(*t) && !t->is_enqueued_task() ) {
            if( task* critical_task = get_critical_task( __TBB_ISOLATION_EXPR(isolation) ) ) {
                __TBB_ASSERT( internal::is_critical(*critical_task),
                              "Received task must be critical one" );
                ITT_NOTIFY(sync_acquired, &my_arena->my_critical_task_stream);
                t->prefix().state = task::allocated;
                my_innermost_running_task = t; // required during spawn to propagate isolation
                local_spawn(t, t->prefix().next);
                t = critical_task;
            } else {
#endif /* __TBB_PREVIEW_CRITICAL_TASKS */
#if __TBB_TASK_PRIORITY
                intptr_t p = priority(*t);
                if ( p != *my_ref_top_priority
                     && !t->is_enqueued_task() ) {
                    assert_priority_valid(p);
                    if ( p != my_arena->my_top_priority ) {
                        my_market->update_arena_priority( *my_arena, p );
                    }
                    if ( p < effective_reference_priority() ) {
                        if ( !my_offloaded_tasks ) {
                            my_offloaded_task_list_tail_link = &t->prefix().next_offloaded;
                            // Erase possible reference to the owner scheduler
                            // (next_offloaded is a union member)
                            *my_offloaded_task_list_tail_link = NULL;
                        }
                        offload_task( *t, p );
                        t = NULL;
                        if ( is_task_pool_published() ) {
                            t = winnow_task_pool( __TBB_ISOLATION_EXPR( isolation ) );
                            if ( t )
                                continue;
                        } else {
                            // Mark arena as full to unlock arena priority level adjustment
                            // by arena::is_out_of_work(), and ensure worker's presence.
                            my_arena->advertise_new_work<arena::wakeup>();
                        }
                        break; /* exit bypass loop */
                    }
                }
#endif /* __TBB_TASK_PRIORITY */
#if __TBB_PREVIEW_CRITICAL_TASKS
            }
        } // if is not critical
#endif
        task* t_next = NULL;
        my_innermost_running_task = t;
        t->prefix().owner = this;
        t->prefix().state = task::executing;
#if __TBB_TASK_GROUP_CONTEXT
        context_guard.set_ctx( t->prefix().context );
        if ( !t->prefix().context->my_cancellation_requested )
#endif
        {
            GATHER_STATISTIC( ++my_counters.tasks_executed );
            GATHER_STATISTIC( my_counters.avg_arena_concurrency += my_arena->num_workers_active() );
            GATHER_STATISTIC( my_counters.avg_assigned_workers += my_arena->my_num_workers_allotted );
#if __TBB_TASK_PRIORITY
            GATHER_STATISTIC( my_counters.avg_arena_prio += p );
            GATHER_STATISTIC( my_counters.avg_market_prio += my_market->my_global_top_priority );
#endif /* __TBB_TASK_PRIORITY */
            ITT_STACK(SchedulerTraits::itt_possible, callee_enter, t->prefix().context->itt_caller);
            t_next = t->execute();
            ITT_STACK(SchedulerTraits::itt_possible, callee_leave, t->prefix().context->itt_caller);
            if (t_next) {
                assert_task_valid(t_next);
                __TBB_ASSERT( t_next->state()==task::allocated,
                              "if task::execute() returns task, it must be marked as allocated" );
                reset_extra_state(t_next);
                __TBB_ISOLATION_EXPR( t_next->prefix().isolation = t->prefix().isolation );
#if TBB_USE_ASSERT
                affinity_id next_affinity=t_next->prefix().affinity;
                if (next_affinity != 0 && next_affinity != my_affinity_id)
                    GATHER_STATISTIC( ++my_counters.affinity_ignored );
#endif
            } // if there is bypassed task
        }
        assert_task_pool_valid();
        switch( t->state() ) {
            case task::executing: {
                task* s = t->parent();
                __TBB_ASSERT( my_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, __TBB_ISOLATION_ARG( t_next, t->prefix().isolation ) );
                free_task<no_hint>( *t );
                poison_pointer( my_innermost_running_task );
                assert_task_pool_valid();
                break;
            }

            case task::recycle: // set by recycle_as_safe_continuation()
                t->prefix().state = task::allocated;
#if __TBB_RECYCLE_TO_ENQUEUE
                __TBB_fallthrough;
            case task::to_enqueue: // set by recycle_to_enqueue()
#endif
                __TBB_ASSERT( t_next != t, "a task returned from method execute() can not be recycled in another way" );
                reset_extra_state(t);
                // for safe continuation, need atomically decrement ref_count;
                tally_completion_of_predecessor(*t, __TBB_ISOLATION_ARG( t_next, t->prefix().isolation ) );
                assert_task_pool_valid();
                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;
                reset_extra_state(t);
                local_spawn( t, t->prefix().next );
                assert_task_pool_valid();
                break;
            case task::allocated:
                reset_extra_state(t);
                break;
#if __TBB_PREVIEW_RESUMABLE_TASKS
            case task::to_resume:
                __TBB_ASSERT(my_innermost_running_task == t, NULL);
                __TBB_ASSERT(t->prefix().ref_count == 0, "Task still has children after it has been executed");
                t->~task();
                free_task<no_hint>(*t);
                __TBB_ASSERT(!my_properties.genuine && my_properties.outermost,
                    "Only a coroutine on outermost level can be left.");
                // Leave the outermost coroutine
                return false;
#endif
#if TBB_USE_ASSERT
            case task::ready:
                __TBB_ASSERT( false, "task is in READY state upon return from method execute()" );
                break;
#endif
            default:
                __TBB_ASSERT( false, "illegal state" );
                break;
        }
        GATHER_STATISTIC( t_next ? ++my_counters.spawns_bypassed : 0 );
        t = t_next;
    } // end of scheduler bypass loop
    return true;
}

References __TBB_ASSERT, __TBB_ASSERT_EX, __TBB_fallthrough, __TBB_ISOLATION_ARG, __TBB_ISOLATION_EXPR, tbb::task::allocated, tbb::internal::assert_task_valid(), tbb::internal::task_prefix::context, tbb::task::execute(), tbb::task::executing, GATHER_STATISTIC, tbb::internal::is_critical(), tbb::task::is_enqueued_task(), tbb::internal::task_prefix::isolation, tbb::task_group_context::itt_caller, ITT_NOTIFY, ITT_STACK, tbb::task_group_context::my_cancellation_requested, tbb::internal::task_prefix::next, tbb::internal::task_prefix::next_offloaded, tbb::internal::no_isolation, tbb::internal::task_prefix::owner, p, tbb::task::parent(), tbb::internal::poison_pointer(), tbb::task::prefix(), tbb::task::ready, tbb::task::recycle, tbb::task::reexecute, tbb::internal::task_prefix::ref_count, tbb::internal::reset_extra_state(), s, tbb::internal::context_guard_helper< T >::set_ctx(), tbb::internal::task_prefix::state, tbb::task::state(), tbb::internal::arena::wakeup, and tbb::task::~task().

Here is the call graph for this function:

receive_or_steal_task()

template<typename SchedulerTraits >

task * tbb::internal::custom_scheduler< SchedulerTraits >::receive_or_steal_task ( __TBB_ISOLATION_ARG(__TBB_atomic reference_count &completion_ref_count, isolation_tag isolation)  )

virtual

Try getting a task from the mailbox or stealing from another scheduler.

Returns the stolen task or NULL if all attempts fail.

Implements tbb::internal::generic_scheduler.

Definition at line 156 of file custom_scheduler.h.

                                                                                                                                                                 {
    task* t = NULL;
    bool outermost_worker_level = worker_outermost_level();
    bool outermost_dispatch_level = outermost_worker_level || master_outermost_level();
    bool can_steal_here = can_steal();
    bool outermost_current_worker_level = outermost_worker_level;
#if __TBB_PREVIEW_RESUMABLE_TASKS
    outermost_current_worker_level &= my_properties.genuine;
#endif
    my_inbox.set_is_idle( true );
#if __TBB_HOARD_NONLOCAL_TASKS
    __TBB_ASSERT(!my_nonlocal_free_list, NULL);
#endif
#if __TBB_TASK_PRIORITY
    if ( outermost_dispatch_level ) {
        if ( intptr_t skipped_priority = my_arena->my_skipped_fifo_priority ) {
            // This thread can dequeue FIFO tasks, and some priority levels of
            // FIFO tasks have been bypassed (to prevent deadlock caused by
            // dynamic priority changes in nested task group hierarchy).
            if ( my_arena->my_skipped_fifo_priority.compare_and_swap(0, skipped_priority) == skipped_priority
                 && skipped_priority > my_arena->my_top_priority )
            {
                my_market->update_arena_priority( *my_arena, skipped_priority );
            }
        }
    }
#endif /* !__TBB_TASK_PRIORITY */
    // TODO: Try to find a place to reset my_limit (under market's lock)
    // The number of slots potentially used in the arena. Updated once in a while, as my_limit changes rarely.
    size_t n = my_arena->my_limit-1;
    int yield_count = 0;
    // 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) {
        __TBB_ASSERT( my_arena->my_limit > 0, NULL );
        __TBB_ASSERT( my_arena_index <= n, NULL );
        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);
            }
            __TBB_ASSERT( !t, NULL );
            // A worker thread in its outermost dispatch loop (i.e. its execution stack is empty) should
            // exit it either when there is no more work in the current arena, or when revoked by the market.
            __TBB_ASSERT( !outermost_worker_level, NULL );
            __TBB_control_consistency_helper(); // on ref_count
            break; // exit stealing loop and return;
        }
        // Check if the resource manager requires our arena to relinquish some threads
        if ( outermost_current_worker_level ) {
            if ( ( my_arena->my_num_workers_allotted < my_arena->num_workers_active() ) ) {
                if ( SchedulerTraits::itt_possible && failure_count != -1 )
                    ITT_NOTIFY(sync_cancel, this);
                return NULL;
            }
        }
#if __TBB_PREVIEW_RESUMABLE_TASKS
        else if ( *my_arena_slot->my_scheduler_is_recalled ) {
            // Original scheduler was requested, return from stealing loop and recall.
            if ( my_inbox.is_idle_state(true) )
                my_inbox.set_is_idle(false);
            return NULL;
        }
#endif
#if __TBB_TASK_PRIORITY
        const int p = int(my_arena->my_top_priority);
#else /* !__TBB_TASK_PRIORITY */
        static const int p = 0;
#endif
        // Check if there are tasks mailed to this thread via task-to-thread affinity mechanism.
        __TBB_ASSERT(my_affinity_id, NULL);
        if ( n && !my_inbox.empty() ) {
            t = get_mailbox_task( __TBB_ISOLATION_EXPR( isolation ) );
#if __TBB_TASK_ISOLATION
            // There is a race with a thread adding a new task (possibly with suitable isolation)
            // to our mailbox, so the below conditions might result in a false positive.
            // Then set_is_idle(false) allows that task to be stolen; it's OK.
            if ( isolation != no_isolation && !t && !my_inbox.empty()
                     && my_inbox.is_idle_state( true ) ) {
                // We have proxy tasks in our mailbox but the isolation blocks their execution.
                // So publish the proxy tasks in mailbox to be available for stealing from owner's task pool.
                my_inbox.set_is_idle( false );
            }
#endif /* __TBB_TASK_ISOLATION */
        }
        if ( t ) {
            GATHER_STATISTIC( ++my_counters.mails_received );
        }
        // Check if there are tasks in starvation-resistant stream.
        // Only allowed at the outermost dispatch level without isolation.
        else if (__TBB_ISOLATION_EXPR(isolation == no_isolation &&) outermost_dispatch_level &&
                 !my_arena->my_task_stream.empty(p) && (
#if __TBB_PREVIEW_CRITICAL_TASKS && __TBB_CPF_BUILD
                     t = my_arena->my_task_stream.pop( p, subsequent_lane_selector(my_arena_slot->hint_for_pop) )
#else
                     t = my_arena->my_task_stream.pop( p, my_arena_slot->hint_for_pop )
#endif
                 ) ) {
            ITT_NOTIFY(sync_acquired, &my_arena->my_task_stream);
            // just proceed with the obtained task
        }
#if __TBB_TASK_PRIORITY
        // Check if any earlier offloaded non-top priority tasks become returned to the top level
        else if ( my_offloaded_tasks && (t = reload_tasks( __TBB_ISOLATION_EXPR( isolation ) )) ) {
            __TBB_ASSERT( !is_proxy(*t), "The proxy task cannot be offloaded" );
            // just proceed with the obtained task
        }
#endif /* __TBB_TASK_PRIORITY */
        else if ( can_steal_here && n && (t = steal_task( __TBB_ISOLATION_EXPR(isolation) )) ) {
            // just proceed with the obtained task
        }
#if __TBB_PREVIEW_CRITICAL_TASKS
        else if( (t = get_critical_task( __TBB_ISOLATION_EXPR(isolation) )) ) {
            __TBB_ASSERT( internal::is_critical(*t), "Received task must be critical one" );
            ITT_NOTIFY(sync_acquired, &my_arena->my_critical_task_stream);
            // just proceed with the obtained task
        }
#endif // __TBB_PREVIEW_CRITICAL_TASKS
        else
            goto fail;
        // A task was successfully obtained somewhere
        __TBB_ASSERT(t,NULL);
#if __TBB_ARENA_OBSERVER
        my_arena->my_observers.notify_entry_observers( my_last_local_observer, is_worker() );
#endif
#if __TBB_SCHEDULER_OBSERVER
        the_global_observer_list.notify_entry_observers( my_last_global_observer, is_worker() );
#endif /* __TBB_SCHEDULER_OBSERVER */
        if ( SchedulerTraits::itt_possible && 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);
        }
        break; // exit stealing loop and return
fail:
        GATHER_STATISTIC( ++my_counters.steals_failed );
        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.
        prolonged_pause();
        const int failure_threshold = 2*int(n+1);
        if( failure_count>=failure_threshold ) {
#if __TBB_YIELD2P
            failure_count = 0;
#else
            failure_count = failure_threshold;
#endif
            __TBB_Yield();
#if __TBB_TASK_PRIORITY
            // Check if there are tasks abandoned by other workers
            if ( my_arena->my_orphaned_tasks ) {
                // Epoch must be advanced before seizing the list pointer
                ++my_arena->my_abandonment_epoch;
                task* orphans = (task*)__TBB_FetchAndStoreW( &my_arena->my_orphaned_tasks, 0 );
                if ( orphans ) {
                    task** link = NULL;
                    // Get local counter out of the way (we've just brought in external tasks)
                    my_local_reload_epoch--;
                    t = reload_tasks( orphans, link, __TBB_ISOLATION_ARG( effective_reference_priority(), isolation ) );
                    if ( orphans ) {
                        *link = my_offloaded_tasks;
                        if ( !my_offloaded_tasks )
                            my_offloaded_task_list_tail_link = link;
                        my_offloaded_tasks = orphans;
                    }
                    __TBB_ASSERT( !my_offloaded_tasks == !my_offloaded_task_list_tail_link, NULL );
                    if ( t ) {
                        if( SchedulerTraits::itt_possible )
                            ITT_NOTIFY(sync_cancel, this);
                        __TBB_ASSERT( !is_proxy(*t), "The proxy task cannot be offloaded" );
                        break; // exit stealing loop and return
                    }
                }
            }
#endif /* __TBB_TASK_PRIORITY */
#if __APPLE__
            // threshold value tuned separately for macOS due to high cost of sched_yield there
            const int yield_threshold = 10;
#else
            const int yield_threshold = 100;
#endif
            if( yield_count++ >= yield_threshold ) {
                // 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 __TBB_TASK_PRIORITY
                if( outermost_current_worker_level || my_arena->my_top_priority > my_arena->my_bottom_priority ) {
                    if ( my_arena->is_out_of_work() && outermost_current_worker_level ) {
#else /* !__TBB_TASK_PRIORITY */
                    if ( outermost_current_worker_level && my_arena->is_out_of_work() ) {
#endif /* !__TBB_TASK_PRIORITY */
                        if( SchedulerTraits::itt_possible )
                            ITT_NOTIFY(sync_cancel, this);
                        return NULL;
                    }
#if __TBB_TASK_PRIORITY
                }
                if ( my_offloaded_tasks ) {
                    // Safeguard against any sloppiness in managing reload epoch
                    // counter (e.g. on the hot path because of performance reasons).
                    my_local_reload_epoch--;
                    // Break the deadlock caused by a higher priority dispatch loop
                    // stealing and offloading a lower priority task. Priority check
                    // at the stealing moment cannot completely preclude such cases
                    // because priorities can changes dynamically.
                    if ( !outermost_worker_level && *my_ref_top_priority > my_arena->my_top_priority ) {
                        GATHER_STATISTIC( ++my_counters.prio_ref_fixups );
                        my_ref_top_priority = &my_arena->my_top_priority;
                        // it's expected that only outermost workers can use global reload epoch
                        __TBB_ASSERT(my_ref_reload_epoch == &my_arena->my_reload_epoch, NULL);
                    }
                }
#endif /* __TBB_TASK_PRIORITY */
            } // end of arena snapshot branch
            // If several attempts did not find work, re-read the arena limit.
            n = my_arena->my_limit-1;
        } // end of yielding branch
    } // end of nonlocal task retrieval loop
    if ( my_inbox.is_idle_state( true ) )
        my_inbox.set_is_idle( false );
    return t;
}

References __TBB_ASSERT, __TBB_control_consistency_helper, __TBB_ISOLATION_ARG, __TBB_ISOLATION_EXPR, __TBB_Yield, GATHER_STATISTIC, int, tbb::internal::is_critical(), ITT_NOTIFY, tbb::internal::no_isolation, p, tbb::internal::prolonged_pause(), and sync_cancel.

Here is the call graph for this function:

tally_completion_of_predecessor()

template<typename SchedulerTraits >

void tbb::internal::custom_scheduler< SchedulerTraits >::tally_completion_of_predecessor ( task &  s, __TBB_ISOLATION_ARG(task *&bypass_slot, isolation_tag isolation)    )

inlineprivate

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.

Definition at line 72 of file custom_scheduler.h.

                                                                                                                        {
        task_prefix& p = s.prefix();
        __TBB_ASSERT(p.ref_count > 0, NULL);
        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 {
            reference_count old_ref_count = __TBB_FetchAndDecrementWrelease(&p.ref_count);
#if __TBB_PREVIEW_RESUMABLE_TASKS
            if (old_ref_count == internal::abandon_flag + 2) {
                // Remove the abandon flag.
                p.ref_count = 1;
                // The wait has been completed. Spawn a resume task.
                tbb::task::resume(p.abandoned_scheduler);
                return;
            }
#endif
            if (old_ref_count > 1) {
                // more references exist
                // '__TBB_cl_evict(&p)' degraded performance of parallel_preorder example
                return;
            }
        }

        // Ordering on p.ref_count (superfluous if SchedulerTraits::has_slow_atomic)
        __TBB_control_consistency_helper();
        __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 __TBB_TASK_ISOLATION
        if ( isolation != no_isolation ) {
            // The parent is allowed not to have isolation (even if a child has isolation) because it has never spawned.
            __TBB_ASSERT(p.isolation == no_isolation || p.isolation == isolation, NULL);
            p.isolation = isolation;
        }
#endif /* __TBB_TASK_ISOLATION */

#if __TBB_RECYCLE_TO_ENQUEUE
        if (p.state==task::to_enqueue) {
            // related to __TBB_TASK_ARENA TODO: try keep priority of the task
            // e.g. rework task_prefix to remember priority of received task and use here
            my_arena->enqueue_task(s, 0, my_random );
        } else
#endif /*__TBB_RECYCLE_TO_ENQUEUE*/
        if( bypass_slot==NULL )
            bypass_slot = &s;
#if __TBB_PREVIEW_CRITICAL_TASKS
        else if( internal::is_critical( s ) ) {
            local_spawn( bypass_slot, bypass_slot->prefix().next );
            bypass_slot = &s;
        }
#endif /* __TBB_PREVIEW_CRITICAL_TASKS */
        else
            local_spawn( &s, s.prefix().next );
    }

References __TBB_ASSERT, __TBB_control_consistency_helper, __TBB_FetchAndDecrementWrelease, tbb::internal::arena::enqueue_task(), tbb::internal::es_ref_count_active, tbb::internal::is_critical(), ITT_NOTIFY, tbb::internal::generic_scheduler::local_spawn(), tbb::internal::scheduler_state::my_arena, tbb::internal::generic_scheduler::my_random, tbb::internal::task_prefix::next, tbb::internal::no_isolation, p, tbb::task::prefix(), s, and sync_releasing.

Here is the call graph for this function:

wait_for_all()

template<typename SchedulerTraits >

void tbb::internal::custom_scheduler< SchedulerTraits >::wait_for_all ( task &  parent, task *  child  )

inlineprivatevirtual

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.

Implements tbb::internal::scheduler.

Definition at line 65 of file custom_scheduler.h.

                                                                  {
        static_cast<custom_scheduler*>(governor::local_scheduler())->scheduler_type::local_wait_for_all( parent, child );
    }

References tbb::internal::governor::local_scheduler(), and parent.

Here is the call graph for this function:

---

The documentation for this class was generated from the following file:

• custom_scheduler.h