X-Git-Url: http://plrg.eecs.uci.edu/git/?a=blobdiff_plain;f=folly%2Ftest%2FMPMCQueueTest.cpp;h=f2467eda65b0205bc98e7d7df8e26e2a8628c606;hb=5bd3bab012ee5b5771f95a9ecfea47b6e75e0802;hp=81dce2633d8d8915e2ae5170cabc4688dfff0cd8;hpb=ec06f66c17f2469fc62221259c66d6bb417f7692;p=folly.git

diff --git a/folly/test/MPMCQueueTest.cpp b/folly/test/MPMCQueueTest.cpp
index 81dce263..f2467eda 100644
--- a/folly/test/MPMCQueueTest.cpp
+++ b/folly/test/MPMCQueueTest.cpp
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Facebook, Inc.
+ * Copyright 2017 Facebook, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,48 +14,66 @@
  * limitations under the License.
  */
 
-#include <folly/MPMCQueue.h>
 #include <folly/Format.h>
+#include <folly/MPMCQueue.h>
 #include <folly/Memory.h>
+#include <folly/portability/GTest.h>
+#include <folly/portability/SysResource.h>
+#include <folly/portability/SysTime.h>
+#include <folly/portability/Unistd.h>
+#include <folly/stop_watch.h>
 #include <folly/test/DeterministicSchedule.h>
 
 #include <boost/intrusive_ptr.hpp>
-#include <memory>
+#include <boost/thread/barrier.hpp>
 #include <functional>
+#include <memory>
 #include <thread>
 #include <utility>
-#include <unistd.h>
-#include <sys/time.h>
-#include <sys/resource.h>
-
-#include <gflags/gflags.h>
-#include <gtest/gtest.h>
 
 FOLLY_ASSUME_FBVECTOR_COMPATIBLE_1(boost::intrusive_ptr);
 
 using namespace folly;
 using namespace detail;
 using namespace test;
+using std::chrono::time_point;
+using std::chrono::steady_clock;
+using std::chrono::seconds;
+using std::chrono::milliseconds;
+using std::string;
+using std::unique_ptr;
+using std::vector;
 
 typedef DeterministicSchedule DSched;
 
+template <template<typename> class Atom>
+void run_mt_sequencer_thread(
+    int numThreads,
+    int numOps,
+    uint32_t init,
+    TurnSequencer<Atom>& seq,
+    Atom<uint32_t>& spinThreshold,
+    int& prev,
+    int i) {
+  for (int op = i; op < numOps; op += numThreads) {
+    seq.waitForTurn(init + op, spinThreshold, (op % 32) == 0);
+    EXPECT_EQ(prev, op - 1);
+    prev = op;
+    seq.completeTurn(init + op);
+  }
+}
 
 template <template<typename> class Atom>
 void run_mt_sequencer_test(int numThreads, int numOps, uint32_t init) {
   TurnSequencer<Atom> seq(init);
-  Atom<int> spinThreshold(0);
+  Atom<uint32_t> spinThreshold(0);
 
   int prev = -1;
-  std::vector<std::thread> threads(numThreads);
+  vector<std::thread> threads(numThreads);
   for (int i = 0; i < numThreads; ++i) {
-    threads[i] = DSched::thread([&, i]{
-      for (int op = i; op < numOps; op += numThreads) {
-        seq.waitForTurn(init + op, spinThreshold, (op % 32) == 0);
-        EXPECT_EQ(prev, op - 1);
-        prev = op;
-        seq.completeTurn(init + op);
-      }
-    });
+    threads[i] = DSched::thread(std::bind(run_mt_sequencer_thread<Atom>,
+          numThreads, numOps, init, std::ref(seq), std::ref(spinThreshold),
+          std::ref(prev), i));
   }
 
   for (auto& thr : threads) {
@@ -71,6 +89,12 @@ TEST(MPMCQueue, sequencer) {
   run_mt_sequencer_test<std::atomic>(100, 10000, -100);
 }
 
+TEST(MPMCQueue, sequencer_emulated_futex) {
+  run_mt_sequencer_test<EmulatedFutexAtomic>(1, 100, 0);
+  run_mt_sequencer_test<EmulatedFutexAtomic>(2, 100000, -100);
+  run_mt_sequencer_test<EmulatedFutexAtomic>(100, 10000, -100);
+}
+
 TEST(MPMCQueue, sequencer_deterministic) {
   DSched sched(DSched::uniform(0));
   run_mt_sequencer_test<DeterministicAtomic>(1, 100, -50);
@@ -78,43 +102,73 @@ TEST(MPMCQueue, sequencer_deterministic) {
   run_mt_sequencer_test<DeterministicAtomic>(10, 1000, -100);
 }
 
-template <typename T>
+template <bool Dynamic = false, typename T>
 void runElementTypeTest(T&& src) {
-  MPMCQueue<T> cq(10);
-  cq.blockingWrite(std::move(src));
+  MPMCQueue<T, std::atomic, Dynamic> cq(10);
+  cq.blockingWrite(std::forward<T>(src));
   T dest;
   cq.blockingRead(dest);
   EXPECT_TRUE(cq.write(std::move(dest)));
   EXPECT_TRUE(cq.read(dest));
+  auto soon1 = std::chrono::system_clock::now() + std::chrono::seconds(1);
+  EXPECT_TRUE(cq.tryWriteUntil(soon1, std::move(dest)));
+  EXPECT_TRUE(cq.read(dest));
+  auto soon2 = std::chrono::steady_clock::now() + std::chrono::seconds(1);
+  EXPECT_TRUE(cq.tryWriteUntil(soon2, std::move(dest)));
+  EXPECT_TRUE(cq.read(dest));
 }
 
 struct RefCounted {
+  static FOLLY_TLS int active_instances;
+
   mutable std::atomic<int> rc;
 
-  RefCounted() : rc(0) {}
+  RefCounted() : rc(0) {
+    ++active_instances;
+  }
+
+  ~RefCounted() {
+    --active_instances;
+  }
 };
+FOLLY_TLS int RefCounted::active_instances;
 
 void intrusive_ptr_add_ref(RefCounted const* p) {
   p->rc++;
 }
 
 void intrusive_ptr_release(RefCounted const* p) {
-  if (--(p->rc)) {
+  if (--(p->rc) == 0) {
     delete p;
   }
 }
 
 TEST(MPMCQueue, lots_of_element_types) {
   runElementTypeTest(10);
-  runElementTypeTest(std::string("abc"));
-  runElementTypeTest(std::make_pair(10, std::string("def")));
-  runElementTypeTest(std::vector<std::string>{ { "abc" } });
+  runElementTypeTest(string("abc"));
+  runElementTypeTest(std::make_pair(10, string("def")));
+  runElementTypeTest(vector<string>{{"abc"}});
   runElementTypeTest(std::make_shared<char>('a'));
-  runElementTypeTest(folly::make_unique<char>('a'));
+  runElementTypeTest(std::make_unique<char>('a'));
   runElementTypeTest(boost::intrusive_ptr<RefCounted>(new RefCounted));
+  EXPECT_EQ(RefCounted::active_instances, 0);
+}
+
+TEST(MPMCQueue, lots_of_element_types_dynamic) {
+  runElementTypeTest<true>(10);
+  runElementTypeTest<true>(string("abc"));
+  runElementTypeTest<true>(std::make_pair(10, string("def")));
+  runElementTypeTest<true>(vector<string>{{"abc"}});
+  runElementTypeTest<true>(std::make_shared<char>('a'));
+  runElementTypeTest<true>(std::make_unique<char>('a'));
+  runElementTypeTest<true>(boost::intrusive_ptr<RefCounted>(new RefCounted));
+  EXPECT_EQ(RefCounted::active_instances, 0);
 }
 
 TEST(MPMCQueue, single_thread_enqdeq) {
+  // Non-dynamic version only.
+  // False positive for dynamic version. Capacity can be temporarily
+  // higher than specified.
   MPMCQueue<int> cq(10);
 
   for (int pass = 0; pass < 10; ++pass) {
@@ -145,9 +199,12 @@ TEST(MPMCQueue, single_thread_enqdeq) {
 }
 
 TEST(MPMCQueue, tryenq_capacity_test) {
+  // Non-dynamic version only.
+  // False positive for dynamic version. Capacity can be temporarily
+  // higher than specified.
   for (size_t cap = 1; cap < 100; ++cap) {
     MPMCQueue<int> cq(cap);
-    for (int i = 0; i < cap; ++i) {
+    for (size_t i = 0; i < cap; ++i) {
       EXPECT_TRUE(cq.write(i));
     }
     EXPECT_FALSE(cq.write(100));
@@ -155,6 +212,9 @@ TEST(MPMCQueue, tryenq_capacity_test) {
 }
 
 TEST(MPMCQueue, enq_capacity_test) {
+  // Non-dynamic version only.
+  // False positive for dynamic version. Capacity can be temporarily
+  // higher than specified.
   for (auto cap : { 1, 100, 10000 }) {
     MPMCQueue<int> cq(cap);
     for (int i = 0; i < cap; ++i) {
@@ -175,37 +235,46 @@ TEST(MPMCQueue, enq_capacity_test) {
   }
 }
 
-template <template<typename> class Atom>
+template <template<typename> class Atom, bool Dynamic = false>
+void runTryEnqDeqThread(
+    int numThreads,
+    int n, /*numOps*/
+    MPMCQueue<int, Atom, Dynamic>& cq,
+    std::atomic<uint64_t>& sum,
+    int t) {
+  uint64_t threadSum = 0;
+  int src = t;
+  // received doesn't reflect any actual values, we just start with
+  // t and increment by numThreads to get the rounding of termination
+  // correct if numThreads doesn't evenly divide numOps
+  int received = t;
+  while (src < n || received < n) {
+    if (src < n && cq.write(src)) {
+      src += numThreads;
+    }
+
+    int dst;
+    if (received < n && cq.read(dst)) {
+      received += numThreads;
+      threadSum += dst;
+    }
+  }
+  sum += threadSum;
+}
+
+template <template<typename> class Atom, bool Dynamic = false>
 void runTryEnqDeqTest(int numThreads, int numOps) {
   // write and read aren't linearizable, so we don't have
   // hard guarantees on their individual behavior.  We can still test
   // correctness in aggregate
-  MPMCQueue<int,Atom> cq(numThreads);
+  MPMCQueue<int,Atom, Dynamic> cq(numThreads);
 
   uint64_t n = numOps;
-  std::vector<std::thread> threads(numThreads);
+  vector<std::thread> threads(numThreads);
   std::atomic<uint64_t> sum(0);
   for (int t = 0; t < numThreads; ++t) {
-    threads[t] = DSched::thread([&,t]{
-      uint64_t threadSum = 0;
-      int src = t;
-      // received doesn't reflect any actual values, we just start with
-      // t and increment by numThreads to get the rounding of termination
-      // correct if numThreads doesn't evenly divide numOps
-      int received = t;
-      while (src < n || received < n) {
-        if (src < n && cq.write(src)) {
-          src += numThreads;
-        }
-
-        int dst;
-        if (received < n && cq.read(dst)) {
-          received += numThreads;
-          threadSum += dst;
-        }
-      }
-      sum += threadSum;
-    });
+    threads[t] = DSched::thread(std::bind(runTryEnqDeqThread<Atom, Dynamic>,
+          numThreads, n, std::ref(cq), std::ref(sum), t));
   }
   for (auto& t : threads) {
     DSched::join(t);
@@ -223,6 +292,33 @@ TEST(MPMCQueue, mt_try_enq_deq) {
   }
 }
 
+TEST(MPMCQueue, mt_try_enq_deq_dynamic) {
+  int nts[] = { 1, 3, 100 };
+
+  int n = 100000;
+  for (int nt : nts) {
+    runTryEnqDeqTest<std::atomic, /* Dynamic = */ true>(nt, n);
+  }
+}
+
+TEST(MPMCQueue, mt_try_enq_deq_emulated_futex) {
+  int nts[] = { 1, 3, 100 };
+
+  int n = 100000;
+  for (int nt : nts) {
+    runTryEnqDeqTest<EmulatedFutexAtomic>(nt, n);
+  }
+}
+
+TEST(MPMCQueue, mt_try_enq_deq_emulated_futex_dynamic) {
+  int nts[] = { 1, 3, 100 };
+
+  int n = 100000;
+  for (int nt : nts) {
+    runTryEnqDeqTest<EmulatedFutexAtomic, /* Dynamic = */ true>(nt, n);
+  }
+}
+
 TEST(MPMCQueue, mt_try_enq_deq_deterministic) {
   int nts[] = { 3, 10 };
 
@@ -239,6 +335,14 @@ TEST(MPMCQueue, mt_try_enq_deq_deterministic) {
       DSched sched(DSched::uniformSubset(seed, 2));
       runTryEnqDeqTest<DeterministicAtomic>(nt, n);
     }
+    {
+      DSched sched(DSched::uniform(seed));
+      runTryEnqDeqTest<DeterministicAtomic, /*Dynamic = */ true>(nt, n);
+    }
+    {
+      DSched sched(DSched::uniformSubset(seed, 2));
+      runTryEnqDeqTest<DeterministicAtomic, /*Dynamic = */ true>(nt, n);
+    }
   }
 }
 
@@ -249,9 +353,59 @@ uint64_t nowMicro() {
 }
 
 template <typename Q>
-std::string producerConsumerBench(Q&& queue, std::string qName,
-                                  int numProducers, int numConsumers,
-                                  int numOps, bool ignoreContents = false) {
+struct WriteMethodCaller {
+  WriteMethodCaller() {}
+  virtual ~WriteMethodCaller() = default;
+  virtual bool callWrite(Q& q, int i) = 0;
+  virtual string methodName() = 0;
+};
+
+template <typename Q>
+struct BlockingWriteCaller : public WriteMethodCaller<Q> {
+  bool callWrite(Q& q, int i) override {
+    q.blockingWrite(i);
+    return true;
+  }
+  string methodName() override { return "blockingWrite"; }
+};
+
+template <typename Q>
+struct WriteIfNotFullCaller : public WriteMethodCaller<Q> {
+  bool callWrite(Q& q, int i) override { return q.writeIfNotFull(i); }
+  string methodName() override { return "writeIfNotFull"; }
+};
+
+template <typename Q>
+struct WriteCaller : public WriteMethodCaller<Q> {
+  bool callWrite(Q& q, int i) override { return q.write(i); }
+  string methodName() override { return "write"; }
+};
+
+template <typename Q,
+          class Clock = steady_clock,
+          class Duration = typename Clock::duration>
+struct TryWriteUntilCaller : public WriteMethodCaller<Q> {
+  const Duration duration_;
+  explicit TryWriteUntilCaller(Duration&& duration) : duration_(duration) {}
+  bool callWrite(Q& q, int i) override {
+    auto then = Clock::now() + duration_;
+    return q.tryWriteUntil(then, i);
+  }
+  string methodName() override {
+    return folly::sformat(
+        "tryWriteUntil({}ms)",
+        std::chrono::duration_cast<milliseconds>(duration_).count());
+  }
+};
+
+template <typename Q>
+string producerConsumerBench(Q&& queue,
+                             string qName,
+                             int numProducers,
+                             int numConsumers,
+                             int numOps,
+                             WriteMethodCaller<Q>& writer,
+                             bool ignoreContents = false) {
   Q& q = queue;
 
   struct rusage beginUsage;
@@ -261,17 +415,20 @@ std::string producerConsumerBench(Q&& queue, std::string qName,
 
   uint64_t n = numOps;
   std::atomic<uint64_t> sum(0);
+  std::atomic<uint64_t> failed(0);
 
-  std::vector<std::thread> producers(numProducers);
+  vector<std::thread> producers(numProducers);
   for (int t = 0; t < numProducers; ++t) {
     producers[t] = DSched::thread([&,t]{
       for (int i = t; i < numOps; i += numProducers) {
-        q.blockingWrite(i);
+        while (!writer.callWrite(q, i)) {
+          ++failed;
+        }
       }
     });
   }
 
-  std::vector<std::thread> consumers(numConsumers);
+  vector<std::thread> consumers(numConsumers);
   for (int t = 0; t < numConsumers; ++t) {
     consumers[t] = DSched::thread([&,t]{
       uint64_t localSum = 0;
@@ -303,69 +460,274 @@ std::string producerConsumerBench(Q&& queue, std::string qName,
   uint64_t nanosPer = (1000 * (endMicro - beginMicro)) / n;
   long csw = endUsage.ru_nvcsw + endUsage.ru_nivcsw -
       (beginUsage.ru_nvcsw + beginUsage.ru_nivcsw);
+  uint64_t failures = failed;
+  size_t allocated = q.allocatedCapacity();
+
+  return folly::sformat(
+      "{}, {} {} producers, {} consumers => {} nanos/handoff, {} csw / {} "
+      "handoff, {} failures, {} allocated",
+      qName,
+      numProducers,
+      writer.methodName(),
+      numConsumers,
+      nanosPer,
+      csw,
+      n,
+      failures,
+      allocated);
+}
 
-  return folly::format(
-      "{}, {} producers, {} consumers => {} nanos/handoff, {} csw / {} handoff",
-      qName, numProducers, numConsumers, nanosPer, csw, n).str();
+template <bool Dynamic = false>
+void runMtProdConsDeterministic(long seed) {
+  // we use the Bench method, but perf results are meaningless under DSched
+  DSched sched(DSched::uniform(seed));
+
+  using QueueType = MPMCQueue<int, DeterministicAtomic, Dynamic>;
+
+  vector<unique_ptr<WriteMethodCaller<QueueType>>> callers;
+  callers.emplace_back(std::make_unique<BlockingWriteCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteIfNotFullCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteCaller<QueueType>>());
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(milliseconds(1)));
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(seconds(2)));
+  size_t cap;
+
+  for (const auto& caller : callers) {
+    cap = 10;
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, Dynamic>(cap),
+        "MPMCQueue<int, DeterministicAtomic, Dynamic>("
+          + folly::to<std::string>(cap)+")",
+        1,
+        1,
+        1000,
+        *caller);
+    cap = 100;
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, Dynamic>(cap),
+        "MPMCQueue<int, DeterministicAtomic, Dynamic>("
+          + folly::to<std::string>(cap)+")",
+        10,
+        10,
+        1000,
+        *caller);
+    cap = 10;
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, Dynamic>(cap),
+        "MPMCQueue<int, DeterministicAtomic, Dynamic>("
+          + folly::to<std::string>(cap)+")",
+        1,
+        1,
+        1000,
+        *caller);
+    cap = 100;
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, Dynamic>(cap),
+        "MPMCQueue<int, DeterministicAtomic, Dynamic>("
+          + folly::to<std::string>(cap)+")",
+        10,
+        10,
+        1000,
+        *caller);
+    cap = 1;
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, Dynamic>(cap),
+        "MPMCQueue<int, DeterministicAtomic, Dynamic>("
+          + folly::to<std::string>(cap)+")",
+        10,
+        10,
+        1000,
+        *caller);
+  }
 }
 
+void runMtProdConsDeterministicDynamic(
+  long seed,
+  uint32_t prods,
+  uint32_t cons,
+  uint32_t numOps,
+  size_t cap,
+  size_t minCap,
+  size_t mult
+) {
+  // we use the Bench method, but perf results are meaningless under DSched
+  DSched sched(DSched::uniform(seed));
+
+  using QueueType = MPMCQueue<int, DeterministicAtomic, true>;
+
+  vector<unique_ptr<WriteMethodCaller<QueueType>>> callers;
+  callers.emplace_back(std::make_unique<BlockingWriteCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteIfNotFullCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteCaller<QueueType>>());
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(milliseconds(1)));
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(seconds(2)));
+
+  for (const auto& caller : callers) {
+    LOG(INFO) <<
+      producerConsumerBench(
+        MPMCQueue<int, DeterministicAtomic, true>(cap, minCap, mult),
+        "MPMCQueue<int, DeterministicAtomic, true>("
+          + folly::to<std::string>(cap) + ", "
+          + folly::to<std::string>(minCap) + ", "
+          + folly::to<std::string>(mult)+")",
+        prods,
+        cons,
+        numOps,
+        *caller);
+  }
+}
 
 TEST(MPMCQueue, mt_prod_cons_deterministic) {
-  // we use the Bench method, but perf results are meaningless under DSched
-  DSched sched(DSched::uniform(0));
+  runMtProdConsDeterministic(0);
+}
+
+TEST(MPMCQueue, mt_prod_cons_deterministic_dynamic) {
+  runMtProdConsDeterministic<true>(0);
+}
 
-  producerConsumerBench(MPMCQueue<int,DeterministicAtomic>(10),
-          "", 1, 1, 1000);
-  producerConsumerBench(MPMCQueue<int,DeterministicAtomic>(100),
-          "", 10, 10, 1000);
-  producerConsumerBench(MPMCQueue<int,DeterministicAtomic>(10),
-          "", 1, 1, 1000);
-  producerConsumerBench(MPMCQueue<int,DeterministicAtomic>(100),
-          "", 10, 10, 1000);
-  producerConsumerBench(MPMCQueue<int,DeterministicAtomic>(1),
-          "", 10, 10, 1000);
+template <typename T>
+void setFromEnv(T& var, const char* envvar) {
+  char* str = std::getenv(envvar);
+  if (str) { var = atoi(str); }
+}
+
+TEST(MPMCQueue, mt_prod_cons_deterministic_dynamic_with_arguments) {
+  long seed = 0;
+  uint32_t prods = 10;
+  uint32_t cons = 10;
+  uint32_t numOps = 1000;
+  size_t cap = 10000;
+  size_t minCap = 9;
+  size_t mult = 3;
+  setFromEnv(seed, "SEED");
+  setFromEnv(prods, "PRODS");
+  setFromEnv(cons, "CONS");
+  setFromEnv(numOps, "NUM_OPS");
+  setFromEnv(cap, "CAP");
+  setFromEnv(minCap, "MIN_CAP");
+  setFromEnv(mult, "MULT");
+  runMtProdConsDeterministicDynamic(
+    seed, prods, cons, numOps, cap, minCap, mult);
 }
 
 #define PC_BENCH(q, np, nc, ...) \
     producerConsumerBench(q, #q, (np), (nc), __VA_ARGS__)
 
+template <bool Dynamic = false>
+void runMtProdCons() {
+  using QueueType = MPMCQueue<int, std::atomic, Dynamic>;
+
+  int n = 100000;
+  setFromEnv(n, "NUM_OPS");
+  vector<unique_ptr<WriteMethodCaller<QueueType>>>
+    callers;
+  callers.emplace_back(std::make_unique<BlockingWriteCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteIfNotFullCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteCaller<QueueType>>());
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(milliseconds(1)));
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(seconds(2)));
+  for (const auto& caller : callers) {
+    LOG(INFO) << PC_BENCH((QueueType(10)), 1, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 10, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 1, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 10, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 1, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 10, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 1, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 10, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(100000)), 32, 100, n, *caller);
+  }
+}
+
 TEST(MPMCQueue, mt_prod_cons) {
+  runMtProdCons();
+}
+
+TEST(MPMCQueue, mt_prod_cons_dynamic) {
+  runMtProdCons</* Dynamic = */ true>();
+}
+
+template <bool Dynamic = false>
+void runMtProdConsEmulatedFutex() {
+  using QueueType = MPMCQueue<int, EmulatedFutexAtomic, Dynamic>;
+
   int n = 100000;
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10), 1, 1, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10), 10, 1, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10), 1, 10, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10), 10, 10, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10000), 1, 1, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10000), 10, 1, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10000), 1, 10, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(10000), 10, 10, n);
-  LOG(INFO) << PC_BENCH(MPMCQueue<int>(100000), 32, 100, n);
+  vector<unique_ptr<WriteMethodCaller<QueueType>>> callers;
+  callers.emplace_back(std::make_unique<BlockingWriteCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteIfNotFullCaller<QueueType>>());
+  callers.emplace_back(std::make_unique<WriteCaller<QueueType>>());
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(milliseconds(1)));
+  callers.emplace_back(
+      std::make_unique<TryWriteUntilCaller<QueueType>>(seconds(2)));
+  for (const auto& caller : callers) {
+    LOG(INFO) << PC_BENCH((QueueType(10)), 1, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 10, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 1, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10)), 10, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 1, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 10, 1, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 1, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(10000)), 10, 10, n, *caller);
+    LOG(INFO) << PC_BENCH((QueueType(100000)), 32, 100, n, *caller);
+  }
 }
 
-template <template<typename> class Atom>
+TEST(MPMCQueue, mt_prod_cons_emulated_futex) {
+  runMtProdConsEmulatedFutex();
+}
+
+TEST(MPMCQueue, mt_prod_cons_emulated_futex_dynamic) {
+  runMtProdConsEmulatedFutex</* Dynamic = */ true>();
+}
+
+template <template <typename> class Atom, bool Dynamic = false>
+void runNeverFailThread(int numThreads,
+                        int n, /*numOps*/
+                        MPMCQueue<int, Atom, Dynamic>& cq,
+                        std::atomic<uint64_t>& sum,
+                        int t) {
+  uint64_t threadSum = 0;
+  for (int i = t; i < n; i += numThreads) {
+    // enq + deq
+    EXPECT_TRUE(cq.writeIfNotFull(i));
+
+    int dest = -1;
+    EXPECT_TRUE(cq.readIfNotEmpty(dest));
+    EXPECT_TRUE(dest >= 0);
+    threadSum += dest;
+  }
+  sum += threadSum;
+}
+
+template <template <typename> class Atom, bool Dynamic = false>
 uint64_t runNeverFailTest(int numThreads, int numOps) {
   // always #enq >= #deq
-  MPMCQueue<int,Atom> cq(numThreads);
+  MPMCQueue<int, Atom, Dynamic> cq(numThreads);
 
   uint64_t n = numOps;
   auto beginMicro = nowMicro();
 
-  std::vector<std::thread> threads(numThreads);
+  vector<std::thread> threads(numThreads);
   std::atomic<uint64_t> sum(0);
   for (int t = 0; t < numThreads; ++t) {
-    threads[t] = DSched::thread([&,t]{
-      uint64_t threadSum = 0;
-      for (int i = t; i < n; i += numThreads) {
-        // enq + deq
-        EXPECT_TRUE(cq.writeIfNotFull(i));
-
-        int dest = -1;
-        EXPECT_TRUE(cq.readIfNotEmpty(dest));
-        EXPECT_TRUE(dest >= 0);
-        threadSum += dest;
-      }
-      sum += threadSum;
-    });
+    threads[t] = DSched::thread(std::bind(runNeverFailThread<Atom, Dynamic>,
+                                          numThreads,
+                                          n,
+                                          std::ref(cq),
+                                          std::ref(sum),
+                                          t));
   }
   for (auto& t : threads) {
     DSched::join(t);
@@ -376,36 +738,134 @@ uint64_t runNeverFailTest(int numThreads, int numOps) {
   return nowMicro() - beginMicro;
 }
 
-TEST(MPMCQueue, mt_never_fail) {
-  int nts[] = { 1, 3, 100 };
-
-  int n = 100000;
+template <template<typename> class Atom, bool Dynamic = false>
+void runMtNeverFail(std::vector<int>& nts, int n) {
   for (int nt : nts) {
-    uint64_t elapsed = runNeverFailTest<std::atomic>(nt, n);
-    LOG(INFO) << (elapsed * 1000.0) / (n * 2) << " nanos per op with "
-              << nt << " threads";
+    uint64_t elapsed = runNeverFailTest<Atom, Dynamic>(nt, n);
+    LOG(INFO) << (elapsed * 1000.0) / (n * 2) << " nanos per op with " << nt
+              << " threads";
   }
 }
 
-TEST(MPMCQueue, mt_never_fail_deterministic) {
-  int nts[] = { 3, 10 };
+// 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.
 
-  long seed = 0; // nowMicro() % 10000;
-  LOG(INFO) << "using seed " << seed;
+TEST(MPMCQueue, mt_never_fail) {
+  std::vector<int> nts {1, 3, 100};
+  int n = 100000;
+  runMtNeverFail<std::atomic>(nts, n);
+}
 
-  int n = 1000;
+TEST(MPMCQueue, mt_never_fail_emulated_futex) {
+  std::vector<int> nts {1, 3, 100};
+  int n = 100000;
+  runMtNeverFail<EmulatedFutexAtomic>(nts, n);
+}
+
+template<bool Dynamic = false>
+void runMtNeverFailDeterministic(std::vector<int>& nts, int n, long seed) {
+  LOG(INFO) << "using seed " << seed;
   for (int nt : nts) {
     {
       DSched sched(DSched::uniform(seed));
-      runNeverFailTest<DeterministicAtomic>(nt, n);
+      runNeverFailTest<DeterministicAtomic, Dynamic>(nt, n);
     }
     {
       DSched sched(DSched::uniformSubset(seed, 2));
-      runNeverFailTest<DeterministicAtomic>(nt, n);
+      runNeverFailTest<DeterministicAtomic, Dynamic>(nt, n);
     }
   }
 }
 
+TEST(MPMCQueue, mt_never_fail_deterministic) {
+  std::vector<int> nts {3, 10};
+  long seed = 0; // nowMicro() % 10000;
+  int n = 1000;
+  runMtNeverFailDeterministic(nts, n, seed);
+}
+
+template <class Clock, template <typename> class Atom, bool Dynamic>
+void runNeverFailUntilThread(int numThreads,
+                             int n, /*numOps*/
+                             MPMCQueue<int, Atom, Dynamic>& cq,
+                             std::atomic<uint64_t>& sum,
+                             int t) {
+  uint64_t threadSum = 0;
+  for (int i = t; i < n; i += numThreads) {
+    // enq + deq
+    auto soon = Clock::now() + std::chrono::seconds(1);
+    EXPECT_TRUE(cq.tryWriteUntil(soon, i));
+
+    int dest = -1;
+    EXPECT_TRUE(cq.readIfNotEmpty(dest));
+    EXPECT_TRUE(dest >= 0);
+    threadSum += dest;
+  }
+  sum += threadSum;
+}
+
+template <class Clock, template <typename> class Atom, bool Dynamic = false>
+uint64_t runNeverFailTest(int numThreads, int numOps) {
+  // always #enq >= #deq
+  MPMCQueue<int, Atom, Dynamic> cq(numThreads);
+
+  uint64_t n = numOps;
+  auto beginMicro = nowMicro();
+
+  vector<std::thread> threads(numThreads);
+  std::atomic<uint64_t> sum(0);
+  for (int t = 0; t < numThreads; ++t) {
+    threads[t] = DSched::thread(std::bind(
+                                  runNeverFailUntilThread<Clock, Atom, Dynamic>,
+                                  numThreads,
+                                  n,
+                                  std::ref(cq),
+                                  std::ref(sum),
+                                  t));
+  }
+  for (auto& t : threads) {
+    DSched::join(t);
+  }
+  EXPECT_TRUE(cq.isEmpty());
+  EXPECT_EQ(n * (n - 1) / 2 - sum, 0);
+
+  return nowMicro() - beginMicro;
+}
+
+template <bool Dynamic = false>
+void runMtNeverFailUntilSystem(std::vector<int>& nts, int n) {
+  for (int nt : nts) {
+    uint64_t elapsed =
+      runNeverFailTest<std::chrono::system_clock, std::atomic, Dynamic>(nt, n);
+    LOG(INFO) << (elapsed * 1000.0) / (n * 2) << " nanos per op with " << nt
+              << " threads";
+  }
+}
+
+TEST(MPMCQueue, mt_never_fail_until_system) {
+  std::vector<int> nts {1, 3, 100};
+  int n = 100000;
+  runMtNeverFailUntilSystem(nts, n);
+}
+
+template <bool Dynamic = false>
+void runMtNeverFailUntilSteady(std::vector<int>& nts, int n) {
+  for (int nt : nts) {
+    uint64_t elapsed =
+      runNeverFailTest<std::chrono::steady_clock, std::atomic, Dynamic>(nt, n);
+    LOG(INFO) << (elapsed * 1000.0) / (n * 2) << " nanos per op with " << nt
+              << " threads";
+  }
+}
+
+TEST(MPMCQueue, mt_never_fail_until_steady) {
+  std::vector<int> nts {1, 3, 100};
+  int n = 100000;
+  runMtNeverFailUntilSteady(nts, n);
+}
+
 enum LifecycleEvent {
   NOTHING = -1,
   DEFAULT_CONSTRUCTOR,
@@ -456,24 +916,25 @@ struct Lifecycle {
     ++lc_counts[DEFAULT_CONSTRUCTOR];
   }
 
-  explicit Lifecycle(int n, char const* s) noexcept : constructed(true) {
+  explicit Lifecycle(int /* n */, char const* /* s */) noexcept
+      : constructed(true) {
     ++lc_counts[TWO_ARG_CONSTRUCTOR];
   }
 
-  Lifecycle(const Lifecycle& rhs) noexcept : constructed(true) {
+  Lifecycle(const Lifecycle& /* rhs */) noexcept : constructed(true) {
     ++lc_counts[COPY_CONSTRUCTOR];
   }
 
-  Lifecycle(Lifecycle&& rhs) noexcept : constructed(true) {
+  Lifecycle(Lifecycle&& /* rhs */) noexcept : constructed(true) {
     ++lc_counts[MOVE_CONSTRUCTOR];
   }
 
-  Lifecycle& operator= (const Lifecycle& rhs) noexcept {
+  Lifecycle& operator=(const Lifecycle& /* rhs */) noexcept {
     ++lc_counts[COPY_OPERATOR];
     return *this;
   }
 
-  Lifecycle& operator= (Lifecycle&& rhs) noexcept {
+  Lifecycle& operator=(Lifecycle&& /* rhs */) noexcept {
     ++lc_counts[MOVE_OPERATOR];
     return *this;
   }
@@ -492,7 +953,8 @@ void runPerfectForwardingTest() {
   EXPECT_EQ(lc_outstanding(), 0);
 
   {
-    MPMCQueue<Lifecycle<R>> queue(50);
+    // Non-dynamic only. False positive for dynamic.
+    MPMCQueue<Lifecycle<R>, std::atomic> queue(50);
     LIFECYCLE_STEP(NOTHING);
 
     for (int pass = 0; pass < 10; ++pass) {
@@ -568,19 +1030,21 @@ TEST(MPMCQueue, perfect_forwarding_relocatable) {
   runPerfectForwardingTest<std::true_type>();
 }
 
-TEST(MPMCQueue, queue_moving) {
+template <bool Dynamic = false>
+void run_queue_moving() {
   lc_snap();
   EXPECT_EQ(lc_outstanding(), 0);
 
   {
-    MPMCQueue<Lifecycle<std::false_type>> a(50);
+    MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> a(50);
     LIFECYCLE_STEP(NOTHING);
 
     a.blockingWrite();
     LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
 
     // move constructor
-    MPMCQueue<Lifecycle<std::false_type>> b = std::move(a);
+    MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> b
+      = std::move(a);
     LIFECYCLE_STEP(NOTHING);
     EXPECT_EQ(a.capacity(), 0);
     EXPECT_EQ(a.size(), 0);
@@ -591,7 +1055,7 @@ TEST(MPMCQueue, queue_moving) {
     LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
 
     // move operator
-    MPMCQueue<Lifecycle<std::false_type>> c;
+    MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> c;
     LIFECYCLE_STEP(NOTHING);
     c = std::move(b);
     LIFECYCLE_STEP(NOTHING);
@@ -606,7 +1070,7 @@ TEST(MPMCQueue, queue_moving) {
 
       {
         // swap
-        MPMCQueue<Lifecycle<std::false_type>> d(10);
+        MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> d(10);
         LIFECYCLE_STEP(NOTHING);
         std::swap(c, d);
         LIFECYCLE_STEP(NOTHING);
@@ -631,8 +1095,124 @@ TEST(MPMCQueue, queue_moving) {
   LIFECYCLE_STEP(DESTRUCTOR);
 }
 
-int main(int argc, char ** argv) {
-  testing::InitGoogleTest(&argc, argv);
-  google::ParseCommandLineFlags(&argc, &argv, true);
-  return RUN_ALL_TESTS();
+TEST(MPMCQueue, queue_moving) {
+  run_queue_moving();
+}
+
+TEST(MPMCQueue, queue_moving_dynamic) {
+  run_queue_moving<true>();
+}
+
+TEST(MPMCQueue, explicit_zero_capacity_fail) {
+  ASSERT_THROW(MPMCQueue<int> cq(0), std::invalid_argument);
+
+  using DynamicMPMCQueueInt = MPMCQueue<int, std::atomic, true>;
+  ASSERT_THROW(DynamicMPMCQueueInt cq(0), std::invalid_argument);
+}
+
+template <bool Dynamic>
+void testTryReadUntil() {
+  MPMCQueue<int, std::atomic, Dynamic> q{1};
+
+  const auto wait = std::chrono::milliseconds(100);
+  stop_watch<> watch;
+  bool rets[2];
+  int vals[2];
+  std::vector<std::thread> threads;
+  boost::barrier b{3};
+  for (int i = 0; i < 2; i++) {
+    threads.emplace_back([&, i] {
+      b.wait();
+      rets[i] = q.tryReadUntil(watch.getCheckpoint() + wait, vals[i]);
+    });
+  }
+
+  b.wait();
+  EXPECT_TRUE(q.write(42));
+
+  for (int i = 0; i < 2; i++) {
+    threads[i].join();
+  }
+
+  for (int i = 0; i < 2; i++) {
+    int other = (i + 1) % 2;
+    if (rets[i]) {
+      EXPECT_EQ(42, vals[i]);
+      EXPECT_FALSE(rets[other]);
+    }
+  }
+
+  EXPECT_TRUE(watch.elapsed(wait));
+}
+
+template <bool Dynamic>
+void testTryWriteUntil() {
+  MPMCQueue<int, std::atomic, Dynamic> q{1};
+  EXPECT_TRUE(q.write(42));
+
+  const auto wait = std::chrono::milliseconds(100);
+  stop_watch<> watch;
+  bool rets[2];
+  std::vector<std::thread> threads;
+  boost::barrier b{3};
+  for (int i = 0; i < 2; i++) {
+    threads.emplace_back([&, i] {
+      b.wait();
+      rets[i] = q.tryWriteUntil(watch.getCheckpoint() + wait, i);
+    });
+  }
+
+  b.wait();
+  int x;
+  EXPECT_TRUE(q.read(x));
+  EXPECT_EQ(42, x);
+
+  for (int i = 0; i < 2; i++) {
+    threads[i].join();
+  }
+  EXPECT_TRUE(q.read(x));
+
+  for (int i = 0; i < 2; i++) {
+    int other = (i + 1) % 2;
+    if (rets[i]) {
+      EXPECT_EQ(i, x);
+      EXPECT_FALSE(rets[other]);
+    }
+  }
+
+  EXPECT_TRUE(watch.elapsed(wait));
+}
+
+TEST(MPMCQueue, try_read_until) {
+  testTryReadUntil<false>();
+}
+
+TEST(MPMCQueue, try_read_until_dynamic) {
+  testTryReadUntil<true>();
+}
+
+TEST(MPMCQueue, try_write_until) {
+  testTryWriteUntil<false>();
+}
+
+TEST(MPMCQueue, try_write_until_dynamic) {
+  testTryWriteUntil<true>();
+}
+
+template <bool Dynamic>
+void testTimeout(MPMCQueue<int, std::atomic, Dynamic>& q) {
+  CHECK(q.write(1));
+  /* The following must not block forever */
+  q.tryWriteUntil(
+      std::chrono::system_clock::now() + std::chrono::microseconds(10000), 2);
+}
+
+TEST(MPMCQueue, try_write_until_timeout) {
+  folly::MPMCQueue<int, std::atomic, false> queue(1);
+  testTimeout<false>(queue);
+}
+
+TEST(MPMCQueue, must_fail_try_write_until_dynamic) {
+  folly::MPMCQueue<int, std::atomic, true> queue(200, 1, 2);
+  testTimeout<true>(queue);
 }