Commit 1292801d authored by Maged Michael's avatar Maged Michael Committed by Facebook Github Bot

Dynamic MPMCQueue: Backout of last change as it may deadlock

Summary: The previous change can lead to deadlock. Backing out.

Reviewed By: djwatson

Differential Revision: D5957084

fbshipit-source-id: 72ea1cb6236367912b4b087da7e4d57f8a2daed0
parent 36b8f9c6
...@@ -118,22 +118,8 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> { ...@@ -118,22 +118,8 @@ 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 if needed up to the specified capacity. /// specified and expand only if needed. Users may optionally specify
/// Shrinking is not supported at this point. /// the initial capacity and the expansion multiplier.
///
/// 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
...@@ -168,6 +154,21 @@ class MPMCQueue : public detail::MPMCQueueBase<MPMCQueue<T,Atom,Dynamic>> { ...@@ -168,6 +154,21 @@ 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>
...@@ -271,31 +272,36 @@ class MPMCQueue<T,Atom,true> : ...@@ -271,31 +272,36 @@ class MPMCQueue<T,Atom,true> :
int stride; int stride;
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
while (true) { do {
while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { if (!trySeqlockReadSection(state, slots, cap, stride)) {
asm_volatile_pause(); asm_volatile_pause();
continue;
} }
maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride); if (maybeUpdateFromClosed(state, ticket, offset, slots, cap, stride)) {
if (LIKELY(slots[this->idx((ticket - offset), cap, stride)].mayEnqueue( // There was an expansion after this ticket was issued.
this->turn(ticket - offset, cap)))) {
// A slot is ready. Fast path. No need to expand.
break; break;
} }
// Slow path if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
auto head = this->popTicket_.load(std::memory_order_relaxed); this->turn(ticket - offset, cap))) {
auto avail = std::max(head, offset) + cap; // A slot is ready. No need to expand.
if (ticket < avail) {
// May block, but a pop is in progress. No need to expand.
break; break;
} } else if (
// Try to expand, otherwise this operation may block this->popTicket_.load(std::memory_order_relaxed) + cap > ticket) {
// indefinitely awaiting a consumer to unblock it. // May block, but a pop is in progress. No need to expand.
if (!tryExpand(state, cap)) { // Get seqlock read section info again in case an expansion
// Can't expand. Block. // 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;
} }
// 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)...);
} }
...@@ -307,7 +313,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -307,7 +313,7 @@ class MPMCQueue<T,Atom,true> :
int stride; int stride;
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { while (!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
...@@ -318,9 +324,9 @@ class MPMCQueue<T,Atom,true> : ...@@ -318,9 +324,9 @@ class MPMCQueue<T,Atom,true> :
private: private:
enum { enum {
kSeqlockBits = 8, kSeqlockBits = 6,
kDefaultMinDynamicCapacity = 16, kDefaultMinDynamicCapacity = 10,
kDefaultExpansionMultiplier = 8, kDefaultExpansionMultiplier = 10,
}; };
size_t dmult_; size_t dmult_;
...@@ -347,86 +353,69 @@ class MPMCQueue<T,Atom,true> : ...@@ -347,86 +353,69 @@ 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;
while (true) { do {
ticket = this->pushTicket_.load(std::memory_order_acquire); // A ticket = this->pushTicket_.load(std::memory_order_acquire); // A
if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { if (!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(
this->turn(ticket - offset, cap)))) { if (slots[this->idx((ticket - offset), cap, stride)].mayEnqueue(
this->turn(ticket - offset, cap))) {
// A slot is ready. // A slot is ready.
if (UNLIKELY(!this->pushTicket_.compare_exchange_strong( if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
ticket, ticket + 1))) { // Adjust ticket
continue;
}
// Validate that state is the same
if (LIKELY(state == this->dstate_.load(std::memory_order_acquire))) {
ticket -= offset;
return true;
}
// 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; ticket -= offset;
return true; return true;
} } else {
// slow path - no ready ticket
if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
// Ticket changed. Start over.
continue; continue;
} }
auto head = this->popTicket_.load(std::memory_order_acquire); } else {
auto avail = std::max(head, offset) + cap; if (ticket != this->pushTicket_.load(std::memory_order_relaxed)) { // B
if (ticket < avail) { // Try again. Ticket changed.
// 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; continue;
} }
// Likely to block. Try to expand. // Likely to block.
// Try to expand unless the ticket is for a closed array
if (offset == getOffset(state)) {
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;
while (true) { do {
ticket = this->pushTicket_.load(std::memory_order_acquire); ticket = this->pushTicket_.load(std::memory_order_acquire);
auto head = this->popTicket_.load(std::memory_order_acquire); auto numPops = this->popTicket_.load(std::memory_order_acquire);
if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { if (!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;
if (UNLIKELY(ticket >= avail)) { int64_t n = ticket - numPops;
if (tryExpand(state, cap)) {
// Space may be available. Start over. if (n >= static_cast<ssize_t>(cap)) {
if ((cap == curCap) && tryExpand(state, cap)) {
// This or another thread started an expansion. Start over.
continue; continue;
} }
// Can't expand. // Can't expand.
...@@ -434,110 +423,69 @@ class MPMCQueue<T,Atom,true> : ...@@ -434,110 +423,69 @@ class MPMCQueue<T,Atom,true> :
return false; return false;
} }
if (UNLIKELY((!this->pushTicket_.compare_exchange_strong( if (this->pushTicket_.compare_exchange_strong(ticket, ticket + 1)) {
ticket, ticket + 1)))) { // Adjust ticket
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;
while (true) { do {
ticket = this->popTicket_.load(std::memory_order_relaxed); ticket = this->popTicket_.load(std::memory_order_relaxed);
if (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { if (!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(
this->turn(ticket - offset, cap)))) { if (slots[this->idx((ticket - offset), cap, stride)].mayDequeue(
return false; this->turn(ticket - offset, cap))) {
} if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
if (UNLIKELY( // Adjust ticket
!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;
} }
// Obtained ticket but info is out-of-date - Update info } else {
while (true) { return false;
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;
while (true) { do {
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 (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { if (!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 (UNLIKELY( if (this->popTicket_.compare_exchange_strong(ticket, ticket + 1)) {
!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
...@@ -549,8 +497,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -549,8 +497,7 @@ class MPMCQueue<T,Atom,true> :
uint64_t state; uint64_t state;
uint64_t offset; uint64_t offset;
while (UNLIKELY(!trySeqlockReadSection(state, slots, cap, stride))) { while (!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
...@@ -570,30 +517,20 @@ class MPMCQueue<T,Atom,true> : ...@@ -570,30 +517,20 @@ 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 progresse. Returns /// successful or a concurent expansion is in progress. 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 (LIKELY(cap == this->capacity_)) { if (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());
auto head = this->popTicket_.load(std::memory_order_acquire); uint64_t ticket =
auto avail = std::max(head, getOffset(state)) + cap; 1 + std::max(this->pushTicket_.load(), this->popTicket_.load());
uint64_t newOffset = avail;
size_t newCapacity = 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];
...@@ -605,7 +542,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -605,7 +542,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 list of closed slots arrays // calculate index in closed array
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
...@@ -618,8 +555,13 @@ class MPMCQueue<T,Atom,true> : ...@@ -618,8 +555,13 @@ 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((newOffset << kSeqlockBits) + (2 * (index + 1))); this->dstate_.store((ticket << kSeqlockBits) + (2 * (index + 1)));
return true; return true;
} else { // failed to acquire seqlock
// Someone acaquired the seqlock. Go back to the caller and get
// up-to-date info.
return true;
}
} }
/// Seqlock read-only section /// Seqlock read-only section
...@@ -627,7 +569,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -627,7 +569,7 @@ class MPMCQueue<T,Atom,true> :
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 (UNLIKELY(state & 1)) { if (state & 1) {
// Locked. // Locked.
return false; return false;
} }
...@@ -637,7 +579,7 @@ class MPMCQueue<T,Atom,true> : ...@@ -637,7 +579,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 LIKELY(state == this->dstate_.load(std::memory_order_relaxed)); return (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
...@@ -651,32 +593,21 @@ class MPMCQueue<T,Atom,true> : ...@@ -651,32 +593,21 @@ class MPMCQueue<T,Atom,true> :
size_t& cap, size_t& cap,
int& stride) noexcept { int& stride) noexcept {
offset = getOffset(state); offset = getOffset(state);
if (LIKELY(ticket >= offset)) { if (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; return true;
} }
} }
// 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; return false;
} }
}; };
......
...@@ -748,6 +748,11 @@ void runMtNeverFail(std::vector<int>& nts, int n) { ...@@ -748,6 +748,11 @@ 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;
...@@ -760,18 +765,6 @@ TEST(MPMCQueue, mt_never_fail_emulated_futex) { ...@@ -760,18 +765,6 @@ 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;
...@@ -794,13 +787,6 @@ TEST(MPMCQueue, mt_never_fail_deterministic) { ...@@ -794,13 +787,6 @@ 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*/
...@@ -865,12 +851,6 @@ TEST(MPMCQueue, mt_never_fail_until_system) { ...@@ -865,12 +851,6 @@ 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) {
...@@ -887,12 +867,6 @@ TEST(MPMCQueue, mt_never_fail_until_steady) { ...@@ -887,12 +867,6 @@ 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,
...@@ -1239,7 +1213,7 @@ TEST(MPMCQueue, try_write_until_timeout) { ...@@ -1239,7 +1213,7 @@ TEST(MPMCQueue, try_write_until_timeout) {
testTimeout<false>(queue); testTimeout<false>(queue);
} }
TEST(MPMCQueue, try_write_until_timeout_dynamic) { TEST(MPMCQueue, must_fail_try_write_until_dynamic) {
folly::MPMCQueue<int, std::atomic, true> queue(1); folly::MPMCQueue<int, std::atomic, true> queue(200, 1, 2);
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