21 #ifndef __TBB__concurrent_queue_impl_H 22 #define __TBB__concurrent_queue_impl_H 24 #ifndef __TBB_concurrent_queue_H 25 #error Do not #include this internal file directly; use public TBB headers instead. 28 #include "../tbb_stddef.h" 29 #include "../tbb_machine.h" 30 #include "../atomic.h" 31 #include "../spin_mutex.h" 32 #include "../cache_aligned_allocator.h" 33 #include "../tbb_exception.h" 34 #include "../tbb_profiling.h" 36 #include __TBB_STD_SWAP_HEADER 41 #if !__TBB_TEMPLATE_FRIENDS_BROKEN 44 namespace strict_ppl {
45 template<
typename T,
typename A>
class concurrent_queue;
48 template<
typename T,
typename A>
class concurrent_bounded_queue;
53 namespace strict_ppl {
77 static const size_t phi = 3;
81 static const size_t n_queue = 8;
107 return uintptr_t(
p)>1;
125 #if _MSC_VER && !defined(__INTEL_COMPILER) 127 #pragma warning( push ) 128 #pragma warning( disable: 4146 ) 137 typedef void (*item_constructor_t)(T* location,
const void* src);
149 void copy_item(
page& dst,
size_t dindex,
const void* src, item_constructor_t construct_item ) {
150 construct_item( &get_ref(dst, dindex), src );
154 item_constructor_t construct_item )
156 T& src_item = get_ref( const_cast<page&>(src), sindex );
157 construct_item( &get_ref(dst, dindex), static_cast<const void*>(&src_item) );
161 T& from = get_ref(src,index);
181 return (&static_cast<padded_page*>(static_cast<void*>(&
p))->
last)[index];
193 item_constructor_t construct_item ) ;
198 item_constructor_t construct_item ) ;
201 size_t end_in_page,
ticket& g_index, item_constructor_t construct_item ) ;
203 void invalidate_page_and_rethrow(
ticket k ) ;
220 item_constructor_t construct_item )
230 ++base.
my_rep->n_invalid_entries;
231 invalidate_page_and_rethrow( k );
237 if( tail_counter != k ) spin_wait_until_my_turn( tail_counter, k, *base.
my_rep );
253 copy_item( *
p, index, item, construct_item );
259 ++base.
my_rep->n_invalid_entries;
276 bool success =
false;
279 if(
p->mask & uintptr_t(1)<<index ) {
281 assign_and_destroy_item( dst, *
p, index );
283 --base.
my_rep->n_invalid_entries;
291 item_constructor_t construct_item )
298 ticket g_index = head_counter;
302 size_t end_in_first_page = (index+n_items<base.
my_rep->items_per_page)?(index+n_items):base.
my_rep->items_per_page;
304 head_page = make_copy( base, srcp, index, end_in_first_page, g_index, construct_item );
305 page* cur_page = head_page;
309 cur_page->
next = make_copy( base, srcp, 0, base.
my_rep->items_per_page, g_index, construct_item );
310 cur_page = cur_page->
next;
315 if( last_index==0 ) last_index = base.
my_rep->items_per_page;
317 cur_page->
next = make_copy( base, srcp, 0, last_index, g_index, construct_item );
318 cur_page = cur_page->
next;
320 tail_page = cur_page;
322 invalidate_page_and_rethrow( g_index );
325 head_page = tail_page = NULL;
333 page* invalid_page = (
page*)uintptr_t(1);
339 q->
next = invalid_page;
341 head_page = invalid_page;
342 tail_page = invalid_page;
350 ticket& g_index, item_constructor_t construct_item )
354 new_page->
next = NULL;
356 for( ; begin_in_page!=end_in_page; ++begin_in_page, ++g_index )
357 if( new_page->
mask & uintptr_t(1)<<begin_in_page )
358 copy_item( *new_page, begin_in_page, *src_page, begin_in_page, construct_item );
371 my_ticket(k), my_queue(queue), my_page(
p), allocator(b)
382 my_queue.head_page = q;
384 my_queue.tail_page = NULL;
389 allocator.deallocate_page(
p );
393 #if _MSC_VER && !defined(__INTEL_COMPILER) 394 #pragma warning( pop ) 395 #endif // warning 4146 is back 410 return k*phi%n_queue;
415 return array[index(k)];
445 return reinterpret_cast<page*>(allocate_block ( n ));
451 deallocate_block( reinterpret_cast<void*>(
p), n );
455 virtual void *allocate_block(
size_t n ) = 0;
458 virtual void deallocate_block(
void *
p,
size_t n ) = 0;
466 for(
size_t i=0; i<nq; i++ )
467 __TBB_ASSERT( my_rep->
array[i].tail_page==NULL,
"pages were not freed properly" );
476 r.
choose(k).push( src, k, *
this, construct_item );
481 bool internal_try_pop(
void* dst ) ;
484 size_t internal_size()
const ;
487 bool internal_empty()
const ;
491 void internal_finish_clear() ;
501 #if __TBB_CPP11_RVALUE_REF_PRESENT 511 const size_t item_size =
sizeof(T);
518 my_rep->item_size = item_size;
519 my_rep->items_per_page = item_size<= 8 ? 32 :
520 item_size<= 16 ? 16 :
540 #if defined(_MSC_VER) && defined(_Wp64) 541 #pragma warning (push) 542 #pragma warning (disable: 4267) 545 #if defined(_MSC_VER) && defined(_Wp64) 546 #pragma warning (pop) 552 }
while( !r.
choose( k ).pop( dst, k, *
this ) );
559 __TBB_ASSERT(
sizeof(ptrdiff_t)<=
sizeof(
size_t), NULL );
564 ptrdiff_t sz = tc-hc-nie;
565 return sz<0 ? 0 : size_t(sz);
581 for(
size_t i=0; i<nq; ++i ) {
585 deallocate_page( tp );
586 r.
array[i].tail_page = NULL;
605 for(
size_t i = 0; i < r.
n_queue; ++i )
606 r.
array[i].assign( src.
my_rep->array[i], *
this, construct_item);
609 "the source concurrent queue should not be concurrently modified." );
622 head_counter(queue.my_rep->head_counter),
625 for(
size_t k=0; k<concurrent_queue_rep<T>::n_queue; ++k )
626 array[k] = queue.
my_rep->array[k].head_page;
630 bool get_item( T*& item,
size_t k ) ;
635 if( k==my_queue.my_rep->tail_counter ) {
643 return (
p->mask & uintptr_t(1)<<i)!=0;
649 template<
typename Value>
655 template<
typename C,
typename T,
typename U>
658 template<
typename C,
typename T,
typename U>
666 #if __TBB_GCC_OPTIMIZER_ORDERING_BROKEN 673 :
no_assign(), my_rep(NULL), my_item(NULL) {
693 template<
typename Value>
698 if( !my_rep->get_item(my_item, k) ) advance();
701 template<
typename Value>
703 if( my_rep!=other.
my_rep ) {
716 template<
typename Value>
718 __TBB_ASSERT( my_item,
"attempt to increment iterator past end of queue" );
719 size_t k = my_rep->head_counter;
723 my_rep->get_item(tmp,k);
732 my_rep->head_counter = ++k;
733 if( !my_rep->get_item(my_item, k) ) advance();
746 template<
typename Container,
typename Value>
748 public std::iterator<std::forward_iterator_tag,Value> {
749 #if !__TBB_TEMPLATE_FRIENDS_BROKEN 750 template<
typename T,
class A>
751 friend class ::tbb::strict_ppl::concurrent_queue;
778 return *static_cast<Value*>(this->my_item);
791 Value* result = &operator*();
798 template<
typename C,
typename T,
typename U>
803 template<
typename C,
typename T,
typename U>
853 #if __TBB_PROTECTED_NESTED_CLASS_BROKEN 867 virtual void copy_item(
page& dst,
size_t index,
const void* src ) = 0;
868 virtual void assign_and_destroy_item(
void* dst,
page& src,
size_t index ) = 0;
914 #if __TBB_CPP11_RVALUE_REF_PRESENT 925 void internal_insert_item(
const void* src, copy_specifics op_type );
928 bool internal_insert_if_not_full(
const void* src, copy_specifics op_type );
933 virtual void copy_page_item(
page& dst,
size_t dindex,
const page& src,
size_t sindex ) = 0;
954 virtual void move_item(
page& dst,
size_t index,
const void* src ) = 0;
964 template<
typename C,
typename T,
typename U>
967 template<
typename C,
typename T,
typename U>
1005 template<
typename Container,
typename Value>
1007 public std::iterator<std::forward_iterator_tag,Value> {
1009 #if !__TBB_TEMPLATE_FRIENDS_BROKEN 1010 template<
typename T,
class A>
1011 friend class ::tbb::concurrent_bounded_queue;
1039 return *static_cast<Value*>(
my_item);
1059 template<
typename C,
typename T,
typename U>
1064 template<
typename C,
typename T,
typename U>
virtual concurrent_queue_rep_base::page * allocate_page()=0
friend bool operator!=(const concurrent_queue_iterator< C, T > &i, const concurrent_queue_iterator< C, U > &j)
Meets requirements of a forward iterator for STL.
void invalidate_page_and_rethrow(ticket k)
size_t items_per_page
Always a power of 2.
void push(const void *item, ticket k, concurrent_queue_base_v3< T > &base, item_constructor_t construct_item)
bool __TBB_EXPORTED_METHOD internal_push_move_if_not_full(const void *src)
Attempt to enqueue item onto queue using move operation.
void internal_throw_exception() const
Obsolete.
concurrent_queue_iterator(const concurrent_queue_iterator< Container, typename Container::value_type > &other)
size_t __TBB_EXPORTED_FUNC NFS_GetLineSize()
Cache/sector line size.
Value * operator++(int)
Post increment.
Meets "allocator" requirements of ISO C++ Standard, Section 20.1.5.
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.
size_t items_per_page
Always a power of 2.
Internal representation of a ConcurrentQueue.
A queue using simple locking.
#define __TBB_ASSERT(predicate, comment)
No-op version of __TBB_ASSERT.
concurrent_queue_iterator_base_v3()
Default constructor.
Value * my_item
Pointer to current item.
concurrent_queue_iterator(const concurrent_queue_iterator< Container, typename Container::value_type > &other)
void itt_store_word_with_release(tbb::atomic< T > &dst, U src)
Value & operator*() const
Reference to current item.
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
atomic< ticket > tail_counter
atomic< page * > tail_page
ptrdiff_t my_capacity
Capacity of the queue.
size_t internal_size() const
Get size of queue; result may be invalid if queue is modified concurrently.
base class of concurrent_queue
void copy_item(page &dst, size_t dindex, const void *src, item_constructor_t construct_item)
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 type
T last
Must be last field.
Base class for types that should not be copied or assigned.
micro_queue< T > array[n_queue]
Value * operator->() const
Value * operator++(int)
Post increment.
size_t item_size
Size of an item.
void pause()
Pause for a while.
void swap(atomic< T > &lhs, atomic< T > &rhs)
atomic< ticket > head_counter
#define __TBB_offsetof(class_name, member_name)
Extended variant of the standard offsetof macro.
Constness-independent portion of concurrent_queue_iterator.
virtual ~concurrent_queue_page_allocator()
virtual ~concurrent_queue_base_v3()
Class used to ensure exception-safety of method "pop".
micro_queue< T >::padded_page padded_page
void internal_finish_clear()
free any remaining pages
friend bool operator==(const concurrent_queue_iterator< C, T > &i, const concurrent_queue_iterator< C, U > &j)
void assign_and_destroy_item(void *dst, page &src, size_t index)
parts of concurrent_queue_rep that do not have references to micro_queue
void assign(const concurrent_queue_base_v3 &src, item_constructor_t construct_item)
copy or move internal representation
concurrent_queue_rep< T >::page page
void const char const char int ITT_FORMAT __itt_group_sync p
void __TBB_EXPORTED_METHOD internal_push_move(const void *src)
Enqueue item at tail of queue using move operation.
A lock that occupies a single byte.
void copy_item(page &dst, size_t dindex, const page &src, size_t sindex, item_constructor_t construct_item)
Base class for types that should not be assigned.
void throw_exception(exception_id eid)
Versionless convenience wrapper for throw_exception_v4()
void assign(const concurrent_queue_iterator_base_v3< Value > &other)
Assignment.
auto last(Container &c) -> decltype(begin(c))
Similar to C++0x std::remove_cv.
bool internal_empty() const
check if the queue is empty; thread safe
#define __TBB_EXPORTED_METHOD
~concurrent_queue_iterator_base_v3()
Destructor.
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
concurrent_queue_base_v8(size_t item_sz)
micro_queue< T >::item_constructor_t item_constructor_t
concurrent_queue_page_allocator & allocator
Value * operator->() const
Value & operator*() const
Reference to current item.
void * my_item
Pointer to current item.
atomic< page * > head_page
#define __TBB_compiler_fence()
concurrent_queue_iterator_rep(const concurrent_queue_base_v3< T > &queue)
concurrent_queue_iterator()
const concurrent_queue_base_v3< T > & my_queue
void spin_wait_while_eq(const volatile T &location, U value)
Spin WHILE the value of the variable is equal to a given value.
virtual page * allocate_page() __TBB_override
A queue using simple locking.
micro_queue< T > & my_queue
Represents acquisition of a mutex.
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 value
bool get_item(T *&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...
void __TBB_EXPORTED_METHOD move_content(concurrent_queue_base_v8 &src)
move items
concurrent_queue_iterator & operator++()
Advance to next item in queue.
void call_itt_notify(notify_type, void *)
micro_queue< T >::padded_page padded_page
Meets requirements of a forward iterator for STL.
void advance()
Advance iterator one step towards tail of queue.
concurrent_queue_iterator & operator=(const concurrent_queue_iterator &other)
Iterator assignment.
void operator=(const padded_page &)
Not defined anywhere - exists to quiet warnings.
micro_queue_pop_finalizer(micro_queue< T > &queue, concurrent_queue_base_v3< T > &b, ticket k, page *p)
Class that implements exponential backoff.
atomic< ticket > head_counter
void move(tbb_thread &t1, tbb_thread &t2)
concurrent_queue_iterator & operator++()
Advance to next item in queue.
bool internal_try_pop(void *dst)
Attempt to dequeue item from 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.
concurrent_queue_iterator_base_v3(const concurrent_queue_iterator_base_v3 &i)
Copy constructor.
const size_t NFS_MaxLineSize
Compile-time constant that is upper bound on cache line/sector size.
virtual void move_item(page &dst, size_t index, const void *src)=0
Type-independent portion of concurrent_queue_iterator.
void itt_hide_store_word(T &dst, T src)
micro_queue< T > & choose(ticket k)
concurrent_queue_rep * my_rep
Internal representation.
concurrent_queue_rep_base::page page
micro_queue & assign(const micro_queue &src, concurrent_queue_base_v3< T > &base, item_constructor_t construct_item)
concurrent_queue_iterator & operator=(const concurrent_queue_iterator &other)
Iterator assignment.
atomic< ticket > tail_counter
concurrent_queue_iterator_base_v3()
Default constructor.
value_type compare_and_swap(value_type value, value_type comparand)
concurrent_queue_iterator(const concurrent_queue_base_v3 &queue)
Construct iterator pointing to head of queue.
static const size_t n_queue
virtual void move_page_item(page &dst, size_t dindex, const page &src, size_t sindex)=0
page * make_copy(concurrent_queue_base_v3< T > &base, const page *src_page, size_t begin_in_page, size_t end_in_page, ticket &g_index, item_constructor_t construct_item)
concurrent_queue_rep_base::page page
concurrent_queue_iterator_base_v3(const concurrent_queue_iterator_base_v3 &i)
Copy constructor.
void internal_push(const void *src, item_constructor_t construct_item)
Enqueue item at tail of queue.
concurrent_queue_base_v3()
~micro_queue_pop_finalizer()
bool is_valid_page(const concurrent_queue_rep_base::page *p)
atomic< size_t > n_invalid_entries
number of invalid entries in the queue
bool pop(void *dst, ticket k, concurrent_queue_base_v3< T > &base)
Identifiers declared inside namespace internal should never be used directly by client code.
concurrent_queue_iterator()
bool operator!=(const concurrent_queue_iterator< C, T > &i, const concurrent_queue_iterator< C, U > &j)
Abstract class to define interface for page allocation/deallocation.
void spin_wait_until_my_turn(atomic< ticket > &counter, ticket k, concurrent_queue_rep_base &rb) const
T last
Must be last field.
virtual void deallocate_page(concurrent_queue_rep_base::page *p) __TBB_override
size_t item_size
Size of an item.
representation of concurrent_queue_base
static size_t index(ticket k)
Map ticket to an array index.
concurrent_queue_iterator_rep< Value > * my_rep
Represents concurrent_queue over which we are iterating.
void internal_swap(concurrent_queue_base_v3 &src)
swap internal representation
static T & get_ref(page &p, size_t index)
bool operator==(const concurrent_queue_iterator< C, T > &i, const concurrent_queue_iterator< C, U > &j)
concurrent_queue_iterator_base_v3 concurrent_queue_iterator_base
concurrent_queue_rep< T > * my_rep
Internal representation.
concurrent_queue_iterator_rep * my_rep
concurrent_queue over which we are iterating.
padded_page()
Not defined anywhere - exists to quiet warnings.