concurrent_queue.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_stddef.h"
#include "tbb/tbb_machine.h"
#include "tbb/tbb_exception.h"
// Define required to satisfy test in internal file.
#define __TBB_concurrent_queue_H
#include "tbb/internal/_concurrent_queue_impl.h"
#include "concurrent_monitor.h"
#include "itt_notify.h"
#include <new>
#include <cstring> // for memset()
#if defined(_MSC_VER) && defined(_Wp64)
// Workaround for overzealous compiler warnings in /Wp64 mode
#pragma warning (disable: 4267)
#endif
#define RECORD_EVENTS 0
namespace tbb {
namespace internal {
typedef concurrent_queue_base_v3 concurrent_queue_base;
typedef size_t ticket;
struct micro_queue {
typedef concurrent_queue_base::page page;
friend class micro_queue_pop_finalizer;
atomic<page*> head_page;
atomic<ticket> head_counter;
atomic<page*> tail_page;
atomic<ticket> tail_counter;
spin_mutex page_mutex;
void push( const void* item, ticket k, concurrent_queue_base& base,
concurrent_queue_base::copy_specifics op_type );
void abort_push( ticket k, concurrent_queue_base& base );
bool pop( void* dst, ticket k, concurrent_queue_base& base );
micro_queue& assign( const micro_queue& src, concurrent_queue_base& base,
concurrent_queue_base::copy_specifics op_type );
page* make_copy ( concurrent_queue_base& base, const page* src_page, size_t begin_in_page,
size_t end_in_page, ticket& g_index, concurrent_queue_base::copy_specifics op_type ) ;
void make_invalid( ticket k );
};
// we need to yank it out of micro_queue because of concurrent_queue_base::deallocate_page being virtual.
class micro_queue_pop_finalizer: no_copy {
typedef concurrent_queue_base::page page;
ticket my_ticket;
micro_queue& my_queue;
page* my_page;
concurrent_queue_base &base;
public:
micro_queue_pop_finalizer( micro_queue& queue, concurrent_queue_base& b, ticket k, page* p ) :
my_ticket(k), my_queue(queue), my_page(p), base(b)
{}
~micro_queue_pop_finalizer() {
page* p = my_page;
if( p ) {
spin_mutex::scoped_lock lock( my_queue.page_mutex );
page* q = p->next;
my_queue.head_page = q;
if( !q ) {
my_queue.tail_page = NULL;
}
}
my_queue.head_counter = my_ticket;
if( p )
base.deallocate_page( p );
}
};
struct predicate_leq {
ticket t;
predicate_leq( ticket t_ ) : t(t_) {}
bool operator() ( uintptr_t p ) const {return (ticket)p<=t;}
};
class concurrent_queue_rep {
public:
private:
friend struct micro_queue;
static const size_t phi = 3;
public:
static const size_t n_queue = 8;
static size_t index( ticket k ) {
return k*phi%n_queue;
}
atomic<ticket> head_counter;
concurrent_monitor items_avail;
atomic<size_t> n_invalid_entries;
char pad1[NFS_MaxLineSize-((sizeof(atomic<ticket>)+sizeof(concurrent_monitor)+sizeof(atomic<size_t>))&(NFS_MaxLineSize-1))];
atomic<ticket> tail_counter;
concurrent_monitor slots_avail;
char pad2[NFS_MaxLineSize-((sizeof(atomic<ticket>)+sizeof(concurrent_monitor))&(NFS_MaxLineSize-1))];
micro_queue array[n_queue];
micro_queue& choose( ticket k ) {
// The formula here approximates LRU in a cache-oblivious way.
return array[index(k)];
}
atomic<unsigned> abort_counter;
static const ptrdiff_t infinite_capacity = ptrdiff_t(~size_t(0)/2);
};
#if _MSC_VER && !defined(__INTEL_COMPILER)
// unary minus operator applied to unsigned type, result still unsigned
#pragma warning( push )
#pragma warning( disable: 4146 )
#endif
static void* static_invalid_page;
//------------------------------------------------------------------------
// micro_queue
//------------------------------------------------------------------------
void micro_queue::push( const void* item, ticket k, concurrent_queue_base& base,
concurrent_queue_base::copy_specifics op_type ) {
k &= -concurrent_queue_rep::n_queue;
page* p = NULL;
// find index on page where we would put the data
size_t index = modulo_power_of_two( k/concurrent_queue_rep::n_queue, base.items_per_page );
if( !index ) { // make a new page
__TBB_TRY {
p = base.allocate_page();
} __TBB_CATCH(...) {
++base.my_rep->n_invalid_entries;
make_invalid( k );
__TBB_RETHROW();
}
p->mask = 0;
p->next = NULL;
}
// wait for my turn
if( tail_counter!=k ) // The developer insisted on keeping first check out of the backoff loop
for( atomic_backoff b(true);;b.pause() ) {
ticket tail = tail_counter;
if( tail==k ) break;
else if( tail&0x1 ) {
// no memory. throws an exception; assumes concurrent_queue_rep::n_queue>1
++base.my_rep->n_invalid_entries;
throw_exception( eid_bad_last_alloc );
}
}
if( p ) { // page is newly allocated; insert in micro_queue
spin_mutex::scoped_lock lock( page_mutex );
if( page* q = tail_page )
q->next = p;
else
head_page = p;
tail_page = p;
}
if (item) {
p = tail_page;
ITT_NOTIFY( sync_acquired, p );
__TBB_TRY {
if( concurrent_queue_base::copy == op_type ) {
base.copy_item( *p, index, item );
} else {
__TBB_ASSERT( concurrent_queue_base::move == op_type, NULL );
static_cast<concurrent_queue_base_v8&>(base).move_item( *p, index, item );
}
} __TBB_CATCH(...) {
++base.my_rep->n_invalid_entries;
tail_counter += concurrent_queue_rep::n_queue;
__TBB_RETHROW();
}
ITT_NOTIFY( sync_releasing, p );
// If no exception was thrown, mark item as present.
p->mask |= uintptr_t(1)<<index;
}
else // no item; this was called from abort_push
++base.my_rep->n_invalid_entries;
tail_counter += concurrent_queue_rep::n_queue;
}
void micro_queue::abort_push( ticket k, concurrent_queue_base& base ) {
push(NULL, k, base, concurrent_queue_base::copy);
}
bool micro_queue::pop( void* dst, ticket k, concurrent_queue_base& base ) {
k &= -concurrent_queue_rep::n_queue;
spin_wait_until_eq( head_counter, k );
spin_wait_while_eq( tail_counter, k );
page *p = head_page;
__TBB_ASSERT( p, NULL );
size_t index = modulo_power_of_two( k/concurrent_queue_rep::n_queue, base.items_per_page );
bool success = false;
{
micro_queue_pop_finalizer finalizer( *this, base, k+concurrent_queue_rep::n_queue, index==base.items_per_page-1 ? p : NULL );
if( p->mask & uintptr_t(1)<<index ) {
success = true;
ITT_NOTIFY( sync_acquired, dst );
ITT_NOTIFY( sync_acquired, head_page );
base.assign_and_destroy_item( dst, *p, index );
ITT_NOTIFY( sync_releasing, head_page );
} else {
--base.my_rep->n_invalid_entries;
}
}
return success;
}
micro_queue& micro_queue::assign( const micro_queue& src, concurrent_queue_base& base,
concurrent_queue_base::copy_specifics op_type )
{
head_counter = src.head_counter;
tail_counter = src.tail_counter;
const page* srcp = src.head_page;
if( srcp ) {
ticket g_index = head_counter;
__TBB_TRY {
size_t n_items = (tail_counter-head_counter)/concurrent_queue_rep::n_queue;
size_t index = modulo_power_of_two( head_counter/concurrent_queue_rep::n_queue, base.items_per_page );
size_t end_in_first_page = (index+n_items<base.items_per_page)?(index+n_items):base.items_per_page;
head_page = make_copy( base, srcp, index, end_in_first_page, g_index, op_type );
page* cur_page = head_page;
if( srcp != src.tail_page ) {
for( srcp = srcp->next; srcp!=src.tail_page; srcp=srcp->next ) {
cur_page->next = make_copy( base, srcp, 0, base.items_per_page, g_index, op_type );
cur_page = cur_page->next;
}
__TBB_ASSERT( srcp==src.tail_page, NULL );
size_t last_index = modulo_power_of_two( tail_counter/concurrent_queue_rep::n_queue, base.items_per_page );
if( last_index==0 ) last_index = base.items_per_page;
cur_page->next = make_copy( base, srcp, 0, last_index, g_index, op_type );
cur_page = cur_page->next;
}
tail_page = cur_page;
} __TBB_CATCH(...) {
make_invalid( g_index );
__TBB_RETHROW();
}
} else {
head_page = tail_page = NULL;
}
return *this;
}
concurrent_queue_base::page* micro_queue::make_copy( concurrent_queue_base& base,
const concurrent_queue_base::page* src_page, size_t begin_in_page, size_t end_in_page,
ticket& g_index, concurrent_queue_base::copy_specifics op_type )
{
page* new_page = base.allocate_page();
new_page->next = NULL;
new_page->mask = src_page->mask;
for( ; begin_in_page!=end_in_page; ++begin_in_page, ++g_index )
if( new_page->mask & uintptr_t(1)<<begin_in_page ) {
if( concurrent_queue_base::copy == op_type ) {
base.copy_page_item( *new_page, begin_in_page, *src_page, begin_in_page );
} else {
__TBB_ASSERT( concurrent_queue_base::move == op_type, NULL );
static_cast<concurrent_queue_base_v8&>(base).move_page_item( *new_page, begin_in_page, *src_page, begin_in_page );
}
}
return new_page;
}
void micro_queue::make_invalid( ticket k )
{
static concurrent_queue_base::page dummy = {static_cast<page*>((void*)1), 0};
// mark it so that no more pushes are allowed.
static_invalid_page = &dummy;
{
spin_mutex::scoped_lock lock( page_mutex );
tail_counter = k+concurrent_queue_rep::n_queue+1;
if( page* q = tail_page )
q->next = static_cast<page*>(static_invalid_page);
else
head_page = static_cast<page*>(static_invalid_page);
tail_page = static_cast<page*>(static_invalid_page);
}
}
#if _MSC_VER && !defined(__INTEL_COMPILER)
#pragma warning( pop )
#endif // warning 4146 is back
//------------------------------------------------------------------------
// concurrent_queue_base
//------------------------------------------------------------------------
concurrent_queue_base_v3::concurrent_queue_base_v3( size_t item_sz ) {
items_per_page = item_sz<= 8 ? 32 :
item_sz<= 16 ? 16 :
item_sz<= 32 ? 8 :
item_sz<= 64 ? 4 :
item_sz<=128 ? 2 :
1;
my_capacity = size_t(-1)/(item_sz>1 ? item_sz : 2);
my_rep = cache_aligned_allocator<concurrent_queue_rep>().allocate(1);
__TBB_ASSERT( is_aligned(my_rep, NFS_GetLineSize()), "alignment error" );
__TBB_ASSERT( is_aligned(&my_rep->head_counter, NFS_GetLineSize()), "alignment error" );
__TBB_ASSERT( is_aligned(&my_rep->tail_counter, NFS_GetLineSize()), "alignment error" );
__TBB_ASSERT( is_aligned(&my_rep->array, NFS_GetLineSize()), "alignment error" );
std::memset(static_cast<void*>(my_rep),0,sizeof(concurrent_queue_rep));
new ( &my_rep->items_avail ) concurrent_monitor();
new ( &my_rep->slots_avail ) concurrent_monitor();
this->item_size = item_sz;
}
concurrent_queue_base_v3::~concurrent_queue_base_v3() {
size_t nq = my_rep->n_queue;
for( size_t i=0; i<nq; i++ )
__TBB_ASSERT( my_rep->array[i].tail_page==NULL, "pages were not freed properly" );
cache_aligned_allocator<concurrent_queue_rep>().deallocate(my_rep,1);
}
void concurrent_queue_base_v3::internal_push( const void* src ) {
internal_insert_item( src, copy );
}
void concurrent_queue_base_v8::internal_push_move( const void* src ) {
internal_insert_item( src, move );
}
void concurrent_queue_base_v3::internal_insert_item( const void* src, copy_specifics op_type ) {
concurrent_queue_rep& r = *my_rep;
unsigned old_abort_counter = r.abort_counter;
ticket k = r.tail_counter++;
ptrdiff_t e = my_capacity;
#if DO_ITT_NOTIFY
bool sync_prepare_done = false;
#endif
if( (ptrdiff_t)(k-r.head_counter)>=e ) { // queue is full
#if DO_ITT_NOTIFY
if( !sync_prepare_done ) {
ITT_NOTIFY( sync_prepare, &sync_prepare_done );
sync_prepare_done = true;
}
#endif
bool slept = false;
concurrent_monitor::thread_context thr_ctx;
r.slots_avail.prepare_wait( thr_ctx, ((ptrdiff_t)(k-e)) );
while( (ptrdiff_t)(k-r.head_counter)>=const_cast<volatile ptrdiff_t&>(e = my_capacity) ) {
__TBB_TRY {
if( r.abort_counter!=old_abort_counter ) {
r.slots_avail.cancel_wait( thr_ctx );
throw_exception( eid_user_abort );
}
slept = r.slots_avail.commit_wait( thr_ctx );
} __TBB_CATCH( tbb::user_abort& ) {
r.choose(k).abort_push(k, *this);
__TBB_RETHROW();
} __TBB_CATCH(...) {
__TBB_RETHROW();
}
if (slept == true) break;
r.slots_avail.prepare_wait( thr_ctx, ((ptrdiff_t)(k-e)) );
}
if( !slept )
r.slots_avail.cancel_wait( thr_ctx );
}
ITT_NOTIFY( sync_acquired, &sync_prepare_done );
__TBB_ASSERT( (ptrdiff_t)(k-r.head_counter)<my_capacity, NULL);
r.choose( k ).push( src, k, *this, op_type );
r.items_avail.notify( predicate_leq(k) );
}
void concurrent_queue_base_v3::internal_pop( void* dst ) {
concurrent_queue_rep& r = *my_rep;
ticket k;
#if DO_ITT_NOTIFY
bool sync_prepare_done = false;
#endif
unsigned old_abort_counter = r.abort_counter;
// This loop is a single pop operation; abort_counter should not be re-read inside
do {
k=r.head_counter++;
if ( (ptrdiff_t)(r.tail_counter-k)<=0 ) { // queue is empty
#if DO_ITT_NOTIFY
if( !sync_prepare_done ) {
ITT_NOTIFY( sync_prepare, dst );
sync_prepare_done = true;
}
#endif
bool slept = false;
concurrent_monitor::thread_context thr_ctx;
r.items_avail.prepare_wait( thr_ctx, k );
while( (ptrdiff_t)(r.tail_counter-k)<=0 ) {
__TBB_TRY {
if( r.abort_counter!=old_abort_counter ) {
r.items_avail.cancel_wait( thr_ctx );
throw_exception( eid_user_abort );
}
slept = r.items_avail.commit_wait( thr_ctx );
} __TBB_CATCH( tbb::user_abort& ) {
r.head_counter--;
__TBB_RETHROW();
} __TBB_CATCH(...) {
__TBB_RETHROW();
}
if (slept == true) break;
r.items_avail.prepare_wait( thr_ctx, k );
}
if( !slept )
r.items_avail.cancel_wait( thr_ctx );
}
__TBB_ASSERT((ptrdiff_t)(r.tail_counter-k)>0, NULL);
} while( !r.choose(k).pop(dst,k,*this) );
// wake up a producer..
r.slots_avail.notify( predicate_leq(k) );
}
void concurrent_queue_base_v3::internal_abort() {
concurrent_queue_rep& r = *my_rep;
++r.abort_counter;
r.items_avail.abort_all();
r.slots_avail.abort_all();
}
bool concurrent_queue_base_v3::internal_pop_if_present( void* dst ) {
concurrent_queue_rep& r = *my_rep;
ticket k;
do {
k = r.head_counter;
for(;;) {
if( (ptrdiff_t)(r.tail_counter-k)<=0 ) {
// Queue is empty
return false;
}
// Queue had item with ticket k when we looked. Attempt to get that item.
ticket tk=k;
k = r.head_counter.compare_and_swap( tk+1, tk );
if( k==tk )
break;
// Another thread snatched the item, retry.
}
} while( !r.choose( k ).pop( dst, k, *this ) );
r.slots_avail.notify( predicate_leq(k) );
return true;
}
bool concurrent_queue_base_v3::internal_push_if_not_full( const void* src ) {
return internal_insert_if_not_full( src, copy );
}
bool concurrent_queue_base_v8::internal_push_move_if_not_full( const void* src ) {
return internal_insert_if_not_full( src, move );
}
bool concurrent_queue_base_v3::internal_insert_if_not_full( const void* src, copy_specifics op_type ) {
concurrent_queue_rep& r = *my_rep;
ticket k = r.tail_counter;
for(;;) {
if( (ptrdiff_t)(k-r.head_counter)>=my_capacity ) {
// Queue is full
return false;
}
// Queue had empty slot with ticket k when we looked. Attempt to claim that slot.
ticket tk=k;
k = r.tail_counter.compare_and_swap( tk+1, tk );
if( k==tk )
break;
// Another thread claimed the slot, so retry.
}
r.choose(k).push(src, k, *this, op_type);
r.items_avail.notify( predicate_leq(k) );
return true;
}
ptrdiff_t concurrent_queue_base_v3::internal_size() const {
__TBB_ASSERT( sizeof(ptrdiff_t)<=sizeof(size_t), NULL );
return ptrdiff_t(my_rep->tail_counter-my_rep->head_counter-my_rep->n_invalid_entries);
}
bool concurrent_queue_base_v3::internal_empty() const {
ticket tc = my_rep->tail_counter;
ticket hc = my_rep->head_counter;
// if tc!=r.tail_counter, the queue was not empty at some point between the two reads.
return ( tc==my_rep->tail_counter && ptrdiff_t(tc-hc-my_rep->n_invalid_entries)<=0 );
}
void concurrent_queue_base_v3::internal_set_capacity( ptrdiff_t capacity, size_t /*item_sz*/ ) {
my_capacity = capacity<0 ? concurrent_queue_rep::infinite_capacity : capacity;
}
void concurrent_queue_base_v3::internal_finish_clear() {
size_t nq = my_rep->n_queue;
for( size_t i=0; i<nq; ++i ) {
page* tp = my_rep->array[i].tail_page;
__TBB_ASSERT( my_rep->array[i].head_page==tp, "at most one page should remain" );
if( tp!=NULL) {
if( tp!=static_invalid_page ) deallocate_page( tp );
my_rep->array[i].tail_page = NULL;
}
}
}
void concurrent_queue_base_v3::internal_throw_exception() const {
throw_exception( eid_bad_alloc );
}
void concurrent_queue_base_v3::internal_assign( const concurrent_queue_base& src, copy_specifics op_type ) {
items_per_page = src.items_per_page;
my_capacity = src.my_capacity;
// copy concurrent_queue_rep.
my_rep->head_counter = src.my_rep->head_counter;
my_rep->tail_counter = src.my_rep->tail_counter;
my_rep->n_invalid_entries = src.my_rep->n_invalid_entries;
my_rep->abort_counter = src.my_rep->abort_counter;
// copy micro_queues
for( size_t i = 0; i<my_rep->n_queue; ++i )
my_rep->array[i].assign( src.my_rep->array[i], *this, op_type );
__TBB_ASSERT( my_rep->head_counter==src.my_rep->head_counter && my_rep->tail_counter==src.my_rep->tail_counter,
"the source concurrent queue should not be concurrently modified." );
}
void concurrent_queue_base_v3::assign( const concurrent_queue_base& src ) {
internal_assign( src, copy );
}
void concurrent_queue_base_v8::move_content( concurrent_queue_base_v8& src ) {
internal_assign( src, move );
}
//------------------------------------------------------------------------
// concurrent_queue_iterator_rep
//------------------------------------------------------------------------
class concurrent_queue_iterator_rep: no_assign {
public:
ticket head_counter;
const concurrent_queue_base& my_queue;
const size_t offset_of_last;
concurrent_queue_base::page* array[concurrent_queue_rep::n_queue];
concurrent_queue_iterator_rep( const concurrent_queue_base& queue, size_t offset_of_last_ ) :
head_counter(queue.my_rep->head_counter),
my_queue(queue),
offset_of_last(offset_of_last_)
{
const concurrent_queue_rep& rep = *queue.my_rep;
for( size_t k=0; k<concurrent_queue_rep::n_queue; ++k )
array[k] = rep.array[k].head_page;
}
bool get_item( void*& item, size_t k ) {
if( k==my_queue.my_rep->tail_counter ) {
item = NULL;
return true;
} else {
concurrent_queue_base::page* p = array[concurrent_queue_rep::index(k)];
__TBB_ASSERT(p,NULL);
size_t i = modulo_power_of_two( k/concurrent_queue_rep::n_queue, my_queue.items_per_page );
item = static_cast<unsigned char*>(static_cast<void*>(p)) + offset_of_last + my_queue.item_size*i;
return (p->mask & uintptr_t(1)<<i)!=0;
}
}
};
//------------------------------------------------------------------------
// concurrent_queue_iterator_base
//------------------------------------------------------------------------
void concurrent_queue_iterator_base_v3::initialize( const concurrent_queue_base& queue, size_t offset_of_last ) {
my_rep = cache_aligned_allocator<concurrent_queue_iterator_rep>().allocate(1);
new( my_rep ) concurrent_queue_iterator_rep(queue,offset_of_last);
size_t k = my_rep->head_counter;
if( !my_rep->get_item(my_item, k) ) advance();
}
concurrent_queue_iterator_base_v3::concurrent_queue_iterator_base_v3( const concurrent_queue_base& queue ) {
initialize(queue,0);
}
concurrent_queue_iterator_base_v3::concurrent_queue_iterator_base_v3( const concurrent_queue_base& queue, size_t offset_of_last ) {
initialize(queue,offset_of_last);
}
void concurrent_queue_iterator_base_v3::assign( const concurrent_queue_iterator_base& other ) {
if( my_rep!=other.my_rep ) {
if( my_rep ) {
cache_aligned_allocator<concurrent_queue_iterator_rep>().deallocate(my_rep, 1);
my_rep = NULL;
}
if( other.my_rep ) {
my_rep = cache_aligned_allocator<concurrent_queue_iterator_rep>().allocate(1);
new( my_rep ) concurrent_queue_iterator_rep( *other.my_rep );
}
}
my_item = other.my_item;
}
void concurrent_queue_iterator_base_v3::advance() {
__TBB_ASSERT( my_item, "attempt to increment iterator past end of queue" );
size_t k = my_rep->head_counter;
const concurrent_queue_base& queue = my_rep->my_queue;
#if TBB_USE_ASSERT
void* tmp;
my_rep->get_item(tmp,k);
__TBB_ASSERT( my_item==tmp, NULL );
#endif /* TBB_USE_ASSERT */
size_t i = modulo_power_of_two( k/concurrent_queue_rep::n_queue, queue.items_per_page );
if( i==queue.items_per_page-1 ) {
concurrent_queue_base::page*& root = my_rep->array[concurrent_queue_rep::index(k)];
root = root->next;
}
// advance k
my_rep->head_counter = ++k;
if( !my_rep->get_item(my_item, k) ) advance();
}
concurrent_queue_iterator_base_v3::~concurrent_queue_iterator_base_v3() {
//delete my_rep;
cache_aligned_allocator<concurrent_queue_iterator_rep>().deallocate(my_rep, 1);
my_rep = NULL;
}
} // namespace internal
} // namespace tbb
_concurrent_queue_impl.h
tbb_exception.h
tbb_machine.h
tbb_stddef.h
__TBB_CATCH
#define __TBB_CATCH(e)
Definition: tbb_stddef.h:284
__TBB_TRY
#define __TBB_TRY
Definition: tbb_stddef.h:283
__TBB_ASSERT
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
Definition: tbb_stddef.h:165
__TBB_RETHROW
#define __TBB_RETHROW()
Definition: tbb_stddef.h:286
concurrent_monitor.h
itt_notify.h
ITT_NOTIFY
#define ITT_NOTIFY(name, obj)
Definition: itt_notify.h:112
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
sync_releasing
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 sync_releasing
Definition: ittnotify_static.h:104
p
void const char const char int ITT_FORMAT __itt_group_sync p
Definition: ittnotify_static.h:91
tail
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 ITT_FORMAT lu const __itt_domain __itt_id __itt_id __itt_string_handle ITT_FORMAT p const __itt_domain ITT_FORMAT p const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_string_handle unsigned long long ITT_FORMAT lu const __itt_domain __itt_id __itt_string_handle __itt_metadata_type size_t void ITT_FORMAT p const __itt_domain __itt_id __itt_string_handle const wchar_t size_t ITT_FORMAT lu const __itt_domain __itt_id __itt_relation __itt_id tail
Definition: ittnotify_static.h:207
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::internal::NFS_GetLineSize
size_t __TBB_EXPORTED_FUNC NFS_GetLineSize()
Cache/sector line size.
Definition: cache_aligned_allocator.cpp:167
tbb
The graph class.
Definition: serial/tbb/parallel_for.h:46
tbb::internal::static_invalid_page
static void * static_invalid_page
Definition: concurrent_queue.cpp:156
tbb::internal::throw_exception
void throw_exception(exception_id eid)
Versionless convenience wrapper for throw_exception_v4()
Definition: tbb_exception.h:105
tbb::internal::ticket
size_t ticket
Definition: concurrent_queue.cpp:42
tbb::internal::spin_wait_until_eq
void spin_wait_until_eq(const volatile T &location, const U value)
Spin UNTIL the value of the variable is equal to a given value.
Definition: tbb_machine.h:402
tbb::internal::is_aligned
bool is_aligned(T *pointer, uintptr_t alignment)
A function to check if passed in pointer is aligned on a specific border.
Definition: tbb_stddef.h:370
tbb::internal::eid_bad_alloc
@ eid_bad_alloc
Definition: tbb_exception.h:68
tbb::internal::eid_bad_last_alloc
@ eid_bad_last_alloc
Definition: tbb_exception.h:69
tbb::internal::eid_user_abort
@ eid_user_abort
Definition: tbb_exception.h:85
tbb::internal::concurrent_queue_base
concurrent_queue_base_v3 concurrent_queue_base
Definition: concurrent_queue.cpp:40
tbb::internal::modulo_power_of_two
argument_integer_type modulo_power_of_two(argument_integer_type arg, divisor_integer_type divisor)
A function to compute arg modulo divisor where divisor is a power of 2.
Definition: tbb_stddef.h:382
tbb::internal::spin_wait_while_eq
void spin_wait_while_eq(const volatile T &location, U value)
Spin WHILE the value of the variable is equal to a given value.
Definition: tbb_machine.h:394
internal
Definition: _flow_graph_async_msg_impl.h:24
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::concurrent_queue_base_v3
For internal use only.
Definition: _concurrent_queue_impl.h:826
tbb::internal::concurrent_queue_base_v3::internal_pop
void __TBB_EXPORTED_METHOD internal_pop(void *dst)
Dequeue item from head of queue.
Definition: concurrent_queue.cpp:414
tbb::internal::concurrent_queue_base_v3::allocate_page
virtual page * allocate_page()=0
custom allocator
tbb::internal::concurrent_queue_base_v3::internal_abort
void __TBB_EXPORTED_METHOD internal_abort()
Abort all pending queue operations.
Definition: concurrent_queue.cpp:460
tbb::internal::concurrent_queue_base_v3::copy_page_item
virtual void copy_page_item(page &dst, size_t dindex, const page &src, size_t sindex)=0
tbb::internal::concurrent_queue_base_v3::assign_and_destroy_item
virtual void assign_and_destroy_item(void *dst, page &src, size_t index)=0
tbb::internal::concurrent_queue_base_v3::~concurrent_queue_base_v3
virtual __TBB_EXPORTED_METHOD ~concurrent_queue_base_v3()
Definition: concurrent_queue.cpp:356
tbb::internal::concurrent_queue_base_v3::assign
void __TBB_EXPORTED_METHOD assign(const concurrent_queue_base_v3 &src)
copy internal representation
Definition: concurrent_queue.cpp:569
tbb::internal::concurrent_queue_base_v3::my_rep
concurrent_queue_rep * my_rep
Internal representation.
Definition: _concurrent_queue_impl.h:829
tbb::internal::concurrent_queue_base_v3::copy_item
virtual void copy_item(page &dst, size_t index, const void *src)=0
tbb::internal::concurrent_queue_base_v3::internal_set_capacity
void __TBB_EXPORTED_METHOD internal_set_capacity(ptrdiff_t capacity, size_t element_size)
Set the queue capacity.
Definition: concurrent_queue.cpp:531
tbb::internal::concurrent_queue_base_v3::item_size
size_t item_size
Size of an item.
Definition: _concurrent_queue_impl.h:850
tbb::internal::concurrent_queue_base_v3::internal_push
void __TBB_EXPORTED_METHOD internal_push(const void *src)
Enqueue item at tail of queue using copy operation.
Definition: concurrent_queue.cpp:363
tbb::internal::concurrent_queue_base_v3::internal_push_if_not_full
bool __TBB_EXPORTED_METHOD internal_push_if_not_full(const void *src)
Attempt to enqueue item onto queue using copy operation.
Definition: concurrent_queue.cpp:491
tbb::internal::concurrent_queue_base_v3::internal_throw_exception
void __TBB_EXPORTED_METHOD internal_throw_exception() const
throw an exception
Definition: concurrent_queue.cpp:547
tbb::internal::concurrent_queue_base_v3::internal_assign
void internal_assign(const concurrent_queue_base_v3 &src, copy_specifics op_type)
Assigns one queue to another using specified operation (copy or move)
Definition: concurrent_queue.cpp:551
tbb::internal::concurrent_queue_base_v3::internal_insert_item
void internal_insert_item(const void *src, copy_specifics op_type)
Enqueues item at tail of queue using specified operation (copy or move)
Definition: concurrent_queue.cpp:371
tbb::internal::concurrent_queue_base_v3::internal_pop_if_present
bool __TBB_EXPORTED_METHOD internal_pop_if_present(void *dst)
Attempt to dequeue item from queue.
Definition: concurrent_queue.cpp:467
tbb::internal::concurrent_queue_base_v3::concurrent_queue_base_v3
__TBB_EXPORTED_METHOD concurrent_queue_base_v3(size_t item_size)
Definition: concurrent_queue.cpp:337
tbb::internal::concurrent_queue_base_v3::my_capacity
ptrdiff_t my_capacity
Capacity of the queue.
Definition: _concurrent_queue_impl.h:844
tbb::internal::concurrent_queue_base_v3::deallocate_page
virtual void deallocate_page(page *p)=0
custom de-allocator
tbb::internal::concurrent_queue_base_v3::internal_insert_if_not_full
bool internal_insert_if_not_full(const void *src, copy_specifics op_type)
Attempts to enqueue at tail of queue using specified operation (copy or move)
Definition: concurrent_queue.cpp:499
tbb::internal::concurrent_queue_base_v3::internal_finish_clear
void __TBB_EXPORTED_METHOD internal_finish_clear()
free any remaining pages
Definition: concurrent_queue.cpp:535
tbb::internal::concurrent_queue_base_v3::internal_size
ptrdiff_t __TBB_EXPORTED_METHOD internal_size() const
Get size of queue.
Definition: concurrent_queue.cpp:519
tbb::internal::concurrent_queue_base_v3::internal_empty
bool __TBB_EXPORTED_METHOD internal_empty() const
Check if the queue is empty.
Definition: concurrent_queue.cpp:524
tbb::internal::concurrent_queue_base_v3::copy_specifics
copy_specifics
Definition: _concurrent_queue_impl.h:852
tbb::internal::concurrent_queue_base_v3::move
@ move
Definition: _concurrent_queue_impl.h:852
tbb::internal::concurrent_queue_base_v3::copy
@ copy
Definition: _concurrent_queue_impl.h:852
tbb::internal::concurrent_queue_base_v3::items_per_page
size_t items_per_page
Always a power of 2.
Definition: _concurrent_queue_impl.h:847
tbb::internal::concurrent_queue_base_v3::page
Prefix on a page.
Definition: _concurrent_queue_impl.h:838
tbb::internal::concurrent_queue_base_v3::page::mask
uintptr_t mask
Definition: _concurrent_queue_impl.h:840
tbb::internal::concurrent_queue_base_v3::page::next
page * next
Definition: _concurrent_queue_impl.h:839
tbb::internal::concurrent_queue_base_v8
For internal use only.
Definition: _concurrent_queue_impl.h:940
tbb::internal::concurrent_queue_base_v8::internal_push_move_if_not_full
bool __TBB_EXPORTED_METHOD internal_push_move_if_not_full(const void *src)
Attempt to enqueue item onto queue using move operation.
Definition: concurrent_queue.cpp:495
tbb::internal::concurrent_queue_base_v8::internal_push_move
void __TBB_EXPORTED_METHOD internal_push_move(const void *src)
Enqueue item at tail of queue using move operation.
Definition: concurrent_queue.cpp:367
tbb::internal::concurrent_queue_base_v8::move_content
void __TBB_EXPORTED_METHOD move_content(concurrent_queue_base_v8 &src)
move items
Definition: concurrent_queue.cpp:573
tbb::internal::concurrent_queue_iterator_base_v3
Type-independent portion of concurrent_queue_iterator.
Definition: _concurrent_queue_impl.h:960
tbb::internal::concurrent_queue_iterator_base_v3::initialize
void initialize(const concurrent_queue_base_v3 &queue, size_t offset_of_data)
Definition: concurrent_queue.cpp:614
tbb::internal::concurrent_queue_iterator_base_v3::my_item
void * my_item
Pointer to current item.
Definition: _concurrent_queue_impl.h:974
tbb::internal::concurrent_queue_iterator_base_v3::my_rep
concurrent_queue_iterator_rep * my_rep
concurrent_queue over which we are iterating.
Definition: _concurrent_queue_impl.h:963
tbb::internal::concurrent_queue_iterator_base_v3::concurrent_queue_iterator_base_v3
concurrent_queue_iterator_base_v3()
Default constructor.
Definition: _concurrent_queue_impl.h:977
tbb::internal::concurrent_queue_iterator_base_v3::assign
void __TBB_EXPORTED_METHOD assign(const concurrent_queue_iterator_base_v3 &i)
Assignment.
Definition: concurrent_queue.cpp:629
tbb::internal::concurrent_queue_iterator_base_v3::advance
void __TBB_EXPORTED_METHOD advance()
Advance iterator one step towards tail of queue.
Definition: concurrent_queue.cpp:643
tbb::internal::concurrent_queue_iterator_base_v3::~concurrent_queue_iterator_base_v3
__TBB_EXPORTED_METHOD ~concurrent_queue_iterator_base_v3()
Destructor.
Definition: concurrent_queue.cpp:662
tbb::spin_mutex
A lock that occupies a single byte.
Definition: spin_mutex.h:39
tbb::spin_mutex::scoped_lock
Represents acquisition of a mutex.
Definition: spin_mutex.h:53
tbb::user_abort
Exception for user-initiated abort.
Definition: tbb_exception.h:46
tbb::internal::atomic_backoff
Class that implements exponential backoff.
Definition: tbb_machine.h:348
tbb::internal::atomic_backoff::pause
void pause()
Pause for a while.
Definition: tbb_machine.h:363
tbb::internal::no_assign
Base class for types that should not be assigned.
Definition: tbb_stddef.h:322
tbb::internal::no_copy
Base class for types that should not be copied or assigned.
Definition: tbb_stddef.h:330
tbb::internal::concurrent_monitor
concurrent_monitor
Definition: concurrent_monitor.h:93
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::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::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::concurrent_monitor::abort_all
void abort_all()
Abort any sleeping threads at the time of the call.
Definition: concurrent_monitor.h:175
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
Definition: concurrent_monitor.h:96
tbb::internal::micro_queue
A queue using simple locking.
Definition: concurrent_queue.cpp:47
tbb::internal::micro_queue::tail_page
atomic< page * > tail_page
Definition: concurrent_queue.cpp:55
tbb::internal::micro_queue::make_invalid
void make_invalid(ticket k)
Definition: concurrent_queue.cpp:314
tbb::internal::micro_queue::assign
micro_queue & assign(const micro_queue &src, concurrent_queue_base &base, concurrent_queue_base::copy_specifics op_type)
Definition: concurrent_queue.cpp:253
tbb::internal::micro_queue::push
void push(const void *item, ticket k, concurrent_queue_base &base, concurrent_queue_base::copy_specifics op_type)
Definition: concurrent_queue.cpp:161
tbb::internal::micro_queue::head_page
atomic< page * > head_page
Definition: concurrent_queue.cpp:52
tbb::internal::micro_queue::abort_push
void abort_push(ticket k, concurrent_queue_base &base)
Definition: concurrent_queue.cpp:226
tbb::internal::micro_queue::head_counter
atomic< ticket > head_counter
Definition: concurrent_queue.cpp:53
tbb::internal::micro_queue::make_copy
page * make_copy(concurrent_queue_base &base, const page *src_page, size_t begin_in_page, size_t end_in_page, ticket &g_index, concurrent_queue_base::copy_specifics op_type)
Definition: concurrent_queue.cpp:295
tbb::internal::micro_queue::pop
bool pop(void *dst, ticket k, concurrent_queue_base &base)
Definition: concurrent_queue.cpp:230
tbb::internal::micro_queue::page_mutex
spin_mutex page_mutex
Definition: concurrent_queue.cpp:58
tbb::internal::micro_queue::tail_counter
atomic< ticket > tail_counter
Definition: concurrent_queue.cpp:56
tbb::internal::micro_queue::page
concurrent_queue_base::page page
Definition: concurrent_queue.cpp:48
tbb::internal::micro_queue_pop_finalizer
Definition: concurrent_queue.cpp:77
tbb::internal::micro_queue_pop_finalizer::my_page
page * my_page
Definition: concurrent_queue.cpp:81
tbb::internal::micro_queue_pop_finalizer::my_queue
micro_queue & my_queue
Definition: concurrent_queue.cpp:80
tbb::internal::micro_queue_pop_finalizer::page
concurrent_queue_base::page page
Definition: concurrent_queue.cpp:78
tbb::internal::micro_queue_pop_finalizer::base
concurrent_queue_base & base
Definition: concurrent_queue.cpp:82
tbb::internal::micro_queue_pop_finalizer::micro_queue_pop_finalizer
micro_queue_pop_finalizer(micro_queue &queue, concurrent_queue_base &b, ticket k, page *p)
Definition: concurrent_queue.cpp:84
tbb::internal::micro_queue_pop_finalizer::~micro_queue_pop_finalizer
~micro_queue_pop_finalizer()
Definition: concurrent_queue.cpp:87
tbb::internal::micro_queue_pop_finalizer::my_ticket
ticket my_ticket
Definition: concurrent_queue.cpp:79
tbb::internal::predicate_leq
Definition: concurrent_queue.cpp:103
tbb::internal::predicate_leq::predicate_leq
predicate_leq(ticket t_)
Definition: concurrent_queue.cpp:105
tbb::internal::predicate_leq::operator()
bool operator()(uintptr_t p) const
Definition: concurrent_queue.cpp:106
tbb::internal::predicate_leq::t
ticket t
Definition: concurrent_queue.cpp:104
tbb::internal::concurrent_queue_rep
Internal representation of a ConcurrentQueue.
Definition: concurrent_queue.cpp:112
tbb::internal::concurrent_queue_rep::array
micro_queue array[n_queue]
Definition: concurrent_queue.cpp:137
tbb::internal::concurrent_queue_rep::items_avail
concurrent_monitor items_avail
Definition: concurrent_queue.cpp:130
tbb::internal::concurrent_queue_rep::index
static size_t index(ticket k)
Map ticket to an array index.
Definition: concurrent_queue.cpp:125
tbb::internal::concurrent_queue_rep::head_counter
atomic< ticket > head_counter
Definition: concurrent_queue.cpp:129
tbb::internal::concurrent_queue_rep::abort_counter
atomic< unsigned > abort_counter
Definition: concurrent_queue.cpp:144
tbb::internal::concurrent_queue_rep::choose
micro_queue & choose(ticket k)
Definition: concurrent_queue.cpp:139
tbb::internal::concurrent_queue_rep::tail_counter
atomic< ticket > tail_counter
Definition: concurrent_queue.cpp:134
tbb::internal::concurrent_queue_rep::pad1
char pad1[NFS_MaxLineSize-((sizeof(atomic< ticket >)+sizeof(concurrent_monitor)+sizeof(atomic< size_t >))&(NFS_MaxLineSize-1))]
Definition: concurrent_queue.cpp:132
tbb::internal::concurrent_queue_rep::slots_avail
concurrent_monitor slots_avail
Definition: concurrent_queue.cpp:135
tbb::internal::concurrent_queue_rep::pad2
char pad2[NFS_MaxLineSize-((sizeof(atomic< ticket >)+sizeof(concurrent_monitor))&(NFS_MaxLineSize-1))]
Definition: concurrent_queue.cpp:136
tbb::internal::concurrent_queue_rep::phi
static const size_t phi
Approximately n_queue/golden ratio.
Definition: concurrent_queue.cpp:118
tbb::internal::concurrent_queue_rep::infinite_capacity
static const ptrdiff_t infinite_capacity
Value for effective_capacity that denotes unbounded queue.
Definition: concurrent_queue.cpp:147
tbb::internal::concurrent_queue_rep::n_queue
static const size_t n_queue
Must be power of 2.
Definition: concurrent_queue.cpp:122
tbb::internal::concurrent_queue_rep::n_invalid_entries
atomic< size_t > n_invalid_entries
Definition: concurrent_queue.cpp:131
tbb::internal::concurrent_queue_iterator_rep
Definition: concurrent_queue.cpp:580
tbb::internal::concurrent_queue_iterator_rep::offset_of_last
const size_t offset_of_last
Definition: concurrent_queue.cpp:584
tbb::internal::concurrent_queue_iterator_rep::get_item
bool get_item(void *&item, size_t k)
Set item to point to kth element. Return true if at end of queue or item is marked valid; false other...
Definition: concurrent_queue.cpp:596
tbb::internal::concurrent_queue_iterator_rep::concurrent_queue_iterator_rep
concurrent_queue_iterator_rep(const concurrent_queue_base &queue, size_t offset_of_last_)
Definition: concurrent_queue.cpp:586
tbb::internal::concurrent_queue_iterator_rep::my_queue
const concurrent_queue_base & my_queue
Definition: concurrent_queue.cpp:583
tbb::internal::concurrent_queue_iterator_rep::head_counter
ticket head_counter
Definition: concurrent_queue.cpp:582
tbb::internal::concurrent_queue_iterator_rep::array
concurrent_queue_base::page * array[concurrent_queue_rep::n_queue]
Definition: concurrent_queue.cpp:585

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.