21 #define __TBB_concurrent_queue_H 28 #if defined(_MSC_VER) && defined(_Wp64) 30 #pragma warning (disable: 4267) 33 #define RECORD_EVENTS 0 60 void push(
const void* item, ticket k, concurrent_queue_base& base,
63 void abort_push( ticket k, concurrent_queue_base& base );
65 bool pop(
void* dst, ticket k, concurrent_queue_base& base );
70 page*
make_copy ( concurrent_queue_base& base,
const page* src_page,
size_t begin_in_page,
82 concurrent_queue_base &
base;
85 my_ticket(k), my_queue(queue), my_page(p), base(b)
106 bool operator() ( uintptr_t
p )
const {
return (ticket)p<=t;}
118 static const size_t phi = 3;
122 static const size_t n_queue = 8;
126 return k*phi%n_queue;
141 return array[index(k)];
147 static const ptrdiff_t infinite_capacity = ptrdiff_t(~
size_t(0)/2);
150 #if _MSC_VER && !defined(__INTEL_COMPILER) 152 #pragma warning( push ) 153 #pragma warning( disable: 4146 ) 180 if( tail_counter!=k )
184 else if( tail&0x1 ) {
217 p->
mask |= uintptr_t(1)<<index;
237 bool success =
false;
240 if( p->
mask & uintptr_t(1)<<index ) {
273 cur_page = cur_page->
next;
281 cur_page->
next =
make_copy( base, srcp, 0, last_index, g_index, op_type );
282 cur_page = cur_page->
next;
300 new_page->
next = NULL;
302 for( ; begin_in_page!=end_in_page; ++begin_in_page, ++g_index )
303 if( new_page->
mask & uintptr_t(1)<<begin_in_page ) {
305 base.
copy_page_item( *new_page, begin_in_page, *src_page, begin_in_page );
318 static_invalid_page = &dummy;
330 #if _MSC_VER && !defined(__INTEL_COMPILER) 331 #pragma warning( pop ) 332 #endif // warning 4146 is back 338 items_per_page = item_sz<= 8 ? 32 :
344 my_capacity = size_t(-1)/(item_sz>1 ? item_sz : 2);
353 this->item_size = item_sz;
357 size_t nq = my_rep->n_queue;
358 for(
size_t i=0; i<nq; i++ )
359 __TBB_ASSERT( my_rep->array[i].tail_page==NULL,
"pages were not freed properly" );
364 internal_insert_item( src, copy );
368 internal_insert_item( src,
move );
375 ptrdiff_t e = my_capacity;
377 bool sync_prepare_done =
false;
381 if( !sync_prepare_done ) {
382 ITT_NOTIFY( sync_prepare, &sync_prepare_done );
383 sync_prepare_done =
true;
389 while( (ptrdiff_t)(k-r.
head_counter)>=const_cast<volatile ptrdiff_t&>(e = my_capacity) ) {
402 if (slept ==
true)
break;
408 ITT_NOTIFY( sync_acquired, &sync_prepare_done );
418 bool sync_prepare_done =
false;
426 if( !sync_prepare_done ) {
428 sync_prepare_done =
true;
447 if (slept ==
true)
break;
454 }
while( !r.
choose(k).
pop(dst,k,*
this) );
484 }
while( !r.
choose( k ).
pop( dst, k, *
this ) );
492 return internal_insert_if_not_full( src, copy );
496 return internal_insert_if_not_full( src,
move );
520 __TBB_ASSERT(
sizeof(ptrdiff_t)<=
sizeof(
size_t), NULL );
521 return ptrdiff_t(my_rep->tail_counter-my_rep->head_counter-my_rep->n_invalid_entries);
525 ticket tc = my_rep->tail_counter;
526 ticket hc = my_rep->head_counter;
528 return ( tc==my_rep->tail_counter && ptrdiff_t(tc-hc-my_rep->n_invalid_entries)<=0 );
536 size_t nq = my_rep->n_queue;
537 for(
size_t i=0; i<nq; ++i ) {
538 page* tp = my_rep->array[i].tail_page;
539 __TBB_ASSERT( my_rep->array[i].head_page==tp,
"at most one page should remain" );
541 if( tp!=static_invalid_page ) deallocate_page( tp );
542 my_rep->array[i].tail_page = NULL;
562 for(
size_t i = 0; i<my_rep->n_queue; ++i )
563 my_rep->array[i].assign( src.
my_rep->
array[i], *
this, op_type );
566 "the source concurrent queue should not be concurrently modified." );
570 internal_assign( src, copy );
574 internal_assign( src,
move );
587 head_counter(queue.my_rep->head_counter),
589 offset_of_last(offset_of_last_)
604 item =
static_cast<unsigned char*
>(
static_cast<void*
>(
p)) + offset_of_last + my_queue.
item_size*i;
605 return (p->
mask & uintptr_t(1)<<i)!=0;
618 if( !my_rep->get_item(my_item, k) ) advance();
626 initialize(queue,offset_of_last);
630 if( my_rep!=other.
my_rep ) {
644 __TBB_ASSERT( my_item,
"attempt to increment iterator past end of queue" );
645 size_t k = my_rep->head_counter;
649 my_rep->get_item(tmp,k);
658 my_rep->head_counter = ++k;
659 if( !my_rep->get_item(my_item, k) ) advance();
micro_queue array[n_queue]
void internal_assign(const concurrent_queue_base_v3 &src, copy_specifics op_type)
Assigns one queue to another using specified operation (copy or move)
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)
atomic< ticket > head_counter
atomic< unsigned > abort_counter
atomic< page * > head_page
void __TBB_EXPORTED_METHOD assign(const concurrent_queue_base_v3 &src)
copy internal representation
atomic< page * > tail_page
virtual void copy_item(page &dst, size_t index, const void *src)=0
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
void move(tbb_thread &t1, tbb_thread &t2)
const size_t NFS_MaxLineSize
Compile-time constant that is upper bound on cache line/sector size.
bool pop(void *dst, ticket k, concurrent_queue_base &base)
__TBB_EXPORTED_METHOD concurrent_queue_base_v3(size_t item_size)
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
Represents acquisition of a mutex.
bool __TBB_EXPORTED_METHOD internal_push_move_if_not_full(const void *src)
Attempt to enqueue item onto queue using move operation.
concurrent_queue_iterator_rep(const concurrent_queue_base &queue, size_t offset_of_last_)
micro_queue_pop_finalizer(micro_queue &queue, concurrent_queue_base &b, ticket k, page *p)
void __TBB_EXPORTED_METHOD internal_throw_exception() const
throw an exception
void pause()
Pause for a while.
size_t items_per_page
Always a power of 2.
virtual __TBB_EXPORTED_METHOD ~concurrent_queue_base_v3()
Class that implements exponential backoff.
ptrdiff_t my_capacity
Capacity of the queue.
void __TBB_EXPORTED_METHOD internal_abort()
Abort all pending queue operations.
Internal representation of a ConcurrentQueue.
bool commit_wait(thread_context &thr)
Commit wait if event count has not changed; otherwise, cancel wait.
Base class for types that should not be assigned.
Exception for user-initiated abort.
micro_queue & assign(const micro_queue &src, concurrent_queue_base &base, concurrent_queue_base::copy_specifics op_type)
size_t item_size
Size of an item.
A queue using simple locking.
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_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
void make_invalid(ticket k)
A lock that occupies a single byte.
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...
concurrent_queue_rep * my_rep
Internal representation.
virtual void copy_page_item(page &dst, size_t dindex, const page &src, size_t sindex)=0
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)
~micro_queue_pop_finalizer()
micro_queue & choose(ticket k)
concurrent_queue_base::page page
void __TBB_EXPORTED_METHOD internal_push(const void *src)
Enqueue item at tail of queue using copy operation.
void abort_all()
Abort any sleeping threads at the time of the call.
atomic< ticket > tail_counter
void push(const void *item, ticket k, concurrent_queue_base &base, concurrent_queue_base::copy_specifics op_type)
atomic< ticket > head_counter
bool is_aligned(T *pointer, uintptr_t alignment)
A function to check if passed in pointer is aligned on a specific border.
size_t __TBB_EXPORTED_FUNC NFS_GetLineSize()
Cache/sector line size.
void throw_exception(exception_id eid)
Versionless convenience wrapper for throw_exception_v4()
Meets "allocator" requirements of ISO C++ Standard, Section 20.1.5.
void __TBB_EXPORTED_METHOD internal_push_move(const void *src)
Enqueue item at tail of queue using move operation.
Type-independent portion of concurrent_queue_iterator.
concurrent_queue_base_v3 concurrent_queue_base
void prepare_wait(thread_context &thr, uintptr_t ctx=0)
prepare wait by inserting 'thr' into the wait queue
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.
#define ITT_NOTIFY(name, obj)
const size_t offset_of_last
atomic< ticket > tail_counter
bool __TBB_EXPORTED_METHOD internal_push_if_not_full(const void *src)
Attempt to enqueue item onto queue using copy operation.
void spin_wait_while_eq(const volatile T &location, U value)
Spin WHILE the value of the variable is equal to a given value.
void __TBB_EXPORTED_METHOD internal_finish_clear()
free any remaining pages
Base class for types that should not be copied or assigned.
void cancel_wait(thread_context &thr)
Cancel the wait. Removes the thread from the wait queue if not removed yet.
virtual void assign_and_destroy_item(void *dst, page &src, size_t index)=0
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
void __TBB_EXPORTED_METHOD advance()
Advance iterator one step towards tail of queue.
static const size_t n_queue
Must be power of 2.
void __TBB_EXPORTED_METHOD move_content(concurrent_queue_base_v8 &src)
move items
atomic< size_t > n_invalid_entries
void internal_insert_item(const void *src, copy_specifics op_type)
Enqueues item at tail of queue using specified operation (copy or move)
void __TBB_EXPORTED_METHOD internal_pop(void *dst)
Dequeue item from head of queue.
concurrent_queue_base::page page
void const char const char int ITT_FORMAT __itt_group_sync p
static const ptrdiff_t infinite_capacity
Value for effective_capacity that denotes unbounded queue.
static size_t index(ticket k)
Map ticket to an array index.
bool __TBB_EXPORTED_METHOD internal_pop_if_present(void *dst)
Attempt to dequeue item from queue.
concurrent_monitor slots_avail
concurrent_queue_iterator_base_v3()
Default constructor.
virtual void deallocate_page(page *p)=0
custom de-allocator
void abort_push(ticket k, concurrent_queue_base &base)
concurrent_monitor items_avail
void notify(const P &predicate)
Notify waiting threads of the event that satisfies the given predicate.
void * my_item
Pointer to current item.
bool __TBB_EXPORTED_METHOD internal_empty() const
Check if the queue is empty.
static void * static_invalid_page
void __TBB_EXPORTED_METHOD internal_set_capacity(ptrdiff_t capacity, size_t element_size)
Set the queue capacity.
void __TBB_EXPORTED_METHOD assign(const concurrent_queue_iterator_base_v3 &i)
Assignment.
__TBB_EXPORTED_METHOD ~concurrent_queue_iterator_base_v3()
Destructor.
concurrent_queue_base & base
const concurrent_queue_base & my_queue
ptrdiff_t __TBB_EXPORTED_METHOD internal_size() const
Get size of queue.
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.
void initialize(const concurrent_queue_base_v3 &queue, size_t offset_of_data)
virtual page * allocate_page()=0
custom allocator
concurrent_queue_iterator_rep * my_rep
concurrent_queue over which we are iterating.