A queue using simple locking. More...

Collaboration diagram for tbb::internal::micro_queue:

Public Types
typedef concurrent_queue_base::page	page

Public Member Functions
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)

Public Attributes
atomic< page * >	head_page

atomic< ticket >	head_counter

atomic< page * >	tail_page

atomic< ticket >	tail_counter

spin_mutex	page_mutex

Friends
class	micro_queue_pop_finalizer

Detailed Description

A queue using simple locking.

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

Definition at line 47 of file concurrent_queue.cpp.

Member Typedef Documentation

◆ page

typedef concurrent_queue_base::page tbb::internal::micro_queue::page

Definition at line 48 of file concurrent_queue.cpp.

Member Function Documentation

◆ abort_push()

void tbb::internal::micro_queue::abort_push	(	ticket	k,
		concurrent_queue_base &	base
	)

Definition at line 226 of file concurrent_queue.cpp.

                                                                     {
     push(NULL, k, base, concurrent_queue_base::copy);
 }

References tbb::internal::concurrent_queue_base_v3::copy, and push().

Referenced by tbb::internal::concurrent_queue_base_v3::internal_insert_item().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ assign()

micro_queue & tbb::internal::micro_queue::assign	(	const micro_queue &	src,
		concurrent_queue_base &	base,
		concurrent_queue_base::copy_specifics	op_type
	)

Definition at line 253 of file concurrent_queue.cpp.

 {
     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;
 }

References __TBB_ASSERT, __TBB_CATCH, __TBB_RETHROW, __TBB_TRY, head_counter, head_page, tbb::internal::concurrent_queue_base_v3::items_per_page, make_copy(), make_invalid(), tbb::internal::modulo_power_of_two(), tbb::internal::concurrent_queue_rep::n_queue, tbb::internal::concurrent_queue_base_v3::page::next, tail_counter, and tail_page.

Referenced by tbb::internal::concurrent_queue_base_v3::internal_assign().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ make_copy()

concurrent_queue_base::page * tbb::internal::micro_queue::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 at line 295 of file concurrent_queue.cpp.

 {
     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;
 }

References __TBB_ASSERT, tbb::internal::concurrent_queue_base_v3::allocate_page(), tbb::internal::concurrent_queue_base_v3::copy, tbb::internal::concurrent_queue_base_v3::copy_page_item(), tbb::internal::concurrent_queue_base_v3::page::mask, tbb::internal::concurrent_queue_base_v3::move, and tbb::internal::concurrent_queue_base_v3::page::next.

Referenced by assign().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ make_invalid()

void tbb::internal::micro_queue::make_invalid ( ticket k )

Definition at line 314 of file concurrent_queue.cpp.

 {
     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);
     }
 }

References head_page, lock, tbb::internal::concurrent_queue_rep::n_queue, page_mutex, tbb::internal::static_invalid_page, tail_counter, and tail_page.

Referenced by assign(), and push().

Here is the caller graph for this function:

◆ pop()

bool tbb::internal::micro_queue::pop	(	void *	dst,
		ticket	k,
		concurrent_queue_base &	base
	)

Definition at line 230 of file concurrent_queue.cpp.

                                                                         {
     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;
 }

References __TBB_ASSERT, tbb::internal::concurrent_queue_base_v3::assign_and_destroy_item(), head_counter, head_page, tbb::internal::concurrent_queue_base_v3::items_per_page, ITT_NOTIFY, tbb::internal::modulo_power_of_two(), tbb::internal::concurrent_queue_base_v3::my_rep, tbb::internal::concurrent_queue_rep::n_invalid_entries, tbb::internal::concurrent_queue_rep::n_queue, p, tbb::internal::spin_wait_until_eq(), tbb::internal::spin_wait_while_eq(), sync_releasing, and tail_counter.

Referenced by tbb::internal::concurrent_queue_base_v3::internal_pop(), and tbb::internal::concurrent_queue_base_v3::internal_pop_if_present().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ push()

void tbb::internal::micro_queue::push	(	const void *	item,
		ticket	k,
		concurrent_queue_base &	base,
		concurrent_queue_base::copy_specifics	op_type
	)

Definition at line 161 of file concurrent_queue.cpp.

                                                                       {
     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;
 }