Commit 9ce176b1 authored by Maged Michael's avatar Maged Michael Committed by Facebook Github Bot

Dynamic MPMCQueue: Eliminate cases of enqueue indefinite blocking and failure...

Dynamic MPMCQueue: Eliminate cases of enqueue indefinite blocking and failure in the extensible version that impossible under the default pre-allocated version

Summary:
Currently under the extensible version (Dynamic == true), some enqueue operations may block indefinitely or fail (return false) even though such outcomes are impossible under the default (Dynamic == false) pre-allocated version.

This diff eliminates such cases by changing the algorithms for the extensible version. Some of the high-level changes:
- The offset formula for an expansion guarantees that no enqueue operation left behind in a closed array does not have an existing dequeue operation that unblocks it. The old formula was 1 + max(head, tail). The new formula is max(head, current offset) + current capacity.
- Conditional operations validate state after the success of CAS.

Reviewed By: djwatson

Differential Revision: D5701013

fbshipit-source-id: 4917c5b35b7e2a2fddfd2e11fb5aeb478502137c
parent f7ec2efe
...@@ -118,8 +118,22 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> { ...@@ -118,8 +118,22 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> {
/// The dynamic version of MPMCQueue allows dynamic expansion of queue /// The dynamic version of MPMCQueue allows dynamic expansion of queue
/// capacity, such that a queue may start with a smaller capacity than /// capacity, such that a queue may start with a smaller capacity than
/// specified and expand only if needed. Users may optionally specify /// specified and expand if needed up to the specified capacity.
/// the initial capacity and the expansion multiplier. /// Shrinking is not supported at this point.
///
/// Users may optionally specify the initial capacity and the
/// expansion multiplier. Otherwise default values are used.
///
/// Operation on the dynamic version have the same semantics as for
/// the default fixed-size version, except that the total queue
/// capacity can temporarily exceed the specified capacity when there
/// are lagging consumers that haven't yet consumed all the elements
/// in closed arrays. Users should not rely on the capacity of dynamic
/// queues for synchronization, e.g., they should not expect that a
/// thread will definitely block on a call to blockingWrite() when the
/// queue size is known to be equal to its capacity.
///
/// Design Overview:
/// ///
/// The design uses a seqlock to enforce mutual exclusion among /// The design uses a seqlock to enforce mutual exclusion among
/// expansion attempts. Regular operations read up-to-date queue /// expansion attempts. Regular operations read up-to-date queue
...@@ -144,7 +158,7 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> { ...@@ -144,7 +158,7 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> {
/// closed arrays instead of the current one. Information about closed /// closed arrays instead of the current one. Information about closed
/// slots arrays (array address, capacity, stride, and offset) is /// slots arrays (array address, capacity, stride, and offset) is
/// maintained in a logarithmic-sized structure. Each entry in that /// maintained in a logarithmic-sized structure. Each entry in that
/// structure never need to be changed once set. The number of closed /// structure never needs to be changed once set. The number of closed
/// arrays is half the value of the seqlock (when unlocked). /// arrays is half the value of the seqlock (when unlocked).
/// ///
/// The acquisition of the seqlock to perform an expansion does not /// The acquisition of the seqlock to perform an expansion does not
...@@ -154,21 +168,6 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> { ...@@ -154,21 +168,6 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> {
/// seqlock read-only section (and hence obtained information for /// seqlock read-only section (and hence obtained information for
/// older closed arrays). /// older closed arrays).
/// ///
/// Note that the total queue capacity can temporarily exceed the
/// specified capacity when there are lagging consumers that haven't
/// yet consumed all the elements in closed arrays. Users should not
/// rely on the capacity of dynamic queues for synchronization, e.g.,
/// they should not expect that a thread will definitely block on a
/// call to blockingWrite() when the queue size is known to be equal
/// to its capacity.
///
/// Note that some writeIfNotFull() and tryWriteUntil() operations may
/// fail even if the size of the queue is less than its maximum
/// capacity and despite the success of expansion, if the operation
/// happens to acquire a ticket that belongs to a closed array. This
/// is a transient condition. Typically, one or two ticket values may
/// be subject to such condition per expansion.
///
/// The dynamic version is a partial specialization of MPMCQueue with /// The dynamic version is a partial specialization of MPMCQueue with
/// Dynamic == true /// Dynamic == true
template <typename T, template <typename> class Atom> template <typename T, template <typename> class Atom>
...@@ -272,36 +271,31 @@ class MPMCQueue<T,Atom,true> : ...@@ -272,36 +271,31 @@ class MPMCQueue<T,Atom,true> :
int stride; int stride;
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
do { while (true) {
if (!trySeqlockReadSection(state, slots, cap, stride)) { while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
continue;
} }
if (maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride)) { maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
// There was an expansion after this ticket was issued. if (LIKELY(slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
this->turn(ticket - offset, cap)))) {
// A slot is ready. Fast path. No need to expand.
break; break;
} }
if (slots[this->idx((ticket-offset), cap, stride)] // Slow path
.mayEnqueue(this->turn(ticket-offset, cap))) { auto head = this->popTicket_.load(std::memory_order_relaxed);
// A slot is ready. No need to expand. auto avail = std::max(head, offset) + cap;
break; if (ticket < avail) {
} else if (this->popTicket_.load(std::memory_order_relaxed) + cap
> ticket) {
// May block, but a pop is in progress. No need to expand. // May block, but a pop is in progress. No need to expand.
// Get seqlock read section info again in case an expansion
// occurred with an equal or higher ticket.
continue;
} else {
// May block. See if we can expand.
if (tryExpand(state, cap)) {
// This or another thread started an expansion. Get updated info.
continue;
} else {
// Can't expand.
break; break;
} }
// Try to expand, otherwise this operation may block
// indefinitely awaiting a consumer to unblock it.
if (!tryExpand(state, cap)) {
// Can't expand. Block.
break;
}
// Either this or another thread started an expansion Get up-to-date info.
} }
} while (true);
this->enqueueWithTicketBase(ticket-offset, slots, cap, stride, this->enqueueWithTicketBase(ticket-offset, slots, cap, stride,
std::forward<Args>(args)...); std::forward<Args>(args)...);
} }
...@@ -313,7 +307,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -313,7 +307,7 @@ class MPMCQueue<T,Atom,true> :
int stride; int stride;
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
while (!trySeqlockReadSection(state, slots, cap, stride)) { while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
} }
// If there was an expansion after the corresponding push ticket // If there was an expansion after the corresponding push ticket
...@@ -323,11 +317,10 @@ class MPMCQueue<T,Atom,true> : ...@@ -323,11 +317,10 @@ class MPMCQueue<T,Atom,true> :
} }
private: private:
enum { enum {
kSeqlockBits = 6, kSeqlockBits = 8,
kDefaultMinDynamicCapacity = 10, kDefaultMinDynamicCapacity = 16,
kDefaultExpansionMultiplier = 10, kDefaultExpansionMultiplier = 8,
}; };
size_t dmult_; size_t dmult_;
...@@ -354,69 +347,86 @@ class MPMCQueue<T,Atom,true> : ...@@ -354,69 +347,86 @@ class MPMCQueue<T,Atom,true> :
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
) noexcept { ) noexcept {
uint64_t state; uint64_t state;
do { while (true) {
ticket = this->pushTicket_.load(std::memory_order_acquire); // A ticket = this->pushTicket_.load(std::memory_order_acquire); // A
if (!trySeqlockReadSection(state, slots, cap, stride)) { if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
continue; continue;
} }
// If there was an expansion with offset greater than this ticket, // If there was an expansion with offset greater than this ticket,
// adjust accordingly // adjust accordingly
uint64_t offset; uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride); maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (LIKELY(slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
if (slots[this->idx((ticket-offset), cap, stride)] this->turn(ticket - offset, cap)))) {
.mayEnqueue(this->turn(ticket-offset, cap))) {
// A slot is ready. // A slot is ready.
if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) { if (UNLIKELY(!this->pushTicket_.compare_exchange_strong(
// Adjust ticket ticket, ticket + 1))) {
continue;
}
// Validate that state is the same
if (LIKELY(state == this->dstate_.load(std::memory_order_acquire))) {
ticket -= offset; ticket -= offset;
return true; return true;
} else {
continue;
} }
} else { // Slow path - state changed - get up-to-date info for obtained ticket
while (true) {
state = this->dstate_.load(std::memory_order_acquire);
if (trySeqlockReadSection(state, slots, cap, stride)) {
break;
}
asm_volatile_pause();
}
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
ticket -= offset;
return true;
}
// slow path - no ready ticket
if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
// Try again. Ticket changed. // Ticket changed. Start over.
continue; continue;
} }
// Likely to block. auto head = this->popTicket_.load(std::memory_order_acquire);
// Try to expand unless the ticket is for a closed array auto avail = std::max(head, offset) + cap;
if (offset == getOffset(state)) { if (ticket < avail) {
// a consumer is in the process of making the slot available
// don't try to expand. Spin if capacity is not
// exhausted. Otherwise return false.
if (cap == this->capacity_) {
return false;
}
asm_volatile_pause();
continue;
}
// Likely to block. Try to expand.
if (tryExpand(state, cap)) { if (tryExpand(state, cap)) {
// This or another thread started an expansion. Get up-to-date info. // This or another thread started an expansion. Get up-to-date info.
continue; continue;
} }
} // No ready ticket and cannot expand
return false; return false;
} }
} while (true);
} }
bool tryObtainPromisedPushTicket( bool tryObtainPromisedPushTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
) noexcept { ) noexcept {
uint64_t state; uint64_t state;
do { while (true) {
ticket = this->pushTicket_.load(std::memory_order_acquire); ticket = this->pushTicket_.load(std::memory_order_acquire);
auto numPops = this->popTicket_.load(std::memory_order_acquire); auto head = this->popTicket_.load(std::memory_order_acquire);
if (!trySeqlockReadSection(state, slots, cap, stride)) { if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
continue; continue;
} }
const auto curCap = cap;
// If there was an expansion with offset greater than this ticket, // If there was an expansion with offset greater than this ticket,
// adjust accordingly // adjust accordingly
uint64_t offset; uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride); maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
auto avail = std::max(offset, head) + cap;
int64_t n = ticket - numPops; if (UNLIKELY(ticket >= avail)) {
if (tryExpand(state, cap)) {
if (n >= static_cast<ssize_t>(cap)) { // Space may be available. Start over.
if ((cap == curCap) && tryExpand(state, cap)) {
// This or another thread started an expansion. Start over.
continue; continue;
} }
// Can't expand. // Can't expand.
...@@ -424,69 +434,110 @@ class MPMCQueue<T,Atom,true> : ...@@ -424,69 +434,110 @@ class MPMCQueue<T,Atom,true> :
return false; return false;
} }
if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) { if (UNLIKELY((!this->pushTicket_.compare_exchange_strong(
// Adjust ticket ticket, ticket + 1)))) {
continue;
}
// Validate that state is the same
if (LIKELY(state == this->dstate_.load(std::memory_order_acquire))) {
ticket -= offset;
return true;
}
// Obtained ticket but info is out-of-date - Update info
while (true) {
state = this->dstate_.load(std::memory_order_acquire);
if (trySeqlockReadSection(state, slots, cap, stride)) {
break;
}
asm_volatile_pause();
}
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
ticket -= offset; ticket -= offset;
return true; return true;
} }
} while (true);
} }
bool tryObtainReadyPopTicket( bool tryObtainReadyPopTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
) noexcept { ) noexcept {
uint64_t state; uint64_t state;
do { while (true) {
ticket = this->popTicket_.load(std::memory_order_relaxed); ticket = this->popTicket_.load(std::memory_order_relaxed);
if (!trySeqlockReadSection(state, slots, cap, stride)) { if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
continue; continue;
} }
// If there was an expansion after the corresponding push ticket // If there was an expansion after the corresponding push ticket
// was issued, adjust accordingly // was issued, adjust accordingly
uint64_t offset; uint64_t offset;
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride); maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (UNLIKELY(!slots[this->idx((ticket - offset), cap, stride)].mayDequeue(
if (slots[this->idx((ticket-offset), cap, stride)] this->turn(ticket - offset, cap)))) {
.mayDequeue(this->turn(ticket-offset, cap))) { return false;
if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) { }
// Adjust ticket if (UNLIKELY(
!this->popTicket_.compare_exchange_strong(ticket, ticket + 1))) {
continue;
}
// Validate that state is the same
if (LIKELY(state == this->dstate_.load(std::memory_order_acquire))) {
ticket -= offset; ticket -= offset;
return true; return true;
} }
} else { // Obtained ticket but info is out-of-date - Update info
return false; while (true) {
state = this->dstate_.load(std::memory_order_acquire);
if (trySeqlockReadSection(state, slots, cap, stride)) {
break;
}
asm_volatile_pause();
}
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
ticket -= offset;
return true;
} }
} while (true);
} }
bool tryObtainPromisedPopTicket( bool tryObtainPromisedPopTicket(
uint64_t& ticket, Slot*& slots, size_t& cap, int& stride uint64_t& ticket, Slot*& slots, size_t& cap, int& stride
) noexcept { ) noexcept {
uint64_t state; uint64_t state;
do { while (true) {
ticket = this->popTicket_.load(std::memory_order_acquire); ticket = this->popTicket_.load(std::memory_order_acquire);
auto numPushes = this->pushTicket_.load(std::memory_order_acquire); auto numPushes = this->pushTicket_.load(std::memory_order_acquire);
if (!trySeqlockReadSection(state, slots, cap, stride)) { if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause(); asm_volatile_pause();
continue; continue;
} }
uint64_t offset; uint64_t offset;
// If there was an expansion after the corresponding push // If there was an expansion after the corresponding push
// ticket was issued, adjust accordingly // ticket was issued, adjust accordingly
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride); maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
if (UNLIKELY(ticket >= numPushes)) {
if (ticket >= numPushes) {
ticket -= offset; ticket -= offset;
return false; return false;
} }
if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) { if (UNLIKELY(
!this->popTicket_.compare_exchange_strong(ticket, ticket + 1))) {
continue;
}
// Validate that state is the same
if (LIKELY(state == this->dstate_.load(std::memory_order_acquire))) {
ticket -= offset;
return true;
}
// Obtained ticket but info is out-of-date - Update info
while (true) {
state = this->dstate_.load(std::memory_order_acquire);
if (trySeqlockReadSection(state, slots, cap, stride)) {
break;
}
asm_volatile_pause();
}
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride);
ticket -= offset; ticket -= offset;
return true; return true;
} }
} while (true);
} }
/// Enqueues an element with a specific ticket number /// Enqueues an element with a specific ticket number
...@@ -498,7 +549,9 @@ class MPMCQueue<T,Atom,true> : ...@@ -498,7 +549,9 @@ class MPMCQueue<T,Atom,true> :
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
while (!trySeqlockReadSection(state, slots, cap, stride)) {} while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) {
asm_volatile_pause();
}
// If there was an expansion after this ticket was issued, adjust // If there was an expansion after this ticket was issued, adjust
// accordingly // accordingly
...@@ -517,22 +570,31 @@ class MPMCQueue<T,Atom,true> : ...@@ -517,22 +570,31 @@ class MPMCQueue<T,Atom,true> :
} }
/// Try to expand the queue. Returns true if this expansion was /// Try to expand the queue. Returns true if this expansion was
/// successful or a concurent expansion is in progress. Returns /// successful or a concurent expansion is in progresse. Returns
/// false if the queue has reached its maximum capacity or /// false if the queue has reached its maximum capacity or
/// allocation has failed. /// allocation has failed.
bool tryExpand(const uint64_t state, const size_t cap) noexcept { bool tryExpand(const uint64_t state, const size_t cap) noexcept {
if (cap == this->capacity_) { if (LIKELY(cap == this->capacity_)) {
return false; return false;
} }
return tryExpandWithSeqlock(state, cap);
}
bool tryExpandWithSeqlock(const uint64_t state, const size_t cap) noexcept {
// Acquire seqlock // Acquire seqlock
uint64_t oldval = state; uint64_t oldval = state;
assert((state & 1) == 0); assert((state & 1) == 0);
if (this->dstate_.compare_exchange_strong(oldval, state + 1)) { if (!this->dstate_.compare_exchange_strong(oldval, state + 1)) {
// Failed to acquire seqlock. Another thread acaquired it.
// Go back to the caller and get up-to-date info.
return true;
}
// Write critical section
assert(cap == this->dcapacity_.load()); assert(cap == this->dcapacity_.load());
uint64_t ticket = 1 + std::max(this->pushTicket_.load(), auto head = this->popTicket_.load(std::memory_order_acquire);
this->popTicket_.load()); auto avail = std::max(head, getOffset(state)) + cap;
size_t newCapacity = uint64_t newOffset = avail;
std::min(dmult_ * cap, this->capacity_); size_t newCapacity = std::min(dmult_ * cap, this->capacity_);
Slot* newSlots = Slot* newSlots =
new (std::nothrow) Slot[newCapacity + 2 * this->kSlotPadding]; new (std::nothrow) Slot[newCapacity + 2 * this->kSlotPadding];
if (newSlots == nullptr) { if (newSlots == nullptr) {
...@@ -543,7 +605,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -543,7 +605,7 @@ class MPMCQueue<T,Atom,true> :
// Successful expansion // Successful expansion
// calculate the current ticket offset // calculate the current ticket offset
uint64_t offset = getOffset(state); uint64_t offset = getOffset(state);
// calculate index in closed array // calculate index in list of closed slots arrays
int index = getNumClosed(state); int index = getNumClosed(state);
assert((index << 1) < (1 << kSeqlockBits)); assert((index << 1) < (1 << kSeqlockBits));
// fill the info for the closed slots array // fill the info for the closed slots array
...@@ -556,21 +618,16 @@ class MPMCQueue<T,Atom,true> : ...@@ -556,21 +618,16 @@ class MPMCQueue<T,Atom,true> :
this->dcapacity_.store(newCapacity); this->dcapacity_.store(newCapacity);
this->dstride_.store(this->computeStride(newCapacity)); this->dstride_.store(this->computeStride(newCapacity));
// Release the seqlock and record the new ticket offset // Release the seqlock and record the new ticket offset
this->dstate_.store((ticket << kSeqlockBits) + (2 * (index + 1))); this->dstate_.store((newOffset << kSeqlockBits) + (2 * (index + 1)));
return true;
} else { // failed to acquire seqlock
// Someone acaquired the seqlock. Go back to the caller and get
// up-to-date info.
return true; return true;
} }
}
/// Seqlock read-only section /// Seqlock read-only section
bool trySeqlockReadSection( bool trySeqlockReadSection(
uint64_t& state, Slot*& slots, size_t& cap, int& stride uint64_t& state, Slot*& slots, size_t& cap, int& stride
) noexcept { ) noexcept {
state = this->dstate_.load(std::memory_order_acquire); state = this->dstate_.load(std::memory_order_acquire);
if (state & 1) { if (UNLIKELY(state & 1)) {
// Locked. // Locked.
return false; return false;
} }
...@@ -580,7 +637,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -580,7 +637,7 @@ class MPMCQueue<T,Atom,true> :
stride = this->dstride_.load(std::memory_order_relaxed); stride = this->dstride_.load(std::memory_order_relaxed);
// End of read-only section. Validate seqlock. // End of read-only section. Validate seqlock.
std::atomic_thread_fence(std::memory_order_acquire); std::atomic_thread_fence(std::memory_order_acquire);
return (state == this->dstate_.load(std::memory_order_relaxed)); return LIKELY(state == this->dstate_.load(std::memory_order_relaxed));
} }
/// If there was an expansion after ticket was issued, update local variables /// If there was an expansion after ticket was issued, update local variables
...@@ -594,21 +651,32 @@ class MPMCQueue<T,Atom,true> : ...@@ -594,21 +651,32 @@ class MPMCQueue<T,Atom,true> :
size_t& cap, size_t& cap,
int& stride) noexcept { int& stride) noexcept {
offset = getOffset(state); offset = getOffset(state);
if (ticket >= offset) { if (LIKELY(ticket >= offset)) {
return false; return false;
} }
updateFromClosed(state, ticket, offset, slots, cap, stride);
return true;
}
void updateFromClosed(
const uint64_t state,
const uint64_t ticket,
uint64_t& offset,
Slot*& slots,
size_t& cap,
int& stride) noexcept {
for (int i = getNumClosed(state) - 1; i >= 0; --i) { for (int i = getNumClosed(state) - 1; i >= 0; --i) {
offset = closed_[i].offset_; offset = closed_[i].offset_;
if (offset <= ticket) { if (offset <= ticket) {
slots = closed_[i].slots_; slots = closed_[i].slots_;
cap = closed_[i].capacity_; cap = closed_[i].capacity_;
stride = closed_[i].stride_; stride = closed_[i].stride_;
return true; return;
} }
} }
// A closed array with offset <= ticket should have been found // A closed array with offset <= ticket should have been found
assert(false); assert(false);
return false; return;
} }
}; };
......
...@@ -748,11 +748,6 @@ void runMtNeverFail(std::vector<int>& nts, int n) { ...@@ -748,11 +748,6 @@ void runMtNeverFail(std::vector<int>& nts, int n) {
} }
} }
// All the never_fail tests are for the non-dynamic version only.
// False positive for dynamic version. Some writeIfNotFull() and
// tryWriteUntil() operations may fail in transient conditions related
// to expansion.
TEST(MPMCQueue, mt_never_fail) { TEST(MPMCQueue, mt_never_fail) {
std::vector<int> nts {1, 3, 100}; std::vector<int> nts {1, 3, 100};
int n = 100000; int n = 100000;
...@@ -765,6 +760,18 @@ TEST(MPMCQueue, mt_never_fail_emulated_futex) { ...@@ -765,6 +760,18 @@ TEST(MPMCQueue, mt_never_fail_emulated_futex) {
runMtNeverFail<EmulatedFutexAtomic>(nts, n); runMtNeverFail<EmulatedFutexAtomic>(nts, n);
} }
TEST(MPMCQueue, mt_never_fail_dynamic) {
std::vector<int> nts{1, 3, 100};
int n = 100000;
runMtNeverFail<std::atomic, true>(nts, n);
}
TEST(MPMCQueue, mt_never_fail_emulated_futex_dynamic) {
std::vector<int> nts{1, 3, 100};
int n = 100000;
runMtNeverFail<EmulatedFutexAtomic, true>(nts, n);
}
template <bool Dynamic = false> template <bool Dynamic = false>
void runMtNeverFailDeterministic(std::vector<int>& nts, int n, long seed) { void runMtNeverFailDeterministic(std::vector<int>& nts, int n, long seed) {
LOG(INFO) << "using seed " << seed; LOG(INFO) << "using seed " << seed;
...@@ -787,6 +794,13 @@ TEST(MPMCQueue, mt_never_fail_deterministic) { ...@@ -787,6 +794,13 @@ TEST(MPMCQueue, mt_never_fail_deterministic) {
runMtNeverFailDeterministic(nts, n, seed); runMtNeverFailDeterministic(nts, n, seed);
} }
TEST(MPMCQueue, mt_never_fail_deterministic_dynamic) {
std::vector<int> nts{3, 10};
long seed = 0; // nowMicro() % 10000;
int n = 1000;
runMtNeverFailDeterministic<true>(nts, n, seed);
}
template <class Clock, template <typename> class Atom, bool Dynamic> template <class Clock, template <typename> class Atom, bool Dynamic>
void runNeverFailUntilThread(int numThreads, void runNeverFailUntilThread(int numThreads,
int n, /*numOps*/ int n, /*numOps*/
...@@ -851,6 +865,12 @@ TEST(MPMCQueue, mt_never_fail_until_system) { ...@@ -851,6 +865,12 @@ TEST(MPMCQueue, mt_never_fail_until_system) {
runMtNeverFailUntilSystem(nts, n); runMtNeverFailUntilSystem(nts, n);
} }
TEST(MPMCQueue, mt_never_fail_until_system_dynamic) {
std::vector<int> nts{1, 3, 100};
int n = 100000;
runMtNeverFailUntilSystem<true>(nts, n);
}
template <bool Dynamic = false> template <bool Dynamic = false>
void runMtNeverFailUntilSteady(std::vector<int>& nts, int n) { void runMtNeverFailUntilSteady(std::vector<int>& nts, int n) {
for (int nt : nts) { for (int nt : nts) {
...@@ -867,6 +887,12 @@ TEST(MPMCQueue, mt_never_fail_until_steady) { ...@@ -867,6 +887,12 @@ TEST(MPMCQueue, mt_never_fail_until_steady) {
runMtNeverFailUntilSteady(nts, n); runMtNeverFailUntilSteady(nts, n);
} }
TEST(MPMCQueue, mt_never_fail_until_steady_dynamic) {
std::vector<int> nts{1, 3, 100};
int n = 100000;
runMtNeverFailUntilSteady<true>(nts, n);
}
enum LifecycleEvent { enum LifecycleEvent {
NOTHING = -1, NOTHING = -1,
DEFAULT_CONSTRUCTOR, DEFAULT_CONSTRUCTOR,
...@@ -1213,7 +1239,7 @@ TEST(MPMCQueue, try_write_until_timeout) { ...@@ -1213,7 +1239,7 @@ TEST(MPMCQueue, try_write_until_timeout) {
testTimeout<false>(queue); testTimeout<false>(queue);
} }
TEST(MPMCQueue, must_fail_try_write_until_dynamic) { TEST(MPMCQueue, try_write_until_timeout_dynamic) {
folly::MPMCQueue<int, std::atomic, true> queue(200, 1, 2); folly::MPMCQueue<int, std::atomic, true> queue(1);
testTimeout<true>(queue); testTimeout<true>(queue);
} }
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment