market.h

Go to the documentation of this file.

/*
    Copyright (c) 2005-2020 Intel Corporation

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

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

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

#ifndef _TBB_market_H
#define _TBB_market_H

#include "tbb/tbb_stddef.h"

#include "scheduler_common.h"
#include "tbb/atomic.h"
#include "tbb/spin_rw_mutex.h"
#include "../rml/include/rml_tbb.h"

#include "intrusive_list.h"

#if defined(_MSC_VER) && defined(_Wp64)
    // Workaround for overzealous compiler warnings in /Wp64 mode
    #pragma warning (push)
    #pragma warning (disable: 4244)
#endif

namespace tbb {

class task_group_context;

namespace internal {

//------------------------------------------------------------------------
// Class market
//------------------------------------------------------------------------

class market : no_copy, rml::tbb_client {
    friend class generic_scheduler;
    friend class arena;
    friend class tbb::interface7::internal::task_arena_base;
    template<typename SchedulerTraits> friend class custom_scheduler;
    friend class tbb::task_group_context;
private:
    friend void ITT_DoUnsafeOneTimeInitialization ();

    typedef intrusive_list<arena> arena_list_type;
    typedef intrusive_list<generic_scheduler> scheduler_list_type;

    static market* theMarket;

    typedef scheduler_mutex_type global_market_mutex_type;

    static global_market_mutex_type  theMarketMutex;

    typedef spin_rw_mutex arenas_list_mutex_type;
    arenas_list_mutex_type my_arenas_list_mutex;

    rml::tbb_server* my_server;


    unsigned my_num_workers_hard_limit;


    unsigned my_num_workers_soft_limit;

    int my_num_workers_requested;


    atomic<unsigned> my_first_unused_worker_idx;

    int my_total_demand;

#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
    int my_mandatory_num_requested;
#endif

#if __TBB_TASK_PRIORITY

    intptr_t my_global_top_priority;


    intptr_t my_global_bottom_priority;


    uintptr_t my_global_reload_epoch;

    struct priority_level_info {
        arena_list_type arenas;


        arena *next_arena;

        int workers_requested;

        int workers_available;
    }; // struct priority_level_info

    priority_level_info my_priority_levels[num_priority_levels];

#else /* !__TBB_TASK_PRIORITY */

    arena_list_type my_arenas;


    arena *my_next_arena;
#endif /* !__TBB_TASK_PRIORITY */

    uintptr_t my_arenas_aba_epoch;

    unsigned my_ref_count;

    unsigned my_public_ref_count;

    size_t my_stack_size;

    bool my_join_workers;

    static const unsigned skip_soft_limit_warning = ~0U;

    unsigned my_workers_soft_limit_to_report;
#if __TBB_COUNT_TASK_NODES

    atomic<intptr_t> my_task_node_count;
#endif /* __TBB_COUNT_TASK_NODES */

    market ( unsigned workers_soft_limit, unsigned workers_hard_limit, size_t stack_size );

    static market& global_market ( bool is_public, unsigned max_num_workers = 0, size_t stack_size = 0 );

    void destroy ();

    int update_workers_request();

#if __TBB_TASK_PRIORITY
    arena* arena_in_need ( arena* prev_arena );


    void update_allotment ( intptr_t highest_affected_priority );

    void update_arena_top_priority ( arena& a, intptr_t newPriority );

    inline void update_global_top_priority ( intptr_t newPriority );

    inline void reset_global_priority ();

    inline void advance_global_reload_epoch () {
        __TBB_store_with_release( my_global_reload_epoch, my_global_reload_epoch + 1 );
    }

    void assert_market_valid () const {
        __TBB_ASSERT( (my_priority_levels[my_global_top_priority].workers_requested > 0
                           && !my_priority_levels[my_global_top_priority].arenas.empty())
                       || (my_global_top_priority == my_global_bottom_priority &&
                           my_global_top_priority == normalized_normal_priority), NULL );
    }

#else /* !__TBB_TASK_PRIORITY */


    void update_allotment (unsigned effective_soft_limit) {
        if ( my_total_demand )
            update_allotment( my_arenas, my_total_demand, (int)effective_soft_limit );
    }

    // TODO: consider to rewrite the code with is_arena_in_list function
    arena* arena_in_need (arena* prev_arena) {
        if(__TBB_load_with_acquire(my_total_demand) <= 0)
            return NULL;
        arenas_list_mutex_type::scoped_lock lock(my_arenas_list_mutex, /*is_writer=*/false);
        arena* a = NULL;

        // Checks if arena is alive or not
        if ( is_arena_in_list( my_arenas, prev_arena ) ) {
            a = arena_in_need( my_arenas, prev_arena );
        } else {
            a = arena_in_need( my_arenas, my_next_arena );
            if (a) {
                as_atomic(my_next_arena) = a; // a subject for innocent data race under the reader lock
                // TODO: rework global round robin policy to local or random to avoid this write
            }
        }

        return a;
    }

    void assert_market_valid () const {}
#endif /* !__TBB_TASK_PRIORITY */

    // Helpers to unify code branches dependent on priority feature presence

    void insert_arena_into_list ( arena& a );

    void remove_arena_from_list ( arena& a );

    arena* arena_in_need ( arena_list_type &arenas, arena *hint );

    int update_allotment ( arena_list_type& arenas, int total_demand, int max_workers );

    bool is_arena_in_list( arena_list_type &arenas, arena *a );


    // Implementation of rml::tbb_client interface methods

    version_type version () const __TBB_override { return 0; }

    unsigned max_job_count () const __TBB_override { return my_num_workers_hard_limit; }

    size_t min_stack_size () const __TBB_override { return worker_stack_size(); }

    policy_type policy () const __TBB_override { return throughput; }

    job* create_one_job () __TBB_override;

    void cleanup( job& j ) __TBB_override;

    void acknowledge_close_connection () __TBB_override;

    void process( job& j ) __TBB_override;

public:

    static arena* create_arena ( int num_slots, int num_reserved_slots, size_t stack_size );

    void try_destroy_arena ( arena*, uintptr_t aba_epoch );

    void detach_arena ( arena& );

    bool release ( bool is_public, bool blocking_terminate );

#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
    void enable_mandatory_concurrency_impl ( arena *a );

    void enable_mandatory_concurrency ( arena *a );

    void disable_mandatory_concurrency_impl(arena* a);

    void mandatory_concurrency_disable ( arena *a );
#endif /* __TBB_ENQUEUE_ENFORCED_CONCURRENCY */


    void adjust_demand ( arena&, int delta );

    bool must_join_workers () const { return my_join_workers; }

    size_t worker_stack_size () const { return my_stack_size; }

    static void set_active_num_workers( unsigned w );

    static unsigned app_parallelism_limit();

#if _WIN32||_WIN64
    void register_master( ::rml::server::execution_resource_t& rsc_handle ) {
        __TBB_ASSERT( my_server, "RML server not defined?" );
        // the server may ignore registration and set master_exec_resource to NULL.
        my_server->register_master( rsc_handle );
    }

    void unregister_master( ::rml::server::execution_resource_t& rsc_handle ) const {
        my_server->unregister_master( rsc_handle );
    }
#endif /* WIN */

#if __TBB_TASK_GROUP_CONTEXT

    template <typename T>
    bool propagate_task_group_state ( T task_group_context::*mptr_state, task_group_context& src, T new_state );
#endif /* __TBB_TASK_GROUP_CONTEXT */

#if __TBB_TASK_PRIORITY

    bool lower_arena_priority ( arena& a, intptr_t new_priority, uintptr_t old_reload_epoch );


    bool update_arena_priority ( arena& a, intptr_t new_priority );
#endif /* __TBB_TASK_PRIORITY */

#if __TBB_COUNT_TASK_NODES

    void update_task_node_count( intptr_t delta ) { my_task_node_count += delta; }
#endif /* __TBB_COUNT_TASK_NODES */

#if __TBB_TASK_GROUP_CONTEXT
    scheduler_list_type my_masters;


    generic_scheduler* my_workers[1];
#endif /* __TBB_TASK_GROUP_CONTEXT */

    static unsigned max_num_workers() {
        global_market_mutex_type::scoped_lock lock( theMarketMutex );
        return theMarket? theMarket->my_num_workers_hard_limit : 0;
    }
}; // class market

} // namespace internal
} // namespace tbb

#if defined(_MSC_VER) && defined(_Wp64)
    // Workaround for overzealous compiler warnings in /Wp64 mode
    #pragma warning (pop)
#endif // warning 4244 is back

#endif /* _TBB_market_H */