scheduler.h

Go to the documentation of this file.

/*
    Copyright (c) 2005-2019 Intel Corporation

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.




*/

#ifndef _TBB_scheduler_H
#define _TBB_scheduler_H

#include "scheduler_common.h"
#include "tbb/spin_mutex.h"
#include "mailbox.h"
#include "tbb_misc.h" // for FastRandom
#include "itt_notify.h"
#include "../rml/include/rml_tbb.h"

#include "intrusive_list.h"

#if __TBB_SURVIVE_THREAD_SWITCH
#include "cilk-tbb-interop.h"
#endif /* __TBB_SURVIVE_THREAD_SWITCH */

namespace tbb {
namespace internal {

template<typename SchedulerTraits> class custom_scheduler;

//------------------------------------------------------------------------
// generic_scheduler
//------------------------------------------------------------------------

#define EmptyTaskPool ((task**)0)
#define LockedTaskPool ((task**)~(intptr_t)0)

struct scheduler_properties {
    static const bool worker = false;
    static const bool master = true;
    bool type : 1;

    bool outermost : 1;
#if __TBB_PREVIEW_CRITICAL_TASKS
    bool has_taken_critical_task : 1;

    unsigned char : 5;
#else
    unsigned char : 6;
#endif /* __TBB_PREVIEW_CRITICAL_TASKS */
};

struct scheduler_state {
    size_t my_arena_index; // TODO: make it unsigned and pair with my_affinity_id to fit into cache line

    arena_slot* my_arena_slot;

    arena* my_arena;

    task* my_innermost_running_task;


    mail_inbox my_inbox;


    affinity_id my_affinity_id;

    scheduler_properties my_properties;

#if __TBB_SCHEDULER_OBSERVER
    observer_proxy* my_last_global_observer;
#endif

#if __TBB_ARENA_OBSERVER
    observer_proxy* my_last_local_observer;
#endif
#if __TBB_TASK_PRIORITY

    volatile intptr_t *my_ref_top_priority;

    volatile uintptr_t *my_ref_reload_epoch;
#endif /* __TBB_TASK_PRIORITY */
};


class generic_scheduler: public scheduler
                       , public ::rml::job
                       , public intrusive_list_node
                       , public scheduler_state {
public: // almost every class in TBB uses generic_scheduler

    static const size_t quick_task_size = 256-task_prefix_reservation_size;

    static bool is_version_3_task( task& t ) {
#if __TBB_PREVIEW_CRITICAL_TASKS
        return (t.prefix().extra_state & 0x7)>=0x1;
#else
        return (t.prefix().extra_state & 0x0F)>=0x1;
#endif
    }

    uintptr_t my_stealing_threshold;
#if __TBB_ipf
    uintptr_t my_rsb_stealing_threshold;
#endif

    static const size_t null_arena_index = ~size_t(0);

    inline bool is_task_pool_published () const;

    inline bool is_local_task_pool_quiescent () const;

    inline bool is_quiescent_local_task_pool_empty () const;

    inline bool is_quiescent_local_task_pool_reset () const;

    market* my_market;

    FastRandom my_random;

    task* my_free_list;

#if __TBB_HOARD_NONLOCAL_TASKS
    task* my_nonlocal_free_list;
#endif

    task* my_dummy_task;


    long my_ref_count;

    inline void attach_mailbox( affinity_id id );

    /* A couple of bools can be located here because space is otherwise just padding after my_affinity_id. */

    bool my_auto_initialized;

#if __TBB_COUNT_TASK_NODES
    intptr_t my_task_node_count;
#endif /* __TBB_COUNT_TASK_NODES */

    void init_stack_info ();

    bool can_steal () {
        int anchor;
        // TODO IDEA: Add performance warning?
#if __TBB_ipf
        return my_stealing_threshold < (uintptr_t)&anchor && (uintptr_t)__TBB_get_bsp() < my_rsb_stealing_threshold;
#else
        return my_stealing_threshold < (uintptr_t)&anchor;
#endif
    }


    void publish_task_pool();


    void leave_task_pool();


    inline void reset_task_pool_and_leave ();


    task** lock_task_pool( arena_slot* victim_arena_slot ) const;


    void unlock_task_pool( arena_slot* victim_arena_slot, task** victim_task_pool ) const;


    void acquire_task_pool() const;


    void release_task_pool() const;


    task* prepare_for_spawning( task* t );

    inline void commit_spawned_tasks( size_t new_tail );


    inline void commit_relocated_tasks( size_t new_tail );


    task* get_task( __TBB_ISOLATION_EXPR( isolation_tag isolation ) );


#if __TBB_TASK_ISOLATION
    task* get_task( size_t T, isolation_tag isolation, bool& tasks_omitted );
#else
    task* get_task( size_t T );
#endif /* __TBB_TASK_ISOLATION */

    task* get_mailbox_task( __TBB_ISOLATION_EXPR( isolation_tag isolation ) );

    static bool is_proxy( const task& t ) {
        return t.prefix().extra_state==es_task_proxy;
    }

    task* steal_task( __TBB_ISOLATION_EXPR(isolation_tag isolation) );

    task* steal_task_from( __TBB_ISOLATION_ARG( arena_slot& victim_arena_slot, isolation_tag isolation ) );

#if __TBB_PREVIEW_CRITICAL_TASKS
    task* get_critical_task( __TBB_ISOLATION_EXPR(isolation_tag isolation) );

    bool handled_as_critical( task& t );
#endif

    static const size_t min_task_pool_size = 64;


    size_t prepare_task_pool( size_t n );

    static generic_scheduler* create_master( arena* a );

    bool cleanup_master( bool blocking_terminate );

    static generic_scheduler* create_worker( market& m, size_t index );

    static void cleanup_worker( void* arg, bool worker );

protected:
    template<typename SchedulerTraits> friend class custom_scheduler;
    generic_scheduler( market & );

public:
#if TBB_USE_ASSERT > 1

    void assert_task_pool_valid() const;
#else
    void assert_task_pool_valid() const {}
#endif /* TBB_USE_ASSERT <= 1 */

    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;

    void spawn_root_and_wait( task& first, task*& next ) __TBB_override;

    void enqueue( task&, void* reserved ) __TBB_override;

    void local_spawn( task* first, task*& next );
    void local_spawn_root_and_wait( task* first, task*& next );
    virtual void local_wait_for_all( task& parent, task* child ) = 0;

    void free_scheduler();


    task& allocate_task( size_t number_of_bytes,
                       __TBB_CONTEXT_ARG(task* parent, task_group_context* context) );


    template<free_task_hint h>
    void free_task( task& t );

    inline void deallocate_task( task& t );

    inline bool is_worker() const;

    inline bool outermost_level() const;


    inline bool master_outermost_level () const;

    inline bool worker_outermost_level () const;

    unsigned max_threads_in_arena();

#if __TBB_COUNT_TASK_NODES
    intptr_t get_task_node_count( bool count_arena_workers = false );
#endif /* __TBB_COUNT_TASK_NODES */

    static task* plugged_return_list() {return (task*)(intptr_t)(-1);}

    __TBB_atomic intptr_t my_small_task_count;

    // TODO IDEA: see if putting my_return_list on separate cache line improves performance
    task* my_return_list;


    virtual task* receive_or_steal_task( __TBB_ISOLATION_ARG( __TBB_atomic reference_count& completion_ref_count, isolation_tag isolation ) ) = 0;

    void free_nonlocal_small_task( task& t );

#if __TBB_TASK_GROUP_CONTEXT

    inline task_group_context* default_context ();

    char _padding1[NFS_MaxLineSize - sizeof(context_list_node_t)];

    context_list_node_t my_context_list_head;

    // TODO: check whether it can be deadly preempted and replace by spinning/sleeping mutex
    spin_mutex my_context_list_mutex;


    uintptr_t my_context_state_propagation_epoch;


    tbb::atomic<uintptr_t> my_local_ctx_list_update;

#if __TBB_TASK_PRIORITY
    inline intptr_t effective_reference_priority () const;

    // TODO: move into slots and fix is_out_of_work
    task* my_offloaded_tasks;

    task** my_offloaded_task_list_tail_link;

    uintptr_t my_local_reload_epoch;

    volatile bool my_pool_reshuffling_pending;


    task* reload_tasks( __TBB_ISOLATION_EXPR( isolation_tag isolation ) );

    task* reload_tasks( task*& offloaded_tasks, task**& offloaded_task_list_link, __TBB_ISOLATION_ARG( intptr_t top_priority, isolation_tag isolation ) );


    task* winnow_task_pool ( __TBB_ISOLATION_EXPR( isolation_tag isolation ) );


    task *get_task_and_activate_task_pool( size_t H0 , __TBB_ISOLATION_ARG( size_t T0, isolation_tag isolation ) );

    inline void offload_task ( task& t, intptr_t task_priority );
#endif /* __TBB_TASK_PRIORITY */


    void cleanup_local_context_list ();

    template <typename T>
    void propagate_task_group_state ( T task_group_context::*mptr_state, task_group_context& src, T new_state );

    // check consistency
    static void assert_context_valid(const task_group_context *tgc) {
        suppress_unused_warning(tgc);
#if TBB_USE_ASSERT
        __TBB_ASSERT(tgc, NULL);
        uintptr_t ctx = tgc->my_version_and_traits;
        __TBB_ASSERT(is_alive(ctx), "referenced task_group_context was destroyed");
        static const char *msg = "task_group_context is invalid";
        __TBB_ASSERT(!(ctx&~(3|(7<<task_group_context::traits_offset))), msg); // the value fits known values of versions and traits
        __TBB_ASSERT(tgc->my_kind < task_group_context::dying, msg);
        __TBB_ASSERT(tgc->my_cancellation_requested == 0 || tgc->my_cancellation_requested == 1, msg);
        __TBB_ASSERT(tgc->my_state < task_group_context::low_unused_state_bit, msg);
        if(tgc->my_kind != task_group_context::isolated) {
            __TBB_ASSERT(tgc->my_owner, msg);
            __TBB_ASSERT(tgc->my_node.my_next && tgc->my_node.my_prev, msg);
        }
#if __TBB_TASK_PRIORITY
        assert_priority_valid(tgc->my_priority);
#endif
        if(tgc->my_parent)
#if TBB_USE_ASSERT > 1
            assert_context_valid(tgc->my_parent);
#else
            __TBB_ASSERT(is_alive(tgc->my_parent->my_version_and_traits), msg);
#endif
#endif
    }
#endif /* __TBB_TASK_GROUP_CONTEXT */

#if _WIN32||_WIN64
private:
    ::rml::server::execution_resource_t master_exec_resource;
public:
#endif /* _WIN32||_WIN64 */

#if __TBB_TASK_GROUP_CONTEXT

    tbb::atomic<uintptr_t> my_nonlocal_ctx_list_update;
#endif /* __TBB_TASK_GROUP_CONTEXT */

#if __TBB_SURVIVE_THREAD_SWITCH
    __cilk_tbb_unwatch_thunk my_cilk_unwatch_thunk;
#if TBB_USE_ASSERT

    enum cilk_state_t {
        cs_none=0xF000, // Start at nonzero value so that we can detect use of zeroed memory.
        cs_running,
        cs_limbo,
        cs_freed
    };
    cilk_state_t my_cilk_state;
#endif /* TBB_USE_ASSERT */
#endif /* __TBB_SURVIVE_THREAD_SWITCH */

#if __TBB_STATISTICS

    mutable statistics_counters my_counters;
#endif /* __TBB_STATISTICS */

}; // class generic_scheduler


} // namespace internal
} // namespace tbb

#include "arena.h"
#include "governor.h"

namespace tbb {
namespace internal {

inline bool generic_scheduler::is_task_pool_published () const {
    __TBB_ASSERT(my_arena_slot, 0);
    return my_arena_slot->task_pool != EmptyTaskPool;
}

inline bool generic_scheduler::is_local_task_pool_quiescent () const {
    __TBB_ASSERT(my_arena_slot, 0);
    task** tp = my_arena_slot->task_pool;
    return tp == EmptyTaskPool || tp == LockedTaskPool;
}

inline bool generic_scheduler::is_quiescent_local_task_pool_empty () const {
    __TBB_ASSERT( is_local_task_pool_quiescent(), "Task pool is not quiescent" );
    return __TBB_load_relaxed(my_arena_slot->head) == __TBB_load_relaxed(my_arena_slot->tail);
}

inline bool generic_scheduler::is_quiescent_local_task_pool_reset () const {
    __TBB_ASSERT( is_local_task_pool_quiescent(), "Task pool is not quiescent" );
    return __TBB_load_relaxed(my_arena_slot->head) == 0 && __TBB_load_relaxed(my_arena_slot->tail) == 0;
}

inline bool generic_scheduler::outermost_level () const {
    return my_properties.outermost;
}

inline bool generic_scheduler::master_outermost_level () const {
    return !is_worker() && outermost_level();
}

inline bool generic_scheduler::worker_outermost_level () const {
    return is_worker() && outermost_level();
}

#if __TBB_TASK_GROUP_CONTEXT
inline task_group_context* generic_scheduler::default_context () {
    return my_dummy_task->prefix().context;
}
#endif /* __TBB_TASK_GROUP_CONTEXT */

inline void generic_scheduler::attach_mailbox( affinity_id id ) {
    __TBB_ASSERT(id>0,NULL);
    my_inbox.attach( my_arena->mailbox(id) );
    my_affinity_id = id;
}

inline bool generic_scheduler::is_worker() const {
    return my_properties.type == scheduler_properties::worker;
}

inline unsigned generic_scheduler::max_threads_in_arena() {
    __TBB_ASSERT(my_arena, NULL);
    return my_arena->my_num_slots;
}

inline void generic_scheduler::deallocate_task( task& t ) {
#if TBB_USE_ASSERT
    task_prefix& p = t.prefix();
    p.state = 0xFF;
    p.extra_state = 0xFF;
    poison_pointer(p.next);
#endif /* TBB_USE_ASSERT */
    NFS_Free((char*)&t-task_prefix_reservation_size);
#if __TBB_COUNT_TASK_NODES
    --my_task_node_count;
#endif /* __TBB_COUNT_TASK_NODES */
}

#if __TBB_COUNT_TASK_NODES
inline intptr_t generic_scheduler::get_task_node_count( bool count_arena_workers ) {
    return my_task_node_count + (count_arena_workers? my_arena->workers_task_node_count(): 0);
}
#endif /* __TBB_COUNT_TASK_NODES */

inline void generic_scheduler::reset_task_pool_and_leave () {
    __TBB_ASSERT( my_arena_slot->task_pool == LockedTaskPool, "Task pool must be locked when resetting task pool" );
    __TBB_store_relaxed( my_arena_slot->tail, 0 );
    __TBB_store_relaxed( my_arena_slot->head, 0 );
    leave_task_pool();
}

//TODO: move to arena_slot
inline void generic_scheduler::commit_spawned_tasks( size_t new_tail ) {
    __TBB_ASSERT ( new_tail <= my_arena_slot->my_task_pool_size, "task deque end was overwritten" );
    // emit "task was released" signal
    ITT_NOTIFY(sync_releasing, (void*)((uintptr_t)my_arena_slot+sizeof(uintptr_t)));
    // Release fence is necessary to make sure that previously stored task pointers
    // are visible to thieves.
    __TBB_store_with_release( my_arena_slot->tail, new_tail );
}

void generic_scheduler::commit_relocated_tasks ( size_t new_tail ) {
    __TBB_ASSERT( is_local_task_pool_quiescent(),
                  "Task pool must be locked when calling commit_relocated_tasks()" );
    __TBB_store_relaxed( my_arena_slot->head, 0 );
    // Tail is updated last to minimize probability of a thread making arena
    // snapshot being misguided into thinking that this task pool is empty.
    __TBB_store_release( my_arena_slot->tail, new_tail );
    release_task_pool();
}

template<free_task_hint hint>
void generic_scheduler::free_task( task& t ) {
#if __TBB_HOARD_NONLOCAL_TASKS
    static const int h = hint&(~local_task);
#else
    static const free_task_hint h = hint;
#endif
    GATHER_STATISTIC(--my_counters.active_tasks);
    task_prefix& p = t.prefix();
    // Verify that optimization hints are correct.
    __TBB_ASSERT( h!=small_local_task || p.origin==this, NULL );
    __TBB_ASSERT( !(h&small_task) || p.origin, NULL );
    __TBB_ASSERT( !(h&local_task) || (!p.origin || uintptr_t(p.origin) > uintptr_t(4096)), "local_task means allocated");
    poison_value(p.depth);
    poison_value(p.ref_count);
    poison_pointer(p.owner);
    __TBB_ASSERT( 1L<<t.state() & (1L<<task::executing|1L<<task::allocated), NULL );
    p.state = task::freed;
    if( h==small_local_task || p.origin==this ) {
        GATHER_STATISTIC(++my_counters.free_list_length);
        p.next = my_free_list;
        my_free_list = &t;
    } else if( !(h&local_task) && p.origin && uintptr_t(p.origin) < uintptr_t(4096) ) {
        // a special value reserved for future use, do nothing since
        // origin is not pointing to a scheduler instance
    } else if( !(h&local_task) && p.origin ) {
        GATHER_STATISTIC(++my_counters.free_list_length);
#if __TBB_HOARD_NONLOCAL_TASKS
        if( !(h&no_cache) ) {
            p.next = my_nonlocal_free_list;
            my_nonlocal_free_list = &t;
        } else
#endif
        free_nonlocal_small_task(t);
    } else {
        GATHER_STATISTIC(--my_counters.big_tasks);
        deallocate_task(t);
    }
}

#if __TBB_TASK_PRIORITY
inline intptr_t generic_scheduler::effective_reference_priority () const {
    // Workers on the outermost dispatch level (i.e. with empty stack) use market's
    // priority as a reference point (to speedup discovering process level priority
    // changes). But when there are enough workers to service (even if only partially)
    // a lower priority arena, they should use arena's priority as a reference, lest
    // be trapped in a futile spinning (because market's priority would prohibit
    // executing ANY tasks in this arena).
    return !worker_outermost_level() ||
        (my_arena->my_num_workers_allotted < my_arena->num_workers_active()
#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
         && my_arena->my_concurrency_mode!=arena_base::cm_enforced_global
#endif
            ) ? *my_ref_top_priority : my_arena->my_top_priority;
}

inline void generic_scheduler::offload_task ( task& t, intptr_t /*priority*/ ) {
    GATHER_STATISTIC( ++my_counters.prio_tasks_offloaded );
    __TBB_ASSERT( !is_proxy(t), "The proxy task cannot be offloaded" );
    __TBB_ASSERT( my_offloaded_task_list_tail_link && !*my_offloaded_task_list_tail_link, NULL );
#if TBB_USE_ASSERT
    t.prefix().state = task::ready;
#endif /* TBB_USE_ASSERT */
    t.prefix().next_offloaded = my_offloaded_tasks;
    my_offloaded_tasks = &t;
}
#endif /* __TBB_TASK_PRIORITY */

#if __TBB_PREVIEW_CRITICAL_TASKS
class critical_task_count_guard : internal::no_copy {
public:
    critical_task_count_guard(scheduler_properties& properties, task& t)
        : my_properties(properties),
          my_original_critical_task_state(properties.has_taken_critical_task) {
        my_properties.has_taken_critical_task |= internal::is_critical(t);
    }
    ~critical_task_count_guard() {
        my_properties.has_taken_critical_task = my_original_critical_task_state;
    }
private:
    scheduler_properties& my_properties;
    bool my_original_critical_task_state;
};
#endif /* __TBB_PREVIEW_CRITICAL_TASKS */

#if __TBB_FP_CONTEXT || __TBB_TASK_GROUP_CONTEXT

template <bool report_tasks>
class context_guard_helper {
#if __TBB_TASK_GROUP_CONTEXT
    const task_group_context *curr_ctx;
#endif
#if __TBB_FP_CONTEXT
    cpu_ctl_env guard_cpu_ctl_env;
    cpu_ctl_env curr_cpu_ctl_env;
#endif
public:
    context_guard_helper()
#if __TBB_TASK_GROUP_CONTEXT
        : curr_ctx(NULL)
#endif
    {
#if __TBB_FP_CONTEXT
        guard_cpu_ctl_env.get_env();
        curr_cpu_ctl_env = guard_cpu_ctl_env;
#endif
    }
    ~context_guard_helper() {
#if __TBB_FP_CONTEXT
        if ( curr_cpu_ctl_env != guard_cpu_ctl_env )
            guard_cpu_ctl_env.set_env();
#endif
#if __TBB_TASK_GROUP_CONTEXT
        if (report_tasks && curr_ctx)
            ITT_TASK_END;
#endif
    }
    void set_ctx( const task_group_context *ctx ) {
        generic_scheduler::assert_context_valid(ctx);
#if __TBB_FP_CONTEXT
        const cpu_ctl_env &ctl = *punned_cast<cpu_ctl_env*>(&ctx->my_cpu_ctl_env);
#endif
#if __TBB_TASK_GROUP_CONTEXT
        if(ctx != curr_ctx) {
#endif
#if __TBB_FP_CONTEXT
            if ( ctl != curr_cpu_ctl_env ) {
                curr_cpu_ctl_env = ctl;
                curr_cpu_ctl_env.set_env();
            }
#endif
#if __TBB_TASK_GROUP_CONTEXT
            // if task group context was active, report end of current execution frame.
            if (report_tasks) {
                if (curr_ctx)
                    ITT_TASK_END;
                // reporting begin of new task group context execution frame.
                // using address of task group context object to group tasks (parent).
                // id of task execution frame is NULL and reserved for future use.
                ITT_TASK_BEGIN(ctx,ctx->my_name,NULL);
                curr_ctx = ctx;
            }
        }
#endif
    }
    void restore_default() {
#if __TBB_FP_CONTEXT
        if ( curr_cpu_ctl_env != guard_cpu_ctl_env ) {
            guard_cpu_ctl_env.set_env();
            curr_cpu_ctl_env = guard_cpu_ctl_env;
        }
#endif
    }
};
#else
template <bool T>
struct context_guard_helper {
    void set_ctx( __TBB_CONTEXT_ARG1(task_group_context *) ) {}
    void restore_default() {}
};
#endif /* __TBB_FP_CONTEXT */

} // namespace internal
} // namespace tbb

#endif /* _TBB_scheduler_H */