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 */
atomic.h
spin_rw_mutex.h
tbb_stddef.h
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
__TBB_override
#define __TBB_override
Definition: tbb_stddef.h:240
intrusive_list.h
scheduler_common.h
lock
void const char const char int ITT_FORMAT __itt_group_sync x void const char ITT_FORMAT __itt_group_sync s void ITT_FORMAT __itt_group_sync p void ITT_FORMAT p void ITT_FORMAT p no args __itt_suppress_mode_t unsigned int void size_t ITT_FORMAT d void ITT_FORMAT p void ITT_FORMAT p __itt_model_site __itt_model_site_instance ITT_FORMAT p __itt_model_task __itt_model_task_instance ITT_FORMAT p void * lock
Definition: ittnotify_static.h:121
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::__TBB_load_with_acquire
T __TBB_load_with_acquire(const volatile T &location)
Definition: tbb_machine.h:712
tbb::internal::scheduler_mutex_type
__TBB_SCHEDULER_MUTEX_TYPE scheduler_mutex_type
Mutex type for global locks in the scheduler.
Definition: scheduler_common.h:133
tbb::internal::as_atomic
atomic< T > & as_atomic(T &t)
Definition: atomic.h:572
tbb::internal::num_priority_levels
static const intptr_t num_priority_levels
Definition: scheduler_common.h:129
tbb::internal::__TBB_store_with_release
void __TBB_store_with_release(volatile T &location, V value)
Definition: tbb_machine.h:716
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::spin_rw_mutex_v3
Fast, unfair, spinning reader-writer lock with backoff and writer-preference.
Definition: spin_rw_mutex.h:38
tbb::spin_rw_mutex_v3::scoped_lock
The scoped locking pattern.
Definition: spin_rw_mutex.h:86
tbb::task_group_context
Used to form groups of tasks.
Definition: task.h:358
tbb::interface7::internal::task_arena_base
Definition: task_arena.h:105
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::arena
Definition: arena.h:277
tbb::internal::custom_scheduler
A scheduler with a customized evaluation loop.
Definition: custom_scheduler.h:52
tbb::internal::intrusive_list< arena >
tbb::internal::market
Definition: market.h:45
tbb::internal::market::my_arenas_aba_epoch
uintptr_t my_arenas_aba_epoch
ABA prevention marker to assign to newly created arenas.
Definition: market.h:143
tbb::internal::market::global_market
static market & global_market(bool is_public, unsigned max_num_workers=0, size_t stack_size=0)
Factory method creating new market object.
Definition: market.cpp:96
tbb::internal::market::my_num_workers_hard_limit
unsigned my_num_workers_hard_limit
Maximal number of workers allowed for use by the underlying resource manager.
Definition: market.h:74
tbb::internal::market::max_job_count
unsigned max_job_count() const __TBB_override
Definition: market.h:263
tbb::internal::market::theMarket
static market * theMarket
Currently active global market.
Definition: market.h:58
tbb::internal::market::arena_list_type
intrusive_list< arena > arena_list_type
Definition: market.h:54
tbb::internal::market::adjust_demand
void adjust_demand(arena &, int delta)
Request that arena's need in workers should be adjusted.
Definition: market.cpp:557
tbb::internal::market::my_first_unused_worker_idx
atomic< unsigned > my_first_unused_worker_idx
First unused index of worker.
Definition: market.h:86
tbb::internal::market::my_join_workers
bool my_join_workers
Shutdown mode.
Definition: market.h:155
tbb::internal::market::cleanup
void cleanup(job &j) __TBB_override
Definition: market.cpp:681
tbb::internal::market::my_public_ref_count
unsigned my_public_ref_count
Count of master threads attached.
Definition: market.h:149
tbb::internal::market::version
version_type version() const __TBB_override
Definition: market.h:261
tbb::internal::market::max_num_workers
static unsigned max_num_workers()
Definition: market.h:377
tbb::internal::market::skip_soft_limit_warning
static const unsigned skip_soft_limit_warning
The value indicating that the soft limit warning is unnecessary.
Definition: market.h:158
tbb::internal::market::set_active_num_workers
static void set_active_num_workers(unsigned w)
Set number of active workers.
Definition: market.cpp:235
tbb::internal::market::process
void process(job &j) __TBB_override
Definition: market.cpp:659
tbb::internal::market::must_join_workers
bool must_join_workers() const
Used when RML asks for join mode during workers termination.
Definition: market.h:311
tbb::internal::market::detach_arena
void detach_arena(arena &)
Removes the arena from the market's list.
Definition: market.cpp:322
tbb::internal::market::destroy
void destroy()
Destroys and deallocates market object created by market::create()
Definition: market.cpp:165
tbb::internal::market::update_workers_request
int update_workers_request()
Recalculates the number of workers requested from RML and updates the allotment.
Definition: market.cpp:217
tbb::internal::market::arena_in_need
arena * arena_in_need(arena *prev_arena)
Returns next arena that needs more workers, or NULL.
Definition: market.h:221
tbb::internal::market::release
bool release(bool is_public, bool blocking_terminate)
Decrements market's refcount and destroys it in the end.
Definition: market.cpp:175
tbb::internal::market::try_destroy_arena
void try_destroy_arena(arena *, uintptr_t aba_epoch)
Removes the arena from the market's list.
Definition: market.cpp:333
tbb::internal::market::min_stack_size
size_t min_stack_size() const __TBB_override
Definition: market.h:265
tbb::internal::market::insert_arena_into_list
void insert_arena_into_list(arena &a)
Definition: market.cpp:29
tbb::internal::market::ITT_DoUnsafeOneTimeInitialization
friend void ITT_DoUnsafeOneTimeInitialization()
tbb::internal::market::app_parallelism_limit
static unsigned app_parallelism_limit()
Reports active parallelism level according to user's settings.
Definition: tbb_main.cpp:512
tbb::internal::market::my_num_workers_requested
int my_num_workers_requested
Number of workers currently requested from RML.
Definition: market.h:81
tbb::internal::market::my_arenas_list_mutex
arenas_list_mutex_type my_arenas_list_mutex
Definition: market.h:67
tbb::internal::market::generic_scheduler
friend class generic_scheduler
Definition: market.h:46
tbb::internal::market::arenas_list_mutex_type
spin_rw_mutex arenas_list_mutex_type
Lightweight mutex guarding accounting operations with arenas list.
Definition: market.h:66
tbb::internal::market::remove_arena_from_list
void remove_arena_from_list(arena &a)
Definition: market.cpp:42
tbb::internal::market::create_arena
static arena * create_arena(int num_slots, int num_reserved_slots, size_t stack_size)
Creates an arena object.
Definition: market.cpp:308
tbb::internal::market::my_arenas
arena_list_type my_arenas
List of registered arenas.
Definition: market.h:135
tbb::internal::market::scheduler_list_type
intrusive_list< generic_scheduler > scheduler_list_type
Definition: market.h:55
tbb::internal::market::is_arena_in_list
bool is_arena_in_list(arena_list_type &arenas, arena *a)
Definition: market.cpp:424
tbb::internal::market::assert_market_valid
void assert_market_valid() const
Definition: market.h:241
tbb::internal::market::acknowledge_close_connection
void acknowledge_close_connection() __TBB_override
Definition: market.cpp:695
tbb::internal::market::my_workers_soft_limit_to_report
unsigned my_workers_soft_limit_to_report
Either workers soft limit to be reported via runtime_warning() or skip_soft_limit_warning.
Definition: market.h:161
tbb::internal::market::policy
policy_type policy() const __TBB_override
Definition: market.h:267
tbb::internal::market::my_total_demand
int my_total_demand
Number of workers that were requested by all arenas.
Definition: market.h:89
tbb::internal::market::my_ref_count
unsigned my_ref_count
Reference count controlling market object lifetime.
Definition: market.h:146
tbb::internal::market::worker_stack_size
size_t worker_stack_size() const
Returns the requested stack size of worker threads.
Definition: market.h:314
tbb::internal::market::market
market(unsigned workers_soft_limit, unsigned workers_hard_limit, size_t stack_size)
Constructor.
Definition: market.cpp:64
tbb::internal::market::my_stack_size
size_t my_stack_size
Stack size of worker threads.
Definition: market.h:152
tbb::internal::market::arena
friend class arena
Definition: market.h:47
tbb::internal::market::update_allotment
void update_allotment(unsigned effective_soft_limit)
Recalculates the number of workers assigned to each arena in the list.
Definition: market.h:214
tbb::internal::market::theMarketMutex
static global_market_mutex_type theMarketMutex
Mutex guarding creation/destruction of theMarket, insertions/deletions in my_arenas,...
Definition: market.h:63
tbb::internal::market::create_one_job
job * create_one_job() __TBB_override
Definition: market.cpp:699
tbb::internal::market::my_num_workers_soft_limit
unsigned my_num_workers_soft_limit
Current application-imposed limit on the number of workers (see set_active_num_workers())
Definition: market.h:78
tbb::internal::market::global_market_mutex_type
scheduler_mutex_type global_market_mutex_type
Definition: market.h:60
tbb::internal::market::my_server
rml::tbb_server * my_server
Pointer to the RML server object that services this TBB instance.
Definition: market.h:70
tbb::internal::market::my_next_arena
arena * my_next_arena
The first arena to be checked when idle worker seeks for an arena to enter.
Definition: market.h:139
tbb::internal::generic_scheduler
Work stealing task scheduler.
Definition: scheduler.h:140

Copyright © 2005-2020 Intel Corporation. All Rights Reserved.

Intel, Pentium, Intel Xeon, Itanium, Intel XScale and VTune are registered trademarks or trademarks of Intel Corporation or its subsidiaries in the United States and other countries.

* Other names and brands may be claimed as the property of others.