private_server.cpp
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.
*/
#include "../rml/include/rml_tbb.h"
#include "../rml/server/thread_monitor.h"
#include "tbb/atomic.h"
#include "tbb/cache_aligned_allocator.h"
#include "scheduler_common.h"
#include "governor.h"
#include "tbb_misc.h"
using rml::internal::thread_monitor;
namespace tbb {
namespace internal {
namespace rml {
typedef thread_monitor::handle_type thread_handle;
class private_server;
class private_worker: no_copy {
private:
enum state_t {
st_init,
st_starting,
st_normal,
st_quit
};
atomic<state_t> my_state;
private_server& my_server;
tbb_client& my_client;
const size_t my_index;
thread_monitor my_thread_monitor;
thread_handle my_handle;
private_worker* my_next;
friend class private_server;
void run();
void wake_or_launch();
void start_shutdown();
static __RML_DECL_THREAD_ROUTINE thread_routine( void* arg );
static void release_handle(thread_handle my_handle, bool join);
protected:
private_worker( private_server& server, tbb_client& client, const size_t i ) :
my_server(server), my_client(client), my_index(i),
my_thread_monitor(), my_handle(), my_next()
{
my_state = st_init;
}
};
static const size_t cache_line_size = tbb::internal::NFS_MaxLineSize;
#if _MSC_VER && !defined(__INTEL_COMPILER)
// Suppress overzealous compiler warnings about uninstantiable class
#pragma warning(push)
#pragma warning(disable:4510 4610)
#endif
class padded_private_worker: public private_worker {
char pad[cache_line_size - sizeof(private_worker)%cache_line_size];
public:
padded_private_worker( private_server& server, tbb_client& client, const size_t i )
: private_worker(server,client,i) { suppress_unused_warning(pad); }
};
#if _MSC_VER && !defined(__INTEL_COMPILER)
#pragma warning(pop)
#endif
class private_server: public tbb_server, no_copy {
private:
tbb_client& my_client;
const tbb_client::size_type my_n_thread;
const size_t my_stack_size;
atomic<int> my_slack;
atomic<int> my_ref_count;
padded_private_worker* my_thread_array;
tbb::atomic<private_worker*> my_asleep_list_root;
typedef scheduler_mutex_type asleep_list_mutex_type;
asleep_list_mutex_type my_asleep_list_mutex;
#if TBB_USE_ASSERT
atomic<int> my_net_slack_requests;
#endif /* TBB_USE_ASSERT */
void propagate_chain_reaction() {
// First test of a double-check idiom. Second test is inside wake_some(0).
if( my_asleep_list_root )
wake_some(0);
}
bool try_insert_in_asleep_list( private_worker& t );
void wake_some( int additional_slack );
virtual ~private_server();
void remove_server_ref() {
if( --my_ref_count==0 ) {
my_client.acknowledge_close_connection();
this->~private_server();
tbb::cache_aligned_allocator<private_server>().deallocate( this, 1 );
}
}
friend class private_worker;
public:
private_server( tbb_client& client );
version_type version() const __TBB_override {
return 0;
}
void request_close_connection( bool /*exiting*/ ) __TBB_override {
for( size_t i=0; i<my_n_thread; ++i )
my_thread_array[i].start_shutdown();
remove_server_ref();
}
void yield() __TBB_override {__TBB_Yield();}
void independent_thread_number_changed( int ) __TBB_override {__TBB_ASSERT(false,NULL);}
unsigned default_concurrency() const __TBB_override { return governor::default_num_threads() - 1; }
void adjust_job_count_estimate( int delta ) __TBB_override;
#if _WIN32||_WIN64
void register_master ( ::rml::server::execution_resource_t& ) __TBB_override {}
void unregister_master ( ::rml::server::execution_resource_t ) __TBB_override {}
#endif /* _WIN32||_WIN64 */
};
//------------------------------------------------------------------------
// Methods of private_worker
//------------------------------------------------------------------------
#if _MSC_VER && !defined(__INTEL_COMPILER)
// Suppress overzealous compiler warnings about an initialized variable 'sink_for_alloca' not referenced
#pragma warning(push)
#pragma warning(disable:4189)
#endif
#if __MINGW32__ && __GNUC__==4 &&__GNUC_MINOR__>=2 && !__MINGW64__
// ensure that stack is properly aligned for TBB threads
__attribute__((force_align_arg_pointer))
#endif
__RML_DECL_THREAD_ROUTINE private_worker::thread_routine( void* arg ) {
private_worker* self = static_cast<private_worker*>(arg);
AVOID_64K_ALIASING( self->my_index );
self->run();
return 0;
}
#if _MSC_VER && !defined(__INTEL_COMPILER)
#pragma warning(pop)
#endif
void private_worker::release_handle(thread_handle handle, bool join) {
if (join)
thread_monitor::join(handle);
else
thread_monitor::detach_thread(handle);
}
void private_worker::start_shutdown() {
state_t s;
do {
s = my_state;
__TBB_ASSERT( s!=st_quit, NULL );
} while( my_state.compare_and_swap( st_quit, s )!=s );
if( s==st_normal || s==st_starting ) {
// May have invalidated invariant for sleeping, so wake up the thread.
// Note that the notify() here occurs without maintaining invariants for my_slack.
// It does not matter, because my_state==st_quit overrides checking of my_slack.
my_thread_monitor.notify();
// Do not need release handle in st_init state,
// because in this case the thread wasn't started yet.
// For st_starting release is done at launch site.
if (s==st_normal)
release_handle(my_handle, governor::does_client_join_workers(my_client));
} else if( s==st_init ) {
// Perform action that otherwise would be performed by associated thread when it quits.
my_server.remove_server_ref();
}
}
void private_worker::run() {
my_server.propagate_chain_reaction();
// Transiting to st_normal here would require setting my_handle,
// which would create race with the launching thread and
// complications in handle management on Windows.
::rml::job& j = *my_client.create_one_job();
while( my_state!=st_quit ) {
if( my_server.my_slack>=0 ) {
my_client.process(j);
} else {
thread_monitor::cookie c;
// Prepare to wait
my_thread_monitor.prepare_wait(c);
// Check/set the invariant for sleeping
if( my_state!=st_quit && my_server.try_insert_in_asleep_list(*this) ) {
my_thread_monitor.commit_wait(c);
__TBB_ASSERT( my_state==st_quit || !my_next, "Thread monitor missed a spurious wakeup?" );
my_server.propagate_chain_reaction();
} else {
// Invariant broken
my_thread_monitor.cancel_wait();
}
}
}
my_client.cleanup(j);
++my_server.my_slack;
my_server.remove_server_ref();
}
inline void private_worker::wake_or_launch() {
if( my_state==st_init && my_state.compare_and_swap( st_starting, st_init )==st_init ) {
// after this point, remove_server_ref() must be done by created thread
#if USE_WINTHREAD
my_handle = thread_monitor::launch( thread_routine, this, my_server.my_stack_size, &this->my_index );
#elif USE_PTHREAD
{
affinity_helper fpa;
fpa.protect_affinity_mask( /*restore_process_mask=*/true );
my_handle = thread_monitor::launch( thread_routine, this, my_server.my_stack_size );
// Implicit destruction of fpa resets original affinity mask.
}
#endif /* USE_PTHREAD */
state_t s = my_state.compare_and_swap( st_normal, st_starting );
if (st_starting != s) {
// Do shutdown during startup. my_handle can't be released
// by start_shutdown, because my_handle value might be not set yet
// at time of transition from st_starting to st_quit.
__TBB_ASSERT( s==st_quit, NULL );
release_handle(my_handle, governor::does_client_join_workers(my_client));
}
}
else {
__TBB_ASSERT( !my_next, "Should not wake a thread while it's still in asleep list" );
my_thread_monitor.notify();
}
}
//------------------------------------------------------------------------
// Methods of private_server
//------------------------------------------------------------------------
private_server::private_server( tbb_client& client ) :
my_client(client),
my_n_thread(client.max_job_count()),
my_stack_size(client.min_stack_size()),
my_thread_array(NULL)
{
my_ref_count = my_n_thread+1;
my_slack = 0;
#if TBB_USE_ASSERT
my_net_slack_requests = 0;
#endif /* TBB_USE_ASSERT */
my_asleep_list_root = NULL;
my_thread_array = tbb::cache_aligned_allocator<padded_private_worker>().allocate( my_n_thread );
for( size_t i=0; i<my_n_thread; ++i ) {
private_worker* t = new( &my_thread_array[i] ) padded_private_worker( *this, client, i );
t->my_next = my_asleep_list_root;
my_asleep_list_root = t;
}
}
private_server::~private_server() {
__TBB_ASSERT( my_net_slack_requests==0, NULL );
for( size_t i=my_n_thread; i--; )
my_thread_array[i].~padded_private_worker();
tbb::cache_aligned_allocator<padded_private_worker>().deallocate( my_thread_array, my_n_thread );
tbb::internal::poison_pointer( my_thread_array );
}
inline bool private_server::try_insert_in_asleep_list( private_worker& t ) {
asleep_list_mutex_type::scoped_lock lock;
if( !lock.try_acquire(my_asleep_list_mutex) )
return false;
// Contribute to slack under lock so that if another takes that unit of slack,
// it sees us sleeping on the list and wakes us up.
int k = ++my_slack;
if( k<=0 ) {
t.my_next = my_asleep_list_root;
my_asleep_list_root = &t;
return true;
} else {
--my_slack;
return false;
}
}
void private_server::wake_some( int additional_slack ) {
__TBB_ASSERT( additional_slack>=0, NULL );
private_worker* wakee[2];
private_worker**w = wakee;
{
asleep_list_mutex_type::scoped_lock lock(my_asleep_list_mutex);
while( my_asleep_list_root && w<wakee+2 ) {
if( additional_slack>0 ) {
if (additional_slack+my_slack<=0) // additional demand does not exceed surplus supply
break;
--additional_slack;
} else {
// Chain reaction; Try to claim unit of slack
int old;
do {
old = my_slack;
if( old<=0 ) goto done;
} while( my_slack.compare_and_swap(old-1,old)!=old );
}
// Pop sleeping worker to combine with claimed unit of slack
my_asleep_list_root = (*w++ = my_asleep_list_root)->my_next;
}
if( additional_slack ) {
// Contribute our unused slack to my_slack.
my_slack += additional_slack;
}
}
done:
while( w>wakee ) {
private_worker* ww = *--w;
ww->my_next = NULL;
ww->wake_or_launch();
}
}
void private_server::adjust_job_count_estimate( int delta ) {
#if TBB_USE_ASSERT
my_net_slack_requests+=delta;
#endif /* TBB_USE_ASSERT */
if( delta<0 ) {
my_slack+=delta;
} else if( delta>0 ) {
wake_some( delta );
}
}
tbb_server* make_private_server( tbb_client& client ) {
return new( tbb::cache_aligned_allocator<private_server>().allocate(1) ) private_server(client);
}
} // namespace rml
} // namespace internal
} // namespace tbb
atomic.h
cache_aligned_allocator.h
__TBB_Yield
#define __TBB_Yield()
Definition: ibm_aix51.h:44
__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
governor.h
scheduler_common.h
tbb_misc.h
s
void const char const char int ITT_FORMAT __itt_group_sync s
Definition: ittnotify_static.h:91
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::internal::NFS_MaxLineSize
const size_t NFS_MaxLineSize
Compile-time constant that is upper bound on cache line/sector size.
Definition: tbb_stddef.h:216
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::suppress_unused_warning
void suppress_unused_warning(const T1 &)
Utility template function to prevent "unused" warnings by various compilers.
Definition: tbb_stddef.h:398
tbb::internal::poison_pointer
void poison_pointer(T *__TBB_atomic &)
Definition: tbb_stddef.h:305
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
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::internal::rml::thread_handle
thread_monitor::handle_type thread_handle
Definition: private_server.cpp:31
tbb::internal::rml::make_private_server
tbb_server * make_private_server(tbb_client &client)
Factory method called from task.cpp to create a private_server.
Definition: private_server.cpp:411
tbb::internal::rml::cache_line_size
static const size_t cache_line_size
Definition: private_server.cpp:100
tbb::cache_aligned_allocator
Meets "allocator" requirements of ISO C++ Standard, Section 20.1.5.
Definition: cache_aligned_allocator.h:60
tbb::cache_aligned_allocator::deallocate
void deallocate(pointer p, size_type)
Free block of memory that starts on a cache line.
Definition: cache_aligned_allocator.h:86
tbb::cache_aligned_allocator::allocate
pointer allocate(size_type n, const void *hint=0)
Allocate space for n objects, starting on a cache/sector line.
Definition: cache_aligned_allocator.h:80
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::governor::does_client_join_workers
static bool does_client_join_workers(const tbb::internal::rml::tbb_client &client)
Definition: market.cpp:304
tbb::internal::governor::default_num_threads
static unsigned default_num_threads()
Definition: governor.h:84
tbb::internal::rml::private_worker
Definition: private_server.cpp:35
tbb::internal::rml::private_worker::my_server
private_server & my_server
Associated server.
Definition: private_server.cpp:57
tbb::internal::rml::private_worker::my_handle
thread_handle my_handle
Handle of the OS thread associated with this worker.
Definition: private_server.cpp:71
tbb::internal::rml::private_worker::start_shutdown
void start_shutdown()
Called by a thread (usually not the associated thread) to commence termination.
Definition: private_server.cpp:233
tbb::internal::rml::private_worker::my_next
private_worker * my_next
Link for list of workers that are sleeping or have no associated thread.
Definition: private_server.cpp:74
tbb::internal::rml::private_worker::state_t
state_t
State in finite-state machine that controls the worker.
Definition: private_server.cpp:44
tbb::internal::rml::private_worker::st_init
@ st_init
*this is initialized
Definition: private_server.cpp:46
tbb::internal::rml::private_worker::st_normal
@ st_normal
Associated thread is doing normal life sequence.
Definition: private_server.cpp:50
tbb::internal::rml::private_worker::st_quit
@ st_quit
Associated thread has ended normal life sequence and promises to never touch *this again.
Definition: private_server.cpp:52
tbb::internal::rml::private_worker::st_starting
@ st_starting
*this has associated thread that is starting up.
Definition: private_server.cpp:48
tbb::internal::rml::private_worker::my_thread_monitor
thread_monitor my_thread_monitor
Monitor for sleeping when there is no work to do.
Definition: private_server.cpp:68
tbb::internal::rml::private_worker::release_handle
static void release_handle(thread_handle my_handle, bool join)
Definition: private_server.cpp:226
tbb::internal::rml::private_worker::thread_routine
static __RML_DECL_THREAD_ROUTINE thread_routine(void *arg)
Definition: private_server.cpp:216
tbb::internal::rml::private_worker::private_worker
private_worker(private_server &server, tbb_client &client, const size_t i)
Definition: private_server.cpp:92
tbb::internal::rml::private_worker::my_state
atomic< state_t > my_state
Definition: private_server.cpp:54
tbb::internal::rml::private_worker::wake_or_launch
void wake_or_launch()
Wake up associated thread (or launch a thread if there is none)
Definition: private_server.cpp:288
tbb::internal::rml::private_worker::my_client
tbb_client & my_client
Associated client.
Definition: private_server.cpp:60
tbb::internal::rml::private_worker::my_index
const size_t my_index
index used for avoiding the 64K aliasing problem
Definition: private_server.cpp:63
tbb::internal::rml::private_worker::run
void run()
Actions executed by the associated thread.
Definition: private_server.cpp:256
tbb::internal::rml::padded_private_worker
Definition: private_server.cpp:108
tbb::internal::rml::padded_private_worker::pad
char pad[cache_line_size - sizeof(private_worker)%cache_line_size]
Definition: private_server.cpp:109
tbb::internal::rml::padded_private_worker::padded_private_worker
padded_private_worker(private_server &server, tbb_client &client, const size_t i)
Definition: private_server.cpp:111
tbb::internal::rml::private_server
Definition: private_server.cpp:118
tbb::internal::rml::private_server::version
version_type version() const __TBB_override
Definition: private_server.cpp:180
tbb::internal::rml::private_server::my_stack_size
const size_t my_stack_size
Stack size for each thread. *‍/.
Definition: private_server.cpp:126
tbb::internal::rml::private_server::my_n_thread
const tbb_client::size_type my_n_thread
Maximum number of threads to be created.
Definition: private_server.cpp:123
tbb::internal::rml::private_server::propagate_chain_reaction
void propagate_chain_reaction()
Wake up to two sleeping workers, if there are any sleeping.
Definition: private_server.cpp:154
tbb::internal::rml::private_server::wake_some
void wake_some(int additional_slack)
Equivalent of adding additional_slack to my_slack and waking up to 2 threads if my_slack permits.
Definition: private_server.cpp:364
tbb::internal::rml::private_server::independent_thread_number_changed
void independent_thread_number_changed(int) __TBB_override
Definition: private_server.cpp:192
tbb::internal::rml::private_server::yield
void yield() __TBB_override
Definition: private_server.cpp:190
tbb::internal::rml::private_server::~private_server
virtual ~private_server()
Definition: private_server.cpp:339
tbb::internal::rml::private_server::default_concurrency
unsigned default_concurrency() const __TBB_override
Definition: private_server.cpp:194
tbb::internal::rml::private_server::asleep_list_mutex_type
scheduler_mutex_type asleep_list_mutex_type
Protects my_asleep_list_root.
Definition: private_server.cpp:144
tbb::internal::rml::private_server::try_insert_in_asleep_list
bool try_insert_in_asleep_list(private_worker &t)
Try to add t to list of sleeping workers.
Definition: private_server.cpp:347
tbb::internal::rml::private_server::private_server
private_server(tbb_client &client)
Definition: private_server.cpp:319
tbb::internal::rml::private_server::my_ref_count
atomic< int > my_ref_count
Counter used to determine when to delete this.
Definition: private_server.cpp:136
tbb::internal::rml::private_server::my_slack
atomic< int > my_slack
Number of jobs that could use their associated thread minus number of active threads.
Definition: private_server.cpp:133
tbb::internal::rml::private_server::request_close_connection
void request_close_connection(bool) __TBB_override
Definition: private_server.cpp:184
tbb::internal::rml::private_server::my_asleep_list_root
tbb::atomic< private_worker * > my_asleep_list_root
List of workers that are asleep or committed to sleeping until notified by another thread.
Definition: private_server.cpp:141
tbb::internal::rml::private_server::my_asleep_list_mutex
asleep_list_mutex_type my_asleep_list_mutex
Definition: private_server.cpp:145
tbb::internal::rml::private_server::my_thread_array
padded_private_worker * my_thread_array
Definition: private_server.cpp:138
tbb::internal::rml::private_server::adjust_job_count_estimate
void adjust_job_count_estimate(int delta) __TBB_override
Definition: private_server.cpp:399
tbb::internal::rml::private_server::remove_server_ref
void remove_server_ref()
Definition: private_server.cpp:168
tbb::internal::rml::private_server::my_client
tbb_client & my_client
Definition: private_server.cpp:120
tbb::internal::affinity_helper
Definition: tbb_misc.h:261
tbb::internal::affinity_helper::protect_affinity_mask
void protect_affinity_mask(bool)
Definition: tbb_misc.h:263

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.