concurrent_monitor.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_concurrent_monitor_H
#define __TBB_concurrent_monitor_H
#include "tbb/tbb_stddef.h"
#include "tbb/atomic.h"
#include "tbb/spin_mutex.h"
#include "tbb/tbb_exception.h"
#include "tbb/aligned_space.h"
#include "semaphore.h"
namespace tbb {
namespace internal {
class circular_doubly_linked_list_with_sentinel : no_copy {
public:
struct node_t {
node_t* next;
node_t* prev;
explicit node_t() : next((node_t*)(uintptr_t)0xcdcdcdcd), prev((node_t*)(uintptr_t)0xcdcdcdcd) {}
};
// ctor
circular_doubly_linked_list_with_sentinel() {clear();}
// dtor
~circular_doubly_linked_list_with_sentinel() {__TBB_ASSERT( head.next==&head && head.prev==&head, "the list is not empty" );}
inline size_t size() const {return count;}
inline bool empty() const {return size()==0;}
inline node_t* front() const {return head.next;}
inline node_t* last() const {return head.prev;}
inline node_t* begin() const {return front();}
inline const node_t* end() const {return &head;}
inline void add( node_t* n ) {
__TBB_store_relaxed(count, __TBB_load_relaxed(count) + 1);
n->prev = head.prev;
n->next = &head;
head.prev->next = n;
head.prev = n;
}
inline void remove( node_t& n ) {
__TBB_ASSERT( count > 0, "attempt to remove an item from an empty list" );
__TBB_store_relaxed(count, __TBB_load_relaxed(count) - 1);
n.prev->next = n.next;
n.next->prev = n.prev;
}
inline void flush_to( circular_doubly_linked_list_with_sentinel& lst ) {
if( const size_t l_count = __TBB_load_relaxed(count) ) {
__TBB_store_relaxed(lst.count, l_count);
lst.head.next = head.next;
lst.head.prev = head.prev;
head.next->prev = &lst.head;
head.prev->next = &lst.head;
clear();
}
}
void clear() {head.next = head.prev = &head; __TBB_store_relaxed(count, 0);}
private:
__TBB_atomic size_t count;
node_t head;
};
typedef circular_doubly_linked_list_with_sentinel waitset_t;
typedef circular_doubly_linked_list_with_sentinel::node_t waitset_node_t;
class concurrent_monitor : no_copy {
public:
class thread_context : waitset_node_t, no_copy {
friend class concurrent_monitor;
public:
thread_context() : skipped_wakeup(false), aborted(false), ready(false), context(0) {
epoch = 0;
in_waitset = false;
}
~thread_context() {
if (ready) {
if( skipped_wakeup ) semaphore().P();
semaphore().~binary_semaphore();
}
}
binary_semaphore& semaphore() { return *sema.begin(); }
private:
// Inlining of the method is undesirable, due to extra instructions for
// exception support added at caller side.
__TBB_NOINLINE( void init() );
tbb::aligned_space<binary_semaphore> sema;
__TBB_atomic unsigned epoch;
tbb::atomic<bool> in_waitset;
bool skipped_wakeup;
bool aborted;
bool ready;
uintptr_t context;
};
concurrent_monitor() {__TBB_store_relaxed(epoch, 0);}
~concurrent_monitor() ;
void prepare_wait( thread_context& thr, uintptr_t ctx = 0 );
inline bool commit_wait( thread_context& thr ) {
const bool do_it = thr.epoch == __TBB_load_relaxed(epoch);
// this check is just an optimization
if( do_it ) {
__TBB_ASSERT( thr.ready, "use of commit_wait() without prior prepare_wait()");
thr.semaphore().P();
__TBB_ASSERT( !thr.in_waitset, "still in the queue?" );
if( thr.aborted )
throw_exception( eid_user_abort );
} else {
cancel_wait( thr );
}
return do_it;
}
void cancel_wait( thread_context& thr );
template<typename WaitUntil, typename Context>
void wait( WaitUntil until, Context on );
void notify_one() {atomic_fence(); notify_one_relaxed();}
void notify_one_relaxed();
void notify_all() {atomic_fence(); notify_all_relaxed();}
void notify_all_relaxed();
template<typename P> void notify( const P& predicate ) {atomic_fence(); notify_relaxed( predicate );}
template<typename P> void notify_relaxed( const P& predicate );
void abort_all() {atomic_fence(); abort_all_relaxed(); }
void abort_all_relaxed();
private:
tbb::spin_mutex mutex_ec;
waitset_t waitset_ec;
__TBB_atomic unsigned epoch;
thread_context* to_thread_context( waitset_node_t* n ) { return static_cast<thread_context*>(n); }
};
template<typename WaitUntil, typename Context>
void concurrent_monitor::wait( WaitUntil until, Context on )
{
bool slept = false;
thread_context thr_ctx;
prepare_wait( thr_ctx, on() );
while( !until() ) {
if( (slept = commit_wait( thr_ctx ) )==true )
if( until() ) break;
slept = false;
prepare_wait( thr_ctx, on() );
}
if( !slept )
cancel_wait( thr_ctx );
}
template<typename P>
void concurrent_monitor::notify_relaxed( const P& predicate ) {
if( waitset_ec.empty() )
return;
waitset_t temp;
waitset_node_t* nxt;
const waitset_node_t* end = waitset_ec.end();
{
tbb::spin_mutex::scoped_lock l( mutex_ec );
__TBB_store_relaxed(epoch, __TBB_load_relaxed(epoch) + 1);
for( waitset_node_t* n=waitset_ec.last(); n!=end; n=nxt ) {
nxt = n->prev;
thread_context* thr = to_thread_context( n );
if( predicate( thr->context ) ) {
waitset_ec.remove( *n );
thr->in_waitset = false;
temp.add( n );
}
}
}
end = temp.end();
for( waitset_node_t* n=temp.front(); n!=end; n=nxt ) {
nxt = n->next;
to_thread_context(n)->semaphore().V();
}
#if TBB_USE_ASSERT
temp.clear();
#endif
}
} // namespace internal
} // namespace tbb
#endif /* __TBB_concurrent_monitor_H */
tbb::internal::circular_doubly_linked_list_with_sentinel::count
__TBB_atomic size_t count
Definition: concurrent_monitor.h:84
tbb::spin_mutex
A lock that occupies a single byte.
Definition: spin_mutex.h:39
spin_mutex.h
tbb::internal::concurrent_monitor::cancel_wait
void cancel_wait(thread_context &thr)
Cancel the wait. Removes the thread from the wait queue if not removed yet.
Definition: concurrent_monitor.cpp:50
tbb::internal::circular_doubly_linked_list_with_sentinel
Circular doubly-linked list with sentinel.
Definition: concurrent_monitor.h:33
internal
Definition: _flow_graph_async_msg_impl.h:24
tbb::internal::binary_semaphore
binary_semaphore for concurrent monitor
Definition: semaphore.h:222
tbb::internal::concurrent_monitor::thread_context::sema
tbb::aligned_space< binary_semaphore > sema
Definition: concurrent_monitor.h:115
tbb::internal::concurrent_monitor::notify_relaxed
void notify_relaxed(const P &predicate)
Notify waiting threads of the event that satisfies the given predicate; Relaxed version.
Definition: concurrent_monitor.h:204
tbb::internal::concurrent_monitor::~concurrent_monitor
~concurrent_monitor()
dtor
Definition: concurrent_monitor.cpp:27
tbb::internal::concurrent_monitor::thread_context::epoch
__TBB_atomic unsigned epoch
Definition: concurrent_monitor.h:116
end
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 ITT_FORMAT p void ITT_FORMAT p void size_t ITT_FORMAT d void ITT_FORMAT p const wchar_t ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s const char ITT_FORMAT s no args void ITT_FORMAT p size_t ITT_FORMAT d no args const wchar_t const wchar_t ITT_FORMAT s __itt_heap_function void size_t int ITT_FORMAT d __itt_heap_function void ITT_FORMAT p __itt_heap_function void void size_t int ITT_FORMAT d no args no args unsigned int ITT_FORMAT u const __itt_domain __itt_id ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain __itt_id ITT_FORMAT p const __itt_domain __itt_id __itt_timestamp __itt_timestamp end
Definition: ittnotify_static.h:182
tbb::internal::circular_doubly_linked_list_with_sentinel::head
node_t head
Definition: concurrent_monitor.h:85
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::concurrent_monitor::thread_context::aborted
bool aborted
Definition: concurrent_monitor.h:119
tbb::internal::concurrent_monitor::abort_all
void abort_all()
Abort any sleeping threads at the time of the call.
Definition: concurrent_monitor.h:175
tbb::internal::throw_exception
void throw_exception(exception_id eid)
Versionless convenience wrapper for throw_exception_v4()
Definition: tbb_exception.h:105
tbb::internal::circular_doubly_linked_list_with_sentinel::begin
node_t * begin() const
Definition: concurrent_monitor.h:50
tbb::internal::concurrent_monitor::to_thread_context
thread_context * to_thread_context(waitset_node_t *n)
Definition: concurrent_monitor.h:184
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::circular_doubly_linked_list_with_sentinel::front
node_t * front() const
Definition: concurrent_monitor.h:48
tbb::internal::circular_doubly_linked_list_with_sentinel::size
size_t size() const
Definition: concurrent_monitor.h:46
tbb::internal::circular_doubly_linked_list_with_sentinel::node_t::next
node_t * next
Definition: concurrent_monitor.h:36
tbb::internal::concurrent_monitor
concurrent_monitor
Definition: concurrent_monitor.h:93
semaphore.h
tbb::internal::concurrent_monitor::wait
void wait(WaitUntil until, Context on)
Wait for a condition to be satisfied with waiting-on context.
Definition: concurrent_monitor.h:188
aligned_space.h
tbb::internal::concurrent_monitor::waitset_ec
waitset_t waitset_ec
Definition: concurrent_monitor.h:182
__TBB_atomic
#define __TBB_atomic
Definition: tbb_stddef.h:237
tbb::internal::concurrent_monitor::prepare_wait
void prepare_wait(thread_context &thr, uintptr_t ctx=0)
prepare wait by inserting 'thr' into the wait queue
Definition: concurrent_monitor.cpp:32
tbb::internal::concurrent_monitor::thread_context::semaphore
binary_semaphore & semaphore()
Definition: concurrent_monitor.h:109
tbb::internal::concurrent_monitor::epoch
__TBB_atomic unsigned epoch
Definition: concurrent_monitor.h:183
tbb::internal::__TBB_store_relaxed
void __TBB_store_relaxed(volatile T &location, V value)
Definition: tbb_machine.h:742
tbb::internal::circular_doubly_linked_list_with_sentinel::remove
void remove(node_t &n)
remove node 'n'
Definition: concurrent_monitor.h:63
tbb_exception.h
tbb::internal::concurrent_monitor::thread_context::skipped_wakeup
bool skipped_wakeup
Definition: concurrent_monitor.h:118
tbb::internal::circular_doubly_linked_list_with_sentinel::circular_doubly_linked_list_with_sentinel
circular_doubly_linked_list_with_sentinel()
Definition: concurrent_monitor.h:42
tbb::internal::circular_doubly_linked_list_with_sentinel::node_t::prev
node_t * prev
Definition: concurrent_monitor.h:37
atomic.h
tbb::internal::circular_doubly_linked_list_with_sentinel::~circular_doubly_linked_list_with_sentinel
~circular_doubly_linked_list_with_sentinel()
Definition: concurrent_monitor.h:44
tbb::internal::concurrent_monitor::thread_context::ready
bool ready
Definition: concurrent_monitor.h:120
tbb::internal::concurrent_monitor::thread_context::context
uintptr_t context
Definition: concurrent_monitor.h:121
tbb::internal::circular_doubly_linked_list_with_sentinel::empty
bool empty() const
Definition: concurrent_monitor.h:47
tbb::internal::concurrent_monitor::notify
void notify(const P &predicate)
Notify waiting threads of the event that satisfies the given predicate.
Definition: concurrent_monitor.h:169
tbb::internal::waitset_t
circular_doubly_linked_list_with_sentinel waitset_t
Definition: concurrent_monitor.h:88
tbb::internal::concurrent_monitor::thread_context::__TBB_NOINLINE
__TBB_NOINLINE(void init())
The method for lazy initialization of the thread_context's semaphore.
tbb::internal::concurrent_monitor::notify_all
void notify_all()
Notify all waiting threads of the event.
Definition: concurrent_monitor.h:163
tbb::internal::concurrent_monitor::notify_all_relaxed
void notify_all_relaxed()
Notify all waiting threads of the event; Relaxed version.
Definition: concurrent_monitor.cpp:84
tbb::internal::concurrent_monitor::thread_context
Definition: concurrent_monitor.h:96
tbb::internal::eid_user_abort
Definition: tbb_exception.h:85
tbb::internal::concurrent_monitor::abort_all_relaxed
void abort_all_relaxed()
Abort any sleeping threads at the time of the call; Relaxed version.
Definition: concurrent_monitor.cpp:107
tbb::internal::circular_doubly_linked_list_with_sentinel::clear
void clear()
Definition: concurrent_monitor.h:82
tbb::internal::circular_doubly_linked_list_with_sentinel::end
const node_t * end() const
Definition: concurrent_monitor.h:51
tbb::internal::concurrent_monitor::mutex_ec
tbb::spin_mutex mutex_ec
Definition: concurrent_monitor.h:181
tbb::internal::concurrent_monitor::thread_context::thread_context
thread_context()
Definition: concurrent_monitor.h:99
tbb::internal::concurrent_monitor::concurrent_monitor
concurrent_monitor()
ctor
Definition: concurrent_monitor.h:125
tbb::internal::concurrent_monitor::notify_one
void notify_one()
Notify one thread about the event.
Definition: concurrent_monitor.h:157
tbb_stddef.h
tbb::internal::waitset_node_t
circular_doubly_linked_list_with_sentinel::node_t waitset_node_t
Definition: concurrent_monitor.h:89
tbb::atomic_fence
void atomic_fence()
Sequentially consistent full memory fence.
Definition: tbb_machine.h:342
tbb::internal::binary_semaphore::~binary_semaphore
~binary_semaphore()
dtor
Definition: semaphore.h:230
tbb::internal::circular_doubly_linked_list_with_sentinel::node_t::node_t
node_t()
Definition: concurrent_monitor.h:38
tbb::internal::concurrent_monitor::commit_wait
bool commit_wait(thread_context &thr)
Commit wait if event count has not changed; otherwise, cancel wait.
Definition: concurrent_monitor.h:135
tbb::internal::concurrent_monitor::thread_context::~thread_context
~thread_context()
Definition: concurrent_monitor.h:103
tbb::internal::binary_semaphore::P
void P()
wait/acquire
Definition: semaphore.h:235
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
tbb::internal::circular_doubly_linked_list_with_sentinel::node_t
Definition: concurrent_monitor.h:35
tbb::internal::concurrent_monitor::thread_context::in_waitset
tbb::atomic< bool > in_waitset
Definition: concurrent_monitor.h:117
tbb::internal::circular_doubly_linked_list_with_sentinel::add
void add(node_t *n)
add to the back of the list
Definition: concurrent_monitor.h:54
tbb::internal::circular_doubly_linked_list_with_sentinel::last
node_t * last() const
Definition: concurrent_monitor.h:49
tbb::internal::concurrent_monitor::notify_one_relaxed
void notify_one_relaxed()
Notify one thread about the event. Relaxed version.
Definition: concurrent_monitor.cpp:66
tbb::internal::binary_semaphore::V
void V()
post/release
Definition: semaphore.h:240
tbb::internal::__TBB_load_relaxed
T __TBB_load_relaxed(const volatile T &location)
Definition: tbb_machine.h:738
tbb::internal::circular_doubly_linked_list_with_sentinel::flush_to
void flush_to(circular_doubly_linked_list_with_sentinel &lst)
move all elements to 'lst' and initialize the 'this' list
Definition: concurrent_monitor.h:71

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.