| /* |
| Copyright 2005-2010 Intel Corporation. All Rights Reserved. |
| |
| This file is part of Threading Building Blocks. |
| |
| Threading Building Blocks is free software; you can redistribute it |
| and/or modify it under the terms of the GNU General Public License |
| version 2 as published by the Free Software Foundation. |
| |
| Threading Building Blocks is distributed in the hope that it will be |
| useful, but WITHOUT ANY WARRANTY; without even the implied warranty |
| of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| GNU General Public License for more details. |
| |
| You should have received a copy of the GNU General Public License |
| along with Threading Building Blocks; if not, write to the Free Software |
| Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA |
| |
| As a special exception, you may use this file as part of a free software |
| library without restriction. Specifically, if other files instantiate |
| templates or use macros or inline functions from this file, or you compile |
| this file and link it with other files to produce an executable, this |
| file does not by itself cause the resulting executable to be covered by |
| the GNU General Public License. This exception does not however |
| invalidate any other reasons why the executable file might be covered by |
| the GNU General Public License. |
| */ |
| |
| #include "tbb/tbb_stddef.h" |
| |
| #if __TBB_ARENA_PER_MASTER |
| |
| #include "market.h" |
| #include "tbb_main.h" |
| #include "governor.h" |
| #include "scheduler.h" |
| |
| namespace tbb { |
| namespace internal { |
| |
| //------------------------------------------------------------------------ |
| // market |
| //------------------------------------------------------------------------ |
| |
| market::market ( unsigned max_num_workers, size_t stack_size ) |
| : my_ref_count(1) |
| , my_stack_size(stack_size) |
| , my_max_num_workers(max_num_workers) |
| { |
| my_next_arena = my_arenas.begin(); |
| |
| // Once created RML server will start initializing workers that will need |
| // global market instance to get worker stack size |
| my_server = governor::create_rml_server( *this ); |
| __TBB_ASSERT( my_server, "Failed to create RML server" ); |
| } |
| |
| |
| market& market::global_market ( unsigned max_num_workers, size_t stack_size ) { |
| global_market_mutex_type::scoped_lock lock( theMarketMutex ); |
| market *m = theMarket; |
| if ( m ) { |
| ++m->my_ref_count; |
| if ( m->my_stack_size < stack_size ) |
| runtime_warning( "Newer master request for larger stack cannot be satisfied\n" ); |
| } |
| else { |
| max_num_workers = max( governor::default_num_threads() - 1, max_num_workers ); |
| // at least 1 worker is required to support starvation resistant tasks |
| if( max_num_workers==0 ) max_num_workers = 1; |
| // Create the global market instance |
| size_t size = sizeof(market); |
| #if __TBB_TASK_GROUP_CONTEXT |
| __TBB_ASSERT( __TBB_offsetof(market, my_workers) + sizeof(generic_scheduler*) == sizeof(market), |
| "my_workers must be the last data field of the market class"); |
| size += sizeof(generic_scheduler*) * (max_num_workers - 1); |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| __TBB_InitOnce::add_ref(); |
| void* storage = NFS_Allocate(size, 1, NULL); |
| memset( storage, 0, size ); |
| // Initialize and publish global market |
| m = new (storage) market( max_num_workers, stack_size ); |
| theMarket = m; |
| } |
| return *m; |
| } |
| |
| void market::destroy () { |
| #if __TBB_COUNT_TASK_NODES |
| if ( my_task_node_count ) |
| runtime_warning( "Leaked %ld task objects\n", (intptr_t)my_task_node_count ); |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| this->~market(); |
| NFS_Free( this ); |
| __TBB_InitOnce::remove_ref(); |
| } |
| |
| void market::release () { |
| __TBB_ASSERT( theMarket == this, "Global market instance was destroyed prematurely?" ); |
| bool do_release = false; |
| { |
| global_market_mutex_type::scoped_lock lock(theMarketMutex); |
| if ( --my_ref_count == 0 ) { |
| do_release = true; |
| theMarket = NULL; |
| } |
| } |
| if( do_release ) |
| my_server->request_close_connection(); |
| } |
| |
| arena& market::create_arena ( unsigned max_num_workers, size_t stack_size ) { |
| market &m = global_market( max_num_workers, stack_size ); // increases market's ref count |
| arena& a = arena::allocate_arena( m, min(max_num_workers, m.my_max_num_workers) ); |
| // Add newly created arena into the existing market's list. |
| spin_mutex::scoped_lock lock(m.my_arenas_list_mutex); |
| m.my_arenas.push_front( a ); |
| if ( m.my_arenas.size() == 1 ) |
| m.my_next_arena = m.my_arenas.begin(); |
| return a; |
| } |
| |
| void market::detach_arena ( arena& a ) { |
| __TBB_ASSERT( theMarket == this, "Global market instance was destroyed prematurely?" ); |
| spin_mutex::scoped_lock lock(my_arenas_list_mutex); |
| __TBB_ASSERT( my_next_arena != my_arenas.end(), NULL ); |
| if ( &*my_next_arena == &a ) |
| if ( ++my_next_arena == my_arenas.end() && my_arenas.size() > 1 ) |
| my_next_arena = my_arenas.begin(); |
| my_arenas.remove( a ); |
| } |
| |
| arena* market::arena_in_need () { |
| spin_mutex::scoped_lock lock(my_arenas_list_mutex); |
| if ( my_arenas.empty() ) |
| return NULL; |
| __TBB_ASSERT( my_next_arena != my_arenas.end(), NULL ); |
| arena_list_type::iterator it = my_next_arena; |
| do { |
| arena& a = *it; |
| if ( ++it == my_arenas.end() ) |
| it = my_arenas.begin(); |
| if ( a.num_workers_active() < a.my_num_workers_allotted ) { |
| ++a.my_num_threads_active; |
| my_next_arena = it; |
| return &a; |
| } |
| } while ( it != my_next_arena ); |
| return NULL; |
| } |
| |
| void market::update_allotment ( int max_workers ) { |
| unsigned carry = 0; |
| spin_mutex::scoped_lock lock(my_arenas_list_mutex); |
| arena_list_type::iterator it = my_arenas.begin(); |
| int total_demand = my_total_demand; |
| max_workers = min(max_workers, total_demand); |
| if ( total_demand > 0 ) { |
| for ( ; it != my_arenas.end(); ++it ) { |
| arena& a = *it; |
| int tmp = a.my_num_workers_requested * max_workers + carry; |
| int allotted = tmp / total_demand; |
| carry = tmp % total_demand; |
| a.my_num_workers_allotted = min( allotted, (int)a.my_max_num_workers ); |
| } |
| } |
| else { |
| for ( ; it != my_arenas.end(); ++it ) { |
| it->my_num_workers_allotted = 0; |
| } |
| } |
| } |
| |
| /** The balancing algorithm may be liable to data races. However the aberrations |
| caused by the races are not fatal and generally only temporarily affect fairness |
| of the workers distribution among arenas. **/ |
| void market::adjust_demand ( arena& a, int delta ) { |
| __TBB_ASSERT( theMarket, "market instance was destroyed prematurely?" ); |
| a.my_num_workers_requested += delta; |
| my_total_demand += delta; |
| update_allotment( my_max_num_workers ); |
| // Must be called outside of any locks |
| my_server->adjust_job_count_estimate( delta ); |
| } |
| |
| void market::process( job& j ) { |
| generic_scheduler& s = static_cast<generic_scheduler&>(j); |
| while ( arena *a = arena_in_need() ) |
| a->process(s); |
| } |
| |
| void market::cleanup( job& j ) { |
| __TBB_ASSERT( theMarket != this, NULL ); |
| generic_scheduler& s = static_cast<generic_scheduler&>(j); |
| generic_scheduler* mine = governor::local_scheduler_if_initialized(); |
| __TBB_ASSERT( !mine || mine->arena_index!=0, NULL ); |
| if( mine!=&s ) { |
| governor::assume_scheduler( &s ); |
| generic_scheduler::cleanup_worker( &s, mine!=NULL ); |
| governor::assume_scheduler( mine ); |
| } else { |
| generic_scheduler::cleanup_worker( &s, true ); |
| } |
| } |
| |
| void market::acknowledge_close_connection() { |
| destroy(); |
| } |
| |
| ::rml::job* market::create_one_job() { |
| unsigned index = ++my_num_workers; |
| __TBB_ASSERT( index > 0, NULL ); |
| // index serves as a hint decreasing conflicts between workers when they migrate between arenas |
| generic_scheduler* s = generic_scheduler::create_worker( *this, index ); |
| #if __TBB_TASK_GROUP_CONTEXT |
| __TBB_ASSERT( !my_workers[index - 1], NULL ); |
| my_workers[index - 1] = s; |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| governor::sign_on(s); |
| return s; |
| } |
| |
| #if __TBB_TASK_GROUP_CONTEXT |
| /** Propagates cancellation down the tree of dependent contexts by walking each |
| thread's local list of contexts **/ |
| void market::propagate_cancellation ( task_group_context& ctx ) { |
| __TBB_ASSERT ( ctx.my_cancellation_requested, "No cancellation request in the context" ); |
| // The whole propagation algorithm is under the lock in order to ensure correctness |
| // in case of parallel cancellations at the different levels of the context tree. |
| // See the note 1 at the bottom of this file. |
| global_market_mutex_type::scoped_lock lock(theMarketMutex); |
| // Advance global cancellation epoch |
| uintptr_t global_epoch = __TBB_FetchAndAddWrelease(&global_cancel_count, 1); |
| // Propagate to all workers and masters and sync up their local epochs with the global one |
| unsigned num_workers = my_num_workers; |
| for ( unsigned i = 0; i < num_workers; ++i ) { |
| generic_scheduler *s = my_workers[i]; |
| // If the worker is only about to be registered, skip it. |
| if ( s ) { |
| s->propagate_cancellation(); |
| s->local_cancel_count = global_epoch; |
| } |
| } |
| arena_list_type::iterator it = my_arenas.begin(); |
| for ( ; it != my_arenas.end(); ++it ) { |
| generic_scheduler *s = it->slot[0].my_scheduler; |
| // If the master is under construction, skip it. |
| if ( s ) { |
| s->propagate_cancellation(); |
| s->local_cancel_count = global_epoch; |
| } |
| } |
| } |
| #endif /* __TBB_TASK_GROUP_CONTEXT */ |
| |
| #if __TBB_COUNT_TASK_NODES |
| intptr_t market::workers_task_node_count() { |
| intptr_t result = 0; |
| spin_mutex::scoped_lock lock(my_arenas_list_mutex); |
| for ( arena_list_type::iterator it = my_arenas.begin(); it != my_arenas.end(); ++it ) |
| result += it->workers_task_node_count(); |
| return result; |
| } |
| #endif /* __TBB_COUNT_TASK_NODES */ |
| |
| } // namespace internal |
| } // namespace tbb |
| |
| #endif /* __TBB_ARENA_PER_MASTER */ |
| |
| /* |
| Notes: |
| |
| 1. Consider parallel cancellations at the different levels of the context tree: |
| |
| Ctx1 <- Cancelled by Thread1 |- Thread2 started processing |
| | | |
| Ctx2 |- Thread1 started processing |
| | T1 |- Thread2 finishes and syncs up local counters |
| Ctx3 <- Cancelled by Thread2 | |
| | |- Ctx5 is bound to Ctx2 |
| Ctx4 | |
| T2 |- Thread1 reaches Ctx2 |
| |
| Thread-propagator of each cancellation increments global counter. However the thread |
| propagating the cancellation from the outermost context (Thread1) may be the last |
| to finish. Which means that the local counters may be synchronized earlier (by Thread2, |
| at Time1) than it propagated cancellation into Ctx2 (at time Time2). If a new context |
| (Ctx5) is created and bound to Ctx2 between Time1 and Time2, checking its parent only |
| (Ctx2) may result in cancellation request being lost. |
| |
| This issue is solved by doing the whole propagation under the lock (the_scheduler_list_mutex). |
| |
| If we need more concurrency while processing parallel cancellations, we could try |
| the following modification of the propagation algorithm: |
| |
| advance global counter and remember it |
| for each thread: |
| scan thread's list of contexts |
| for each thread: |
| sync up its local counter only if the global counter has not been changed |
| |
| However this version of the algorithm requires more analysis and verification. |
| */ |
