tbb::internal::micro_queue Struct Reference

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);
}
tbb::internal::concurrent_queue_base_v3::copy
@ copy
Definition: _concurrent_queue_impl.h:852
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

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;
}
__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
tbb::internal::ticket
size_t ticket
Definition: concurrent_queue.cpp:42
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::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::head_page
atomic< page * > head_page
Definition: concurrent_queue.cpp:52
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::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::concurrent_queue_rep::n_queue
static const size_t n_queue
Must be power of 2.
Definition: concurrent_queue.cpp:122

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;
}
tbb::internal::concurrent_queue_base_v3::move
@ move
Definition: _concurrent_queue_impl.h:852
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

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);
}
}
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
tbb::internal::static_invalid_page
static void * static_invalid_page
Definition: concurrent_queue.cpp:156
tbb::spin_mutex::scoped_lock
friend class scoped_lock
Definition: spin_mutex.h:179
tbb::internal::micro_queue::page_mutex
spin_mutex page_mutex
Definition: concurrent_queue.cpp:58

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;
}
ITT_NOTIFY
#define ITT_NOTIFY(name, obj)
Definition: itt_notify.h:112
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
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::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
tbb::internal::micro_queue::micro_queue_pop_finalizer
friend class micro_queue_pop_finalizer
Definition: concurrent_queue.cpp:50

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;
}
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::throw_exception
void throw_exception(exception_id eid)
Versionless convenience wrapper for throw_exception_v4()
Definition: tbb_exception.h:105
tbb::internal::eid_bad_last_alloc
@ eid_bad_last_alloc
Definition: tbb_exception.h:69

References __TBB_ASSERT, __TBB_CATCH, __TBB_RETHROW, __TBB_TRY, tbb::internal::concurrent_queue_base_v3::allocate_page(), tbb::internal::concurrent_queue_base_v3::copy, tbb::internal::concurrent_queue_base_v3::copy_item(), tbb::internal::eid_bad_last_alloc, head_page, tbb::internal::concurrent_queue_base_v3::items_per_page, ITT_NOTIFY, lock, make_invalid(), tbb::internal::modulo_power_of_two(), tbb::internal::concurrent_queue_base_v3::move, tbb::internal::concurrent_queue_base_v3::my_rep, tbb::internal::concurrent_queue_rep::n_invalid_entries, tbb::internal::concurrent_queue_rep::n_queue, p, page_mutex, tbb::internal::atomic_backoff::pause(), sync_releasing, tail, tail_counter, tail_page, and tbb::internal::throw_exception().

Referenced by abort_push(), tbb::internal::concurrent_queue_base_v3::internal_insert_if_not_full(), and 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:

## Friends And Related Function Documentation

## micro_queue_pop_finalizer

friend class micro_queue_pop_finalizer

friend

Definition at line 50 of file concurrent_queue.cpp.

## Member Data Documentation

## head_counter

atomic<ticket> tbb::internal::micro_queue::head_counter

Definition at line 53 of file concurrent_queue.cpp.

Referenced by assign(), pop(), and tbb::internal::micro_queue_pop_finalizer::~micro_queue_pop_finalizer().

## head_page

atomic<page*> tbb::internal::micro_queue::head_page

Definition at line 52 of file concurrent_queue.cpp.

Referenced by assign(), tbb::internal::concurrent_queue_iterator_rep::concurrent_queue_iterator_rep(), tbb::internal::concurrent_queue_base_v3::internal_finish_clear(), make_invalid(), pop(), push(), and tbb::internal::micro_queue_pop_finalizer::~micro_queue_pop_finalizer().

## page_mutex

spin_mutex tbb::internal::micro_queue::page_mutex

Definition at line 58 of file concurrent_queue.cpp.

Referenced by make_invalid(), push(), and tbb::internal::micro_queue_pop_finalizer::~micro_queue_pop_finalizer().

## tail_counter

atomic<ticket> tbb::internal::micro_queue::tail_counter

Definition at line 56 of file concurrent_queue.cpp.

Referenced by assign(), make_invalid(), pop(), and push().

## tail_page

atomic<page*> tbb::internal::micro_queue::tail_page

Definition at line 55 of file concurrent_queue.cpp.

Referenced by assign(), tbb::internal::concurrent_queue_base_v3::internal_finish_clear(), make_invalid(), push(), tbb::internal::concurrent_queue_base_v3::~concurrent_queue_base_v3(), and tbb::internal::micro_queue_pop_finalizer::~micro_queue_pop_finalizer().

The documentation for this struct was generated from the following file:

- concurrent_queue.cpp

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.