observer_proxy.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 "tbb/tbb_config.h"
#if __TBB_SCHEDULER_OBSERVER
#include "observer_proxy.h"
#include "tbb_main.h"
#include "governor.h"
#include "scheduler.h"
#include "arena.h"
namespace tbb {
namespace internal {
padded<observer_list> the_global_observer_list;
#if TBB_USE_ASSERT
static atomic<int> observer_proxy_count;
struct check_observer_proxy_count {
~check_observer_proxy_count() {
if( observer_proxy_count!=0 ) {
runtime_warning( "Leaked %ld observer_proxy objects\n", long(observer_proxy_count) );
}
}
};
static check_observer_proxy_count the_check_observer_proxy_count;
#endif /* TBB_USE_ASSERT */
#if __TBB_ARENA_OBSERVER
interface6::task_scheduler_observer* observer_proxy::get_v6_observer() {
if(my_version != 6) return NULL;
return static_cast<interface6::task_scheduler_observer*>(my_observer);
}
#endif
#if __TBB_ARENA_OBSERVER
bool observer_proxy::is_global() {
return !get_v6_observer() || get_v6_observer()->my_context_tag == interface6::task_scheduler_observer::global_tag;
}
#endif /* __TBB_ARENA_OBSERVER */
observer_proxy::observer_proxy( task_scheduler_observer_v3& tso )
: my_list(NULL), my_next(NULL), my_prev(NULL), my_observer(&tso)
{
#if TBB_USE_ASSERT
++observer_proxy_count;
#endif /* TBB_USE_ASSERT */
// 1 for observer
my_ref_count = 1;
my_version =
#if __TBB_ARENA_OBSERVER
load<relaxed>(my_observer->my_busy_count)
== interface6::task_scheduler_observer::v6_trait ? 6 :
#endif
0;
__TBB_ASSERT( my_version >= 6 || !load<relaxed>(my_observer->my_busy_count), NULL );
}
#if TBB_USE_ASSERT
observer_proxy::~observer_proxy () {
__TBB_ASSERT( !my_ref_count, "Attempt to destroy proxy still in use" );
poison_value(my_ref_count);
poison_pointer(my_prev);
poison_pointer(my_next);
--observer_proxy_count;
}
#endif /* TBB_USE_ASSERT */
template<memory_semantics M, class T, class V>
T atomic_fetch_and_store ( T* addr, const V& val ) {
return (T)atomic_traits<sizeof(T), M>::fetch_and_store( addr, (T)val );
}
void observer_list::clear () {
__TBB_ASSERT( this != &the_global_observer_list, "Method clear() cannot be used on the list of global observers" );
// Though the method will work fine for the empty list, we require the caller
// to check for the list emptiness before invoking it to avoid extra overhead.
__TBB_ASSERT( !empty(), NULL );
{
scoped_lock lock(mutex(), /*is_writer=*/true);
observer_proxy *next = my_head;
while ( observer_proxy *p = next ) {
__TBB_ASSERT( p->my_version >= 6, NULL );
next = p->my_next;
// Both proxy p and observer p->my_observer (if non-null) are guaranteed
// to be alive while the list is locked.
task_scheduler_observer_v3 *obs = p->my_observer;
// Make sure that possible concurrent observer destruction does not
// conflict with the proxy list cleanup.
if ( !obs || !(p = (observer_proxy*)__TBB_FetchAndStoreW(&obs->my_proxy, 0)) )
continue;
// accessing 'obs' after detaching of obs->my_proxy leads to the race with observer destruction
__TBB_ASSERT( !next || p == next->my_prev, NULL );
__TBB_ASSERT( is_alive(p->my_ref_count), "Observer's proxy died prematurely" );
__TBB_ASSERT( p->my_ref_count == 1, "Reference for observer is missing" );
#if TBB_USE_ASSERT
p->my_observer = NULL;
p->my_ref_count = 0;
#endif /* TBB_USE_ASSERT */
remove(p);
delete p;
}
}
while( my_head )
__TBB_Yield();
}
void observer_list::insert ( observer_proxy* p ) {
scoped_lock lock(mutex(), /*is_writer=*/true);
if ( my_head ) {
p->my_prev = my_tail;
my_tail->my_next = p;
}
else
my_head = p;
my_tail = p;
}
void observer_list::remove ( observer_proxy* p ) {
__TBB_ASSERT( my_head, "Attempt to remove an item from an empty list" );
__TBB_ASSERT( !my_tail->my_next, "Last item's my_next must be NULL" );
if( p == my_tail ) {
__TBB_ASSERT( !p->my_next, NULL );
my_tail = p->my_prev;
}
else {
__TBB_ASSERT( p->my_next, NULL );
p->my_next->my_prev = p->my_prev;
}
if ( p == my_head ) {
__TBB_ASSERT( !p->my_prev, NULL );
my_head = p->my_next;
}
else {
__TBB_ASSERT( p->my_prev, NULL );
p->my_prev->my_next = p->my_next;
}
__TBB_ASSERT( (my_head && my_tail) || (!my_head && !my_tail), NULL );
}
void observer_list::remove_ref( observer_proxy* p ) {
int r = p->my_ref_count;
__TBB_ASSERT( is_alive(r), NULL );
while(r>1) {
__TBB_ASSERT( r!=0, NULL );
int r_old = p->my_ref_count.compare_and_swap(r-1,r);
if( r_old==r ) {
// Successfully decremented count.
return;
}
r = r_old;
}
__TBB_ASSERT( r==1, NULL );
// Reference count might go to zero
{
// Use lock to avoid resurrection by a thread concurrently walking the list
observer_list::scoped_lock lock(mutex(), /*is_writer=*/true);
r = --p->my_ref_count;
if( !r )
remove(p);
}
__TBB_ASSERT( r || !p->my_ref_count, NULL );
if( !r )
delete p;
}
void observer_list::do_notify_entry_observers( observer_proxy*& last, bool worker ) {
// Pointer p marches though the list from last (exclusively) to the end.
observer_proxy *p = last, *prev = p;
for(;;) {
task_scheduler_observer_v3* tso=NULL;
// Hold lock on list only long enough to advance to the next proxy in the list.
{
scoped_lock lock(mutex(), /*is_writer=*/false);
do {
if( p ) {
// We were already processing the list.
if( observer_proxy* q = p->my_next ) {
if( p == prev )
remove_ref_fast(prev); // sets prev to NULL if successful
p = q;
}
else {
// Reached the end of the list.
if( p == prev ) {
// Keep the reference as we store the 'last' pointer in scheduler
__TBB_ASSERT(p->my_ref_count >= 1 + (p->my_observer?1:0), NULL);
} else {
// The last few proxies were empty
__TBB_ASSERT(p->my_ref_count, NULL);
++p->my_ref_count;
if( prev ) {
lock.release();
remove_ref(prev);
}
}
last = p;
return;
}
} else {
// Starting pass through the list
p = my_head;
if( !p )
return;
}
tso = p->my_observer;
} while( !tso );
++p->my_ref_count;
++tso->my_busy_count;
}
__TBB_ASSERT( !prev || p!=prev, NULL );
// Release the proxy pinned before p
if( prev )
remove_ref(prev);
// Do not hold any locks on the list while calling user's code.
// Do not intercept any exceptions that may escape the callback so that
// they are either handled by the TBB scheduler or passed to the debugger.
tso->on_scheduler_entry(worker);
__TBB_ASSERT(p->my_ref_count, NULL);
intptr_t bc = --tso->my_busy_count;
__TBB_ASSERT_EX( bc>=0, "my_busy_count underflowed" );
prev = p;
}
}
void observer_list::do_notify_exit_observers( observer_proxy* last, bool worker ) {
// Pointer p marches though the list from the beginning to last (inclusively).
observer_proxy *p = NULL, *prev = NULL;
for(;;) {
task_scheduler_observer_v3* tso=NULL;
// Hold lock on list only long enough to advance to the next proxy in the list.
{
scoped_lock lock(mutex(), /*is_writer=*/false);
do {
if( p ) {
// We were already processing the list.
if( p != last ) {
__TBB_ASSERT( p->my_next, "List items before 'last' must have valid my_next pointer" );
if( p == prev )
remove_ref_fast(prev); // sets prev to NULL if successful
p = p->my_next;
} else {
// remove the reference from the last item
remove_ref_fast(p);
if( p ) {
lock.release();
remove_ref(p);
}
return;
}
} else {
// Starting pass through the list
p = my_head;
__TBB_ASSERT( p, "Nonzero 'last' must guarantee that the global list is non-empty" );
}
tso = p->my_observer;
} while( !tso );
// The item is already refcounted
if ( p != last ) // the last is already referenced since entry notification
++p->my_ref_count;
++tso->my_busy_count;
}
__TBB_ASSERT( !prev || p!=prev, NULL );
if( prev )
remove_ref(prev);
// Do not hold any locks on the list while calling user's code.
// Do not intercept any exceptions that may escape the callback so that
// they are either handled by the TBB scheduler or passed to the debugger.
tso->on_scheduler_exit(worker);
__TBB_ASSERT(p->my_ref_count || p == last, NULL);
intptr_t bc = --tso->my_busy_count;
__TBB_ASSERT_EX( bc>=0, "my_busy_count underflowed" );
prev = p;
}
}
void task_scheduler_observer_v3::observe( bool enable ) {
if( enable ) {
if( !my_proxy ) {
my_proxy = new observer_proxy( *this );
my_busy_count = 0; // proxy stores versioning information, clear it
#if __TBB_ARENA_OBSERVER
if ( !my_proxy->is_global() ) {
// Local observer activation
generic_scheduler* s = governor::local_scheduler_if_initialized();
__TBB_ASSERT( my_proxy->get_v6_observer(), NULL );
intptr_t tag = my_proxy->get_v6_observer()->my_context_tag;
if( tag != interface6::task_scheduler_observer::implicit_tag ) { // explicit arena
task_arena *a = reinterpret_cast<task_arena*>(tag);
if ( a->my_arena==NULL ) // Avoid recursion during arena initialization
a->initialize();
my_proxy->my_list = &a->my_arena->my_observers;
} else {
if( !(s && s->my_arena) )
s = governor::init_scheduler( task_scheduler_init::automatic, 0, true );
__TBB_ASSERT( __TBB_InitOnce::initialization_done(), NULL );
__TBB_ASSERT( s && s->my_arena, NULL );
my_proxy->my_list = &s->my_arena->my_observers;
}
my_proxy->my_list->insert(my_proxy);
// Notify newly activated observer and other pending ones if it belongs to current arena
if(s && &s->my_arena->my_observers == my_proxy->my_list )
my_proxy->my_list->notify_entry_observers( s->my_last_local_observer, s->is_worker() );
} else
#endif /* __TBB_ARENA_OBSERVER */
{
// Obsolete. Global observer activation
if( !__TBB_InitOnce::initialization_done() )
DoOneTimeInitializations();
my_proxy->my_list = &the_global_observer_list;
my_proxy->my_list->insert(my_proxy);
if( generic_scheduler* s = governor::local_scheduler_if_initialized() ) {
// Notify newly created observer of its own thread.
// Any other pending observers are notified too.
the_global_observer_list.notify_entry_observers( s->my_last_global_observer, s->is_worker() );
}
}
}
} else {
// Make sure that possible concurrent proxy list cleanup does not conflict
// with the observer destruction here.
if ( observer_proxy* proxy = (observer_proxy*)__TBB_FetchAndStoreW(&my_proxy, 0) ) {
// List destruction should not touch this proxy after we've won the above interlocked exchange.
__TBB_ASSERT( proxy->my_observer == this, NULL );
__TBB_ASSERT( is_alive(proxy->my_ref_count), "Observer's proxy died prematurely" );
__TBB_ASSERT( proxy->my_ref_count >= 1, "reference for observer missing" );
observer_list &list = *proxy->my_list;
{
// Ensure that none of the list walkers relies on observer pointer validity
observer_list::scoped_lock lock(list.mutex(), /*is_writer=*/true);
proxy->my_observer = NULL;
// Proxy may still be held by other threads (to track the last notified observer)
if( !--proxy->my_ref_count ) {// nobody can increase it under exclusive lock
list.remove(proxy);
__TBB_ASSERT( !proxy->my_ref_count, NULL );
delete proxy;
}
}
while( my_busy_count ) // other threads are still accessing the callback
__TBB_Yield();
}
}
}
} // namespace internal
} // namespace tbb
#endif /* __TBB_SCHEDULER_OBSERVER */
__TBB_Yield
#define __TBB_Yield()
Definition: ibm_aix51.h:44
tbb_config.h
__TBB_ASSERT_EX
#define __TBB_ASSERT_EX(predicate, comment)
"Extended" version is useful to suppress warnings if a variable is only used with an assert
Definition: tbb_stddef.h:167
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
arena.h
governor.h
observer_proxy.h
scheduler.h
poison_value
#define poison_value(g)
Definition: scheduler_common.h:235
tbb_main.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
addr
void * addr
Definition: ittnotify_static.h:91
p
void const char const char int ITT_FORMAT __itt_group_sync p
Definition: ittnotify_static.h:91
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::runtime_warning
void __TBB_EXPORTED_FUNC runtime_warning(const char *format,...)
Report a runtime warning.
Definition: tbb_assert_impl.h:85
tbb::internal::poison_pointer
void poison_pointer(T *__TBB_atomic &)
Definition: tbb_stddef.h:305
tbb::internal::DoOneTimeInitializations
void DoOneTimeInitializations()
Performs thread-safe lazy one-time general TBB initialization.
Definition: tbb_main.cpp:215
tbb::internal::last
auto last(Container &c) -> decltype(begin(c))
Definition: _range_iterator.h:40
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::interface6::task_scheduler_observer::global_tag
@ global_tag
Definition: task_scheduler_observer.h:104

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.