A queue using simple locking. More...

#include <_concurrent_queue_impl.h>

Inheritance diagram for tbb::strict_ppl::internal::micro_queue< T >:

Collaboration diagram for tbb::strict_ppl::internal::micro_queue< T >:

Classes
class	destroyer
	Class used to ensure exception-safety of method "pop". More...

struct	padded_page

Public Types
typedef void(*	item_constructor_t) (T location, const void src)

Public Member Functions
void	push (const void *item, ticket k, concurrent_queue_base_v3< T > &base, item_constructor_t construct_item)

bool	pop (void *dst, ticket k, concurrent_queue_base_v3< T > &base)

micro_queue &	assign (const micro_queue &src, concurrent_queue_base_v3< T > &base, item_constructor_t construct_item)

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)

void	invalidate_page_and_rethrow (ticket k)

Static Public Member Functions
static T &	get_ref (page &p, size_t index)

Public Attributes
atomic< page * >	head_page

atomic< ticket >	head_counter

atomic< page * >	tail_page

atomic< ticket >	tail_counter

spin_mutex	page_mutex

Private Types
typedef concurrent_queue_rep_base::page	page

Private Member Functions
void	copy_item (page &dst, size_t dindex, const void *src, item_constructor_t construct_item)

void	copy_item (page &dst, size_t dindex, const page &src, size_t sindex, item_constructor_t construct_item)

void	assign_and_destroy_item (void *dst, page &src, size_t index)

void	spin_wait_until_my_turn (atomic< ticket > &counter, ticket k, concurrent_queue_rep_base &rb) const

Private Member Functions inherited from tbb::internal::no_copy
	no_copy (const no_copy &)=delete

	no_copy ()=default

Friends
class	micro_queue_pop_finalizer< T >

Detailed Description

template<typename T>
class tbb::strict_ppl::internal::micro_queue< T >

A queue using simple locking.

For efficiency, this class has no constructor. The caller is expected to zero-initialize it.

Definition at line 58 of file _concurrent_queue_impl.h.

Member Typedef Documentation

◆ item_constructor_t

template<typename T>

typedef void(* tbb::strict_ppl::internal::micro_queue< T >::item_constructor_t) (T *location, const void *src)

Definition at line 133 of file _concurrent_queue_impl.h.

◆ page

template<typename T>

typedef concurrent_queue_rep_base::page tbb::strict_ppl::internal::micro_queue< T >::page

private

Definition at line 135 of file _concurrent_queue_impl.h.

Member Function Documentation

◆ assign()

template<typename T>

micro_queue< T > & tbb::strict_ppl::internal::micro_queue< T >::assign	(	const micro_queue< T > &	src,
		concurrent_queue_base_v3< T > &	base,
		item_constructor_t	construct_item
	)

Definition at line 286 of file _concurrent_queue_impl.h.

 {
     head_counter = src.head_counter;
     tail_counter = src.tail_counter;
 
     const page* srcp = src.head_page;
     if( is_valid_page(srcp) ) {
         ticket g_index = head_counter;
         __TBB_TRY {
             size_t n_items  = (tail_counter-head_counter)/concurrent_queue_rep_base::n_queue;
             size_t index = modulo_power_of_two( head_counter/concurrent_queue_rep_base::n_queue, base.my_rep->items_per_page );
             size_t end_in_first_page = (index+n_items<base.my_rep->items_per_page)?(index+n_items):base.my_rep->items_per_page;
 
             head_page = make_copy( base, srcp, index, end_in_first_page, g_index, construct_item );
             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.my_rep->items_per_page, g_index, construct_item );
                     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_base::n_queue, base.my_rep->items_per_page );
                 if( last_index==0 ) last_index = base.my_rep->items_per_page;
 
                 cur_page->next = make_copy( base, srcp, 0, last_index, g_index, construct_item );
                 cur_page = cur_page->next;
             }
             tail_page = cur_page;
         } __TBB_CATCH (...) {
             invalidate_page_and_rethrow( g_index );
         }
     } else {
         head_page = tail_page = NULL;
     }
     return *this;
 }

◆ assign_and_destroy_item()

template<typename T>

void tbb::strict_ppl::internal::micro_queue< T >::assign_and_destroy_item	(	void *	dst,
		page &	src,
		size_t	index
	)

inlineprivate

Definition at line 156 of file _concurrent_queue_impl.h.

                                                                        {
         T& from = get_ref(src,index);
         destroyer d(from);
         *static_cast<T*>(dst) = tbb::internal::move( from );
     }

◆ copy_item() [1/2]

template<typename T>

void tbb::strict_ppl::internal::micro_queue< T >::copy_item	(	page &	dst,
		size_t	dindex,
		const void *	src,
		item_constructor_t	construct_item
	)

inlineprivate

Definition at line 145 of file _concurrent_queue_impl.h.

                                                                                                    {
         construct_item( &get_ref(dst, dindex), src );
     }

◆ copy_item() [2/2]

template<typename T>

void tbb::strict_ppl::internal::micro_queue< T >::copy_item	(	page &	dst,
		size_t	dindex,
		const page &	src,
		size_t	sindex,
		item_constructor_t	construct_item
	)

inlineprivate

Definition at line 149 of file _concurrent_queue_impl.h.

     {
         T& src_item = get_ref( const_cast<page&>(src), sindex );
         construct_item( &get_ref(dst, dindex), static_cast<const void*>(&src_item) );
     }

◆ get_ref()

template<typename T>

static T& tbb::strict_ppl::internal::micro_queue< T >::get_ref	(	page &	p,
		size_t	index
	)

inlinestatic

Definition at line 176 of file _concurrent_queue_impl.h.

                                                {
         return (&static_cast<padded_page*>(static_cast<void*>(&p))->last)[index];
     }

Referenced by tbb::strict_ppl::internal::concurrent_queue_iterator_rep< Value >::get_item().

Here is the caller graph for this function:

◆ invalidate_page_and_rethrow()

template<typename T >

void tbb::strict_ppl::internal::micro_queue< T >::invalidate_page_and_rethrow ( ticket k )

Definition at line 327 of file _concurrent_queue_impl.h.

                                                            {
     // Append an invalid page at address 1 so that no more pushes are allowed.
     page* invalid_page = (page*)uintptr_t(1);
     {
         spin_mutex::scoped_lock lock( page_mutex );
         itt_store_word_with_release(tail_counter, k+concurrent_queue_rep_base::n_queue+1);
         page* q = tail_page;
         if( is_valid_page(q) )
             q->next = invalid_page;
         else
             head_page = invalid_page;
         tail_page = invalid_page;
     }
     __TBB_RETHROW();
 }

◆ make_copy()

template<typename T>

concurrent_queue_rep_base::page * tbb::strict_ppl::internal::micro_queue< T >::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
	)

Definition at line 344 of file _concurrent_queue_impl.h.

 {
     concurrent_queue_page_allocator& pa = base;
     page* new_page = pa.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 )
             copy_item( *new_page, begin_in_page, *src_page, begin_in_page, construct_item );
     return new_page;
 }

◆ pop()

template<typename T>

bool tbb::strict_ppl::internal::micro_queue< T >::pop	(	void *	dst,
		ticket	k,
		concurrent_queue_base_v3< T > &	base
	)

Definition at line 263 of file _concurrent_queue_impl.h.

                                                                                  {
     k &= -concurrent_queue_rep_base::n_queue;
     if( head_counter!=k ) spin_wait_until_eq( head_counter, k );
     call_itt_notify(acquired, &head_counter);
     if( tail_counter==k ) spin_wait_while_eq( tail_counter, k );
     call_itt_notify(acquired, &tail_counter);
     page *p = head_page;
     __TBB_ASSERT( p, NULL );
     size_t index = modulo_power_of_two( k/concurrent_queue_rep_base::n_queue, base.my_rep->items_per_page );
     bool success = false;
     {
         micro_queue_pop_finalizer<T> finalizer( *this, base, k+concurrent_queue_rep_base::n_queue, index==base.my_rep->items_per_page-1 ? p : NULL );
         if( p->mask & uintptr_t(1)<<index ) {
             success = true;
             assign_and_destroy_item( dst, *p, index );
         } else {
             --base.my_rep->n_invalid_entries;
         }
     }
     return success;
 }

◆ push()

template<typename T>

void tbb::strict_ppl::internal::micro_queue< T >::push	(	const void *	item,
		ticket	k,
		concurrent_queue_base_v3< T > &	base,
		item_constructor_t	construct_item
	)

Definition at line 215 of file _concurrent_queue_impl.h.

 {
     k &= -concurrent_queue_rep_base::n_queue;
     page* p = NULL;
     size_t index = modulo_power_of_two( k/concurrent_queue_rep_base::n_queue, base.my_rep->items_per_page);
     if( !index ) {
         __TBB_TRY {
             concurrent_queue_page_allocator& pa = base;
             p = pa.allocate_page();
         } __TBB_CATCH (...) {
             ++base.my_rep->n_invalid_entries;
             invalidate_page_and_rethrow( k );
         }
         p->mask = 0;
         p->next = NULL;
     }
 
     if( tail_counter != k ) spin_wait_until_my_turn( tail_counter, k, *base.my_rep );
     call_itt_notify(acquired, &tail_counter);
 
     if( p ) {
         spin_mutex::scoped_lock lock( page_mutex );
         page* q = tail_page;
         if( is_valid_page(q) )
             q->next = p;
         else
             head_page = p;
         tail_page = p;
     } else {
         p = tail_page;
     }
 
     __TBB_TRY {
         copy_item( *p, index, item, construct_item );
         // If no exception was thrown, mark item as present.
         itt_hide_store_word(p->mask,  p->mask | uintptr_t(1)<<index);
         call_itt_notify(releasing, &tail_counter);
         tail_counter += concurrent_queue_rep_base::n_queue;
     } __TBB_CATCH (...) {
         ++base.my_rep->n_invalid_entries;
         call_itt_notify(releasing, &tail_counter);
         tail_counter += concurrent_queue_rep_base::n_queue;
         __TBB_RETHROW();
     }
 }

◆ spin_wait_until_my_turn()

template<typename T >

void tbb::strict_ppl::internal::micro_queue< T >::spin_wait_until_my_turn	(	atomic< ticket > &	counter,
		ticket	k,
		concurrent_queue_rep_base &	rb
	)		const

private

Definition at line 203 of file _concurrent_queue_impl.h.

                                                                                                                      {
     for( atomic_backoff b(true);;b.pause() ) {
         ticket c = counter;
         if( c==k ) return;
         else if( c&1 ) {
             ++rb.n_invalid_entries;
             throw_exception( eid_bad_last_alloc );
         }
     }
 }