MPMCQueueTest.cpp File Reference

#include <folly/MPMCQueue.h>
#include <folly/Format.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 <boost/thread/barrier.hpp>
#include <functional>
#include <memory>
#include <thread>
#include <utility>

Go to the source code of this file.

Classes
struct	RefCounted
struct	WriteMethodCaller< Q >
struct	BlockingWriteCaller< Q >
struct	WriteIfNotFullCaller< Q >
struct	WriteCaller< Q >
struct	TryWriteUntilCaller< Q, Clock, Duration >
struct	Lifecycle< R >

Macros
#define	PC_BENCH(q, np, nc, ...)   producerConsumerBench(q, #q, (np), (nc), __VA_ARGS__)
#define	LIFECYCLE_STEP(...)   lc_step(__LINE__, __VA_ARGS__)

Typedefs
typedef DeterministicSchedule	DSched

Enumerations
enum	LifecycleEvent {   NOTHING = -1, DEFAULT_CONSTRUCTOR, COPY_CONSTRUCTOR, MOVE_CONSTRUCTOR,   TWO_ARG_CONSTRUCTOR, COPY_OPERATOR, MOVE_OPERATOR, DESTRUCTOR,   MAX_LIFECYCLE_EVENT }

Functions
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)
template<template< typename > class Atom>
void	run_mt_sequencer_test (int numThreads, int numOps, uint32_t init)
	TEST (MPMCQueue, sequencer)
	TEST (MPMCQueue, sequencer_emulated_futex)
	TEST (MPMCQueue, sequencer_deterministic)
template<bool Dynamic = false, typename T >
void	runElementTypeTest (T &&src)
void	intrusive_ptr_add_ref (RefCounted const *p)
void	intrusive_ptr_release (RefCounted const *p)
	TEST (MPMCQueue, lots_of_element_types)
	TEST (MPMCQueue, lots_of_element_types_dynamic)
	TEST (MPMCQueue, single_thread_enqdeq)
	TEST (MPMCQueue, tryenq_capacity_test)
	TEST (MPMCQueue, enq_capacity_test)
template<template< typename > class Atom, bool Dynamic = false>
void	runTryEnqDeqThread (int numThreads, int n, MPMCQueue< int, Atom, Dynamic > &cq, std::atomic< uint64_t > &sum, int t)
template<template< typename > class Atom, bool Dynamic = false>
void	runTryEnqDeqTest (int numThreads, int numOps)
	TEST (MPMCQueue, mt_try_enq_deq)
	TEST (MPMCQueue, mt_try_enq_deq_dynamic)
	TEST (MPMCQueue, mt_try_enq_deq_emulated_futex)
	TEST (MPMCQueue, mt_try_enq_deq_emulated_futex_dynamic)
	TEST (MPMCQueue, mt_try_enq_deq_deterministic)
uint64_t	nowMicro ()
template<typename Q >
string	producerConsumerBench (Q &&queue, string qName, int numProducers, int numConsumers, int numOps, WriteMethodCaller< Q > &writer, bool ignoreContents=false)
template<bool Dynamic = false>
void	runMtProdConsDeterministic (long seed)
void	runMtProdConsDeterministicDynamic (long seed, uint32_t prods, uint32_t cons, uint32_t numOps, size_t cap, size_t minCap, size_t mult)
	TEST (MPMCQueue, mt_prod_cons_deterministic)
	TEST (MPMCQueue, mt_prod_cons_deterministic_dynamic)
template<typename T >
void	setFromEnv (T &var, const char *envvar)
	TEST (MPMCQueue, mt_prod_cons_deterministic_dynamic_with_arguments)
template<bool Dynamic = false>
void	runMtProdCons ()
	TEST (MPMCQueue, mt_prod_cons)
	TEST (MPMCQueue, mt_prod_cons_dynamic)
template<bool Dynamic = false>
void	runMtProdConsEmulatedFutex ()
	TEST (MPMCQueue, mt_prod_cons_emulated_futex)
	TEST (MPMCQueue, mt_prod_cons_emulated_futex_dynamic)
template<template< typename > class Atom, bool Dynamic = false>
void	runNeverFailThread (int numThreads, int n, MPMCQueue< int, Atom, Dynamic > &cq, std::atomic< uint64_t > &sum, int t)
template<template< typename > class Atom, bool Dynamic = false>
uint64_t	runNeverFailTest (int numThreads, int numOps)
template<template< typename > class Atom, bool Dynamic = false>
void	runMtNeverFail (std::vector< int > &nts, int n)
	TEST (MPMCQueue, mt_never_fail)
	TEST (MPMCQueue, mt_never_fail_emulated_futex)
template<bool Dynamic = false>
void	runMtNeverFailDeterministic (std::vector< int > &nts, int n, long seed)
	TEST (MPMCQueue, mt_never_fail_deterministic)
template<class Clock , template< typename > class Atom, bool Dynamic>
void	runNeverFailUntilThread (int numThreads, int n, MPMCQueue< int, Atom, Dynamic > &cq, std::atomic< uint64_t > &sum, int t)
template<class Clock , template< typename > class Atom, bool Dynamic = false>
uint64_t	runNeverFailTest (int numThreads, int numOps)
template<bool Dynamic = false>
void	runMtNeverFailUntilSystem (std::vector< int > &nts, int n)
	TEST (MPMCQueue, mt_never_fail_until_system)
template<bool Dynamic = false>
void	runMtNeverFailUntilSteady (std::vector< int > &nts, int n)
	TEST (MPMCQueue, mt_never_fail_until_steady)
static int	lc_outstanding ()
static void	lc_snap ()
static void	lc_step (int lineno, int what=NOTHING, int what2=NOTHING)
template<typename R >
void	runPerfectForwardingTest ()
	TEST (MPMCQueue, perfect_forwarding)
	TEST (MPMCQueue, perfect_forwarding_relocatable)
template<bool Dynamic = false>
void	run_queue_moving ()
	TEST (MPMCQueue, queue_moving)
	TEST (MPMCQueue, queue_moving_dynamic)
	TEST (MPMCQueue, explicit_zero_capacity_fail)
template<bool Dynamic>
void	testTryReadUntil ()
template<bool Dynamic>
void	testTryWriteUntil ()
	TEST (MPMCQueue, try_read_until)
	TEST (MPMCQueue, try_read_until_dynamic)
	TEST (MPMCQueue, try_write_until)
	TEST (MPMCQueue, try_write_until_dynamic)
template<bool Dynamic>
void	testTimeout (MPMCQueue< int, std::atomic, Dynamic > &q)
	TEST (MPMCQueue, try_write_until_timeout)
	TEST (MPMCQueue, must_fail_try_write_until_dynamic)

Variables
static FOLLY_TLS int	lc_counts [MAX_LIFECYCLE_EVENT]
static FOLLY_TLS int	lc_prev [MAX_LIFECYCLE_EVENT]

Macro Definition Documentation

#define LIFECYCLE_STEP

(

...

)

lc_step(__LINE__, __VA_ARGS__)

Definition at line 918 of file MPMCQueueTest.cpp.

Referenced by run_queue_moving(), and runPerfectForwardingTest().

#define PC_BENCH	(		q,
			np,
			nc,
			...
	)		producerConsumerBench(q, #q, (np), (nc), __VA_ARGS__)

Definition at line 640 of file MPMCQueueTest.cpp.

Referenced by runMtProdCons(), and runMtProdConsEmulatedFutex().

Typedef Documentation

typedef DeterministicSchedule DSched

Definition at line 47 of file MPMCQueueTest.cpp.

Enumeration Type Documentation

enum LifecycleEvent

Enumerator
NOTHING
DEFAULT_CONSTRUCTOR
COPY_CONSTRUCTOR
MOVE_CONSTRUCTOR
TWO_ARG_CONSTRUCTOR
COPY_OPERATOR
MOVE_OPERATOR
DESTRUCTOR
MAX_LIFECYCLE_EVENT

Definition at line 891 of file MPMCQueueTest.cpp.

                    {
  NOTHING = -1,
  DEFAULT_CONSTRUCTOR,
  COPY_CONSTRUCTOR,
  MOVE_CONSTRUCTOR,
  TWO_ARG_CONSTRUCTOR,
  COPY_OPERATOR,
  MOVE_OPERATOR,
  DESTRUCTOR,
  MAX_LIFECYCLE_EVENT
};

Function Documentation

void intrusive_ptr_add_ref

(

RefCounted const *

p

)

Definition at line 142 of file MPMCQueueTest.cpp.

References RefCounted::rc.

                                                {
  p->rc++;
}

void intrusive_ptr_release

(

RefCounted const *

p

)

Definition at line 146 of file MPMCQueueTest.cpp.

References RefCounted::rc.

                                                {
  if (--(p->rc) == 0) {
    delete p;
  }
}

static int lc_outstanding ( )

static

Definition at line 906 of file MPMCQueueTest.cpp.

References COPY_CONSTRUCTOR, DEFAULT_CONSTRUCTOR, DESTRUCTOR, lc_counts, MOVE_CONSTRUCTOR, and TWO_ARG_CONSTRUCTOR.

Referenced by run_queue_moving(), runPerfectForwardingTest(), and Lifecycle< R >::~Lifecycle().

                            {
  return lc_counts[DEFAULT_CONSTRUCTOR] + lc_counts[COPY_CONSTRUCTOR] +
      lc_counts[MOVE_CONSTRUCTOR] + lc_counts[TWO_ARG_CONSTRUCTOR] -
      lc_counts[DESTRUCTOR];
}

static void lc_snap ( )

static

Definition at line 912 of file MPMCQueueTest.cpp.

References i, lc_counts, lc_prev, and MAX_LIFECYCLE_EVENT.

Referenced by lc_step(), run_queue_moving(), and runPerfectForwardingTest().

                      {
  for (int i = 0; i < MAX_LIFECYCLE_EVENT; ++i) {
    lc_prev[i] = lc_counts[i];
  }
}

static void lc_step ( int  lineno, int  what = NOTHING, int  what2 = NOTHING  )

static

Definition at line 920 of file MPMCQueueTest.cpp.

References EXPECT_EQ, i, lc_counts, lc_prev, lc_snap(), and MAX_LIFECYCLE_EVENT.

                                                                         {
  for (int i = 0; i < MAX_LIFECYCLE_EVENT; ++i) {
    int delta = i == what || i == what2 ? 1 : 0;
    EXPECT_EQ(lc_counts[i] - lc_prev[i], delta)
        << "lc_counts[" << i << "] - lc_prev[" << i << "] was "
        << (lc_counts[i] - lc_prev[i]) << ", expected " << delta
        << ", from line " << lineno;
  }
  lc_snap();
}

uint64_t nowMicro

(

)

Definition at line 360 of file MPMCQueueTest.cpp.

References uint64_t.

Referenced by producerConsumerBench(), and runNeverFailTest().

                    {
  timeval tv;
  gettimeofday(&tv, nullptr);
  return static_cast<uint64_t>(tv.tv_sec) * 1000000 + tv.tv_usec;
}

template<typename Q >

string producerConsumerBench	(	Q &&	queue,
		string	qName,
		int	numProducers,
		int	numConsumers,
		int	numOps,
		WriteMethodCaller< Q > &	writer,
		bool	ignoreContents = false
	)

Definition at line 424 of file MPMCQueueTest.cpp.

References allocated, WriteMethodCaller< Q >::callWrite(), upload::dest, EXPECT_EQ, EXPECT_FALSE, failed, i, folly::test::DeterministicSchedule::join(), WriteMethodCaller< Q >::methodName(), nowMicro(), folly::sformat(), sum(), folly::pushmi::detail::t, folly::test::DeterministicSchedule::thread(), and uint64_t.

Referenced by runMtProdConsDeterministic(), and runMtProdConsDeterministicDynamic().

                                 {
  Q& q = queue;

  struct rusage beginUsage;
  getrusage(RUSAGE_SELF, &beginUsage);

  auto beginMicro = nowMicro();

  uint64_t n = numOps;
  std::atomic<uint64_t> sum(0);
  std::atomic<uint64_t> failed(0);

  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) {
        while (!writer.callWrite(q, i)) {
          ++failed;
        }
      }
    });
  }

  vector<std::thread> consumers(numConsumers);
  for (int t = 0; t < numConsumers; ++t) {
    consumers[t] = DSched::thread([&, t] {
      uint64_t localSum = 0;
      for (int i = t; i < numOps; i += numConsumers) {
        int dest = -1;
        q.blockingRead(dest);
        EXPECT_FALSE(dest == -1);
        localSum += dest;
      }
      sum += localSum;
    });
  }

  for (auto& t : producers) {
    DSched::join(t);
  }
  for (auto& t : consumers) {
    DSched::join(t);
  }
  if (!ignoreContents) {
    EXPECT_EQ(n * (n - 1) / 2 - sum, 0);
  }

  auto endMicro = nowMicro();

  struct rusage endUsage;
  getrusage(RUSAGE_SELF, &endUsage);

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

template<template< typename > class Atom>

void run_mt_sequencer_test	(	int	numThreads,
		int	numOps,
		uint32_t	init
	)

Definition at line 67 of file MPMCQueueTest.cpp.

References folly::netops::bind(), EXPECT_EQ, i, folly::test::DeterministicSchedule::join(), folly::gen::seq(), folly::test::DeterministicSchedule::thread(), and threads.

                                                                      {
  TurnSequencer<Atom> seq(init);
  Atom<uint32_t> spinThreshold(0);

  int prev = -1;
  vector<std::thread> threads(numThreads);
  for (int i = 0; i < numThreads; ++i) {
    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) {
    DSched::join(thr);
  }

  EXPECT_EQ(prev, numOps - 1);
}

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
	)

Definition at line 50 of file MPMCQueueTest.cpp.

References Atom, and EXPECT_EQ.

           {
  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<bool Dynamic = false>

void run_queue_moving

(

)

Definition at line 1056 of file MPMCQueueTest.cpp.

References a, b, c, COPY_CONSTRUCTOR, DEFAULT_CONSTRUCTOR, DESTRUCTOR, EXPECT_EQ, EXPECT_TRUE, lc_outstanding(), lc_snap(), LIFECYCLE_STEP, folly::gen::move, MOVE_CONSTRUCTOR, MOVE_OPERATOR, NOTHING, and folly::f14::swap().

Referenced by TEST().

                        {
  lc_snap();
  EXPECT_EQ(lc_outstanding(), 0);

  {
    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>, std::atomic, Dynamic> b =
        std::move(a);
    LIFECYCLE_STEP(NOTHING);
    EXPECT_EQ(a.capacity(), 0);
    EXPECT_EQ(a.size(), 0);
    EXPECT_EQ(b.capacity(), 50);
    EXPECT_EQ(b.size(), 1);

    b.blockingWrite();
    LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);

    // move operator
    MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> c;
    LIFECYCLE_STEP(NOTHING);
    c = std::move(b);
    LIFECYCLE_STEP(NOTHING);
    EXPECT_EQ(c.capacity(), 50);
    EXPECT_EQ(c.size(), 2);

    {
      Lifecycle<std::false_type> dst;
      LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
      c.blockingRead(dst);
      LIFECYCLE_STEP(DESTRUCTOR, MOVE_OPERATOR);

      {
        // swap
        MPMCQueue<Lifecycle<std::false_type>, std::atomic, Dynamic> d(10);
        LIFECYCLE_STEP(NOTHING);
        std::swap(c, d);
        LIFECYCLE_STEP(NOTHING);
        EXPECT_EQ(c.capacity(), 10);
        EXPECT_TRUE(c.isEmpty());
        EXPECT_EQ(d.capacity(), 50);
        EXPECT_EQ(d.size(), 1);

        d.blockingRead(dst);
        LIFECYCLE_STEP(DESTRUCTOR, MOVE_OPERATOR);

        c.blockingWrite(dst);
        LIFECYCLE_STEP(COPY_CONSTRUCTOR);

        d.blockingWrite(std::move(dst));
        LIFECYCLE_STEP(MOVE_CONSTRUCTOR);
      } // d goes out of scope
      LIFECYCLE_STEP(DESTRUCTOR);
    } // dst goes out of scope
    LIFECYCLE_STEP(DESTRUCTOR);
  } // c goes out of scope
  LIFECYCLE_STEP(DESTRUCTOR);
}

template<bool Dynamic = false, typename T >

void runElementTypeTest

(

T &&

src

)

Definition at line 112 of file MPMCQueueTest.cpp.

References upload::dest, EXPECT_TRUE, folly::gen::move, now(), and T.

Referenced by TEST().

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

template<template< typename > class Atom, bool Dynamic = false>

void runMtNeverFail	(	std::vector< int > &	nts,
		int	n
	)

Definition at line 761 of file MPMCQueueTest.cpp.

References folly::INFO, and uint64_t.

                                                {
  for (int nt : nts) {
    uint64_t elapsed = runNeverFailTest<Atom, Dynamic>(nt, n);
    LOG(INFO) << (elapsed * 1000.0) / (n * 2) << " nanos per op with " << nt
              << " threads";
  }
}

template<bool Dynamic = false>

void runMtNeverFailDeterministic	(	std::vector< int > &	nts,
		int	n,
		long	seed
	)

Definition at line 787 of file MPMCQueueTest.cpp.

References folly::INFO, seed, folly::test::DeterministicSchedule::uniform(), and folly::test::DeterministicSchedule::uniformSubset().

Referenced by TEST().

                                                                        {
  LOG(INFO) << "using seed " << seed;
  for (int nt : nts) {
    {
      DSched sched(DSched::uniform(seed));
      runNeverFailTest<DeterministicAtomic, Dynamic>(nt, n);
    }
    {
      DSched sched(DSched::uniformSubset(seed, 2));
      runNeverFailTest<DeterministicAtomic, Dynamic>(nt, n);
    }
  }
}

template<bool Dynamic = false>

void runMtNeverFailUntilSteady	(	std::vector< int > &	nts,
		int	n
	)

Definition at line 875 of file MPMCQueueTest.cpp.

References folly::INFO, and uint64_t.

Referenced by TEST().

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

template<bool Dynamic = false>

void runMtNeverFailUntilSystem	(	std::vector< int > &	nts,
		int	n
	)

Definition at line 858 of file MPMCQueueTest.cpp.

References folly::INFO, and uint64_t.

Referenced by TEST().

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

template<bool Dynamic = false>

void runMtProdCons

(

)

Definition at line 644 of file MPMCQueueTest.cpp.

References folly::INFO, folly::make_unique(), PC_BENCH, and setFromEnv().

Referenced by TEST().

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

template<bool Dynamic = false>

void runMtProdConsDeterministic

(

long

seed

)

Definition at line 504 of file MPMCQueueTest.cpp.

References folly::INFO, folly::make_unique(), producerConsumerBench(), and folly::test::DeterministicSchedule::uniform().

Referenced by TEST().

                                           {
  // 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
	)

Definition at line 569 of file MPMCQueueTest.cpp.

References folly::INFO, folly::make_unique(), producerConsumerBench(), and folly::test::DeterministicSchedule::uniform().

Referenced by TEST().

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

template<bool Dynamic = false>

void runMtProdConsEmulatedFutex

(

)

Definition at line 679 of file MPMCQueueTest.cpp.

References folly::INFO, folly::make_unique(), and PC_BENCH.

Referenced by TEST().

                                  {
  using QueueType = MPMCQueue<int, EmulatedFutexAtomic, Dynamic>;

  int n = 100000;
  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, bool Dynamic = false>

uint64_t runNeverFailTest	(	int	numThreads,
		int	numOps
	)

Definition at line 733 of file MPMCQueueTest.cpp.

References Atom, folly::netops::bind(), EXPECT_EQ, EXPECT_TRUE, folly::test::DeterministicSchedule::join(), nowMicro(), sum(), folly::pushmi::detail::t, folly::test::DeterministicSchedule::thread(), threads, and uint64_t.

                                                      {
  // 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(
        runNeverFailThread<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<class Clock , template< typename > class Atom, bool Dynamic = false>

uint64_t runNeverFailTest	(	int	numThreads,
		int	numOps
	)

Definition at line 830 of file MPMCQueueTest.cpp.

References folly::netops::bind(), EXPECT_EQ, EXPECT_TRUE, folly::test::DeterministicSchedule::join(), nowMicro(), sum(), folly::pushmi::detail::t, folly::test::DeterministicSchedule::thread(), threads, and uint64_t.

                                                      {
  // 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<template< typename > class Atom, bool Dynamic = false>

void runNeverFailThread	(	int	numThreads,
		int	n,
		MPMCQueue< int, Atom, Dynamic > &	cq,
		std::atomic< uint64_t > &	sum,
		int	t
	)

Definition at line 713 of file MPMCQueueTest.cpp.

References Atom, upload::dest, EXPECT_TRUE, i, and uint64_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<class Clock , template< typename > class Atom, bool Dynamic>

void runNeverFailUntilThread	(	int	numThreads,
		int	n,
		MPMCQueue< int, Atom, Dynamic > &	cq,
		std::atomic< uint64_t > &	sum,
		int	t
	)

Definition at line 809 of file MPMCQueueTest.cpp.

References Atom, upload::dest, EXPECT_TRUE, i, now(), and uint64_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<typename R >

void runPerfectForwardingTest

(

)

Definition at line 973 of file MPMCQueueTest.cpp.

References COPY_CONSTRUCTOR, DEFAULT_CONSTRUCTOR, DESTRUCTOR, EXPECT_EQ, EXPECT_FALSE, EXPECT_TRUE, i, lc_outstanding(), lc_snap(), LIFECYCLE_STEP, folly::gen::move, MOVE_CONSTRUCTOR, MOVE_OPERATOR, NOTHING, TWO_ARG_CONSTRUCTOR, and value.

                                {
  lc_snap();
  EXPECT_EQ(lc_outstanding(), 0);

  {
    // Non-dynamic only. False positive for dynamic.
    MPMCQueue<Lifecycle<R>, std::atomic> queue(50);
    LIFECYCLE_STEP(NOTHING);

    for (int pass = 0; pass < 10; ++pass) {
      for (int i = 0; i < 10; ++i) {
        queue.blockingWrite();
        LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);

        queue.blockingWrite(1, "one");
        LIFECYCLE_STEP(TWO_ARG_CONSTRUCTOR);

        {
          Lifecycle<R> src;
          LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
          queue.blockingWrite(std::move(src));
          LIFECYCLE_STEP(MOVE_CONSTRUCTOR);
        }
        LIFECYCLE_STEP(DESTRUCTOR);

        {
          Lifecycle<R> src;
          LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
          queue.blockingWrite(src);
          LIFECYCLE_STEP(COPY_CONSTRUCTOR);
        }
        LIFECYCLE_STEP(DESTRUCTOR);

        EXPECT_TRUE(queue.write());
        LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);
      }

      EXPECT_EQ(queue.size(), 50);
      EXPECT_FALSE(queue.write(2, "two"));
      LIFECYCLE_STEP(NOTHING);

      for (int i = 0; i < 50; ++i) {
        {
          Lifecycle<R> node;
          LIFECYCLE_STEP(DEFAULT_CONSTRUCTOR);

          queue.blockingRead(node);
          if (R::value) {
            // relocatable, moved via memcpy
            LIFECYCLE_STEP(DESTRUCTOR);
          } else {
            LIFECYCLE_STEP(DESTRUCTOR, MOVE_OPERATOR);
          }
        }
        LIFECYCLE_STEP(DESTRUCTOR);
      }

      EXPECT_EQ(queue.size(), 0);
    }

    // put one element back before destruction
    {
      Lifecycle<R> src(3, "three");
      LIFECYCLE_STEP(TWO_ARG_CONSTRUCTOR);
      queue.write(std::move(src));
      LIFECYCLE_STEP(MOVE_CONSTRUCTOR);
    }
    LIFECYCLE_STEP(DESTRUCTOR); // destroy src
  }
  LIFECYCLE_STEP(DESTRUCTOR); // destroy queue

  EXPECT_EQ(lc_outstanding(), 0);
}

template<template< typename > class Atom, bool Dynamic = false>

void runTryEnqDeqTest	(	int	numThreads,
		int	numOps
	)

Definition at line 272 of file MPMCQueueTest.cpp.

References folly::netops::bind(), EXPECT_EQ, EXPECT_TRUE, folly::test::DeterministicSchedule::join(), sum(), folly::pushmi::detail::t, folly::test::DeterministicSchedule::thread(), threads, and uint64_t.

Referenced by TEST().

                                                  {
  // 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, Dynamic> cq(numThreads);

  uint64_t n = numOps;
  vector<std::thread> threads(numThreads);
  std::atomic<uint64_t> sum(0);
  for (int t = 0; t < numThreads; ++t) {
    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);
  }
  EXPECT_TRUE(cq.isEmpty());
  EXPECT_EQ(n * (n - 1) / 2 - sum, 0);
}

template<template< typename > class Atom, bool Dynamic = false>

void runTryEnqDeqThread	(	int	numThreads,
		int	n,
		MPMCQueue< int, Atom, Dynamic > &	cq,
		std::atomic< uint64_t > &	sum,
		int	t
	)

Definition at line 245 of file MPMCQueueTest.cpp.

References Atom, folly::pushmi::detail::t, and uint64_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<typename T >

void setFromEnv	(	T &	var,
		const char *	envvar
	)

Definition at line 614 of file MPMCQueueTest.cpp.

Referenced by runMtProdCons(), and TEST().

                                            {
  char* str = std::getenv(envvar);
  if (str) {
    var = atoi(str);
  }
}

TEST	(	MPMCQueue	,
		sequencer
	)

Definition at line 92 of file MPMCQueueTest.cpp.

                           {
  run_mt_sequencer_test<std::atomic>(1, 100, 0);
  run_mt_sequencer_test<std::atomic>(2, 100000, -100);
  run_mt_sequencer_test<std::atomic>(100, 10000, -100);
}

TEST	(	MPMCQueue	,
		sequencer_emulated_futex
	)

Definition at line 98 of file MPMCQueueTest.cpp.

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

Definition at line 104 of file MPMCQueueTest.cpp.

References folly::test::DeterministicSchedule::uniform().

                                         {
  DSched sched(DSched::uniform(0));
  run_mt_sequencer_test<DeterministicAtomic>(1, 100, -50);
  run_mt_sequencer_test<DeterministicAtomic>(2, 10000, (1 << 29) - 100);
  run_mt_sequencer_test<DeterministicAtomic>(10, 1000, -100);
}

TEST	(	MPMCQueue	,
		lots_of_element_types
	)

Definition at line 152 of file MPMCQueueTest.cpp.

References RefCounted::active_instances, EXPECT_EQ, and runElementTypeTest().

                                       {
  runElementTypeTest(10);
  runElementTypeTest(string("abc"));
  runElementTypeTest(std::make_pair(10, string("def")));
  runElementTypeTest(vector<string>{{"abc"}});
  runElementTypeTest(std::make_shared<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
	)

Definition at line 163 of file MPMCQueueTest.cpp.

References RefCounted::active_instances, EXPECT_EQ, RefCounted::RefCounted(), and string.

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

Definition at line 174 of file MPMCQueueTest.cpp.

References upload::dest, EXPECT_EQ, EXPECT_FALSE, EXPECT_TRUE, and i.

                                      {
  // 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) {
    for (int i = 0; i < 10; ++i) {
      EXPECT_TRUE(cq.write(i));
    }
    EXPECT_FALSE(cq.write(-1));
    EXPECT_FALSE(cq.isEmpty());
    EXPECT_EQ(cq.size(), 10);

    for (int i = 0; i < 5; ++i) {
      int dest = -1;
      EXPECT_TRUE(cq.read(dest));
      EXPECT_EQ(dest, i);
    }
    for (int i = 5; i < 10; ++i) {
      int dest = -1;
      cq.blockingRead(dest);
      EXPECT_EQ(dest, i);
    }
    int dest = -1;
    EXPECT_FALSE(cq.read(dest));
    EXPECT_EQ(dest, -1);

    EXPECT_TRUE(cq.isEmpty());
    EXPECT_EQ(cq.size(), 0);
  }
}

TEST	(	MPMCQueue	,
		tryenq_capacity_test
	)

Definition at line 207 of file MPMCQueueTest.cpp.

References EXPECT_FALSE, EXPECT_TRUE, and i.

                                      {
  // 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 (size_t i = 0; i < cap; ++i) {
      EXPECT_TRUE(cq.write(i));
    }
    EXPECT_FALSE(cq.write(100));
  }
}

TEST	(	MPMCQueue	,
		enq_capacity_test
	)

Definition at line 220 of file MPMCQueueTest.cpp.

References Atom, dummy(), EXPECT_EQ, i, folly::pushmi::detail::t, and folly::when().

                                   {
  // 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) {
      cq.blockingWrite(i);
    }
    int t = 0;
    int when;
    auto thr = std::thread([&] {
      cq.blockingWrite(100);
      when = t;
    });
    usleep(2000);
    t = 1;
    int dummy;
    cq.blockingRead(dummy);
    thr.join();
    EXPECT_EQ(when, 1);
  }
}

TEST	(	MPMCQueue	,
		mt_try_enq_deq
	)

Definition at line 297 of file MPMCQueueTest.cpp.

                                {
  int nts[] = {1, 3, 100};

  int n = 100000;
  for (int nt : nts) {
    runTryEnqDeqTest<std::atomic>(nt, n);
  }
}

TEST	(	MPMCQueue	,
		mt_try_enq_deq_dynamic
	)

Definition at line 306 of file MPMCQueueTest.cpp.

References runTryEnqDeqTest().

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

Definition at line 315 of file MPMCQueueTest.cpp.

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

Definition at line 324 of file MPMCQueueTest.cpp.

References runTryEnqDeqTest().

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

Definition at line 333 of file MPMCQueueTest.cpp.

References folly::INFO, runTryEnqDeqTest(), seed, folly::test::DeterministicSchedule::uniform(), and folly::test::DeterministicSchedule::uniformSubset().

                                              {
  int nts[] = {3, 10};

  long seed = 0;
  LOG(INFO) << "using seed " << seed;

  int n = 1000;
  for (int nt : nts) {
    {
      DSched sched(DSched::uniform(seed));
      runTryEnqDeqTest<DeterministicAtomic>(nt, n);
    }
    {
      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);
    }
  }
}

TEST	(	MPMCQueue	,
		mt_prod_cons_deterministic
	)

Definition at line 605 of file MPMCQueueTest.cpp.

References runMtProdConsDeterministic().

                                            {
  runMtProdConsDeterministic(0);
}

TEST	(	MPMCQueue	,
		mt_prod_cons_deterministic_dynamic
	)

Definition at line 609 of file MPMCQueueTest.cpp.

                                                    {
  runMtProdConsDeterministic<true>(0);
}

TEST	(	MPMCQueue	,
		mt_prod_cons_deterministic_dynamic_with_arguments
	)

Definition at line 621 of file MPMCQueueTest.cpp.

References runMtProdConsDeterministicDynamic(), seed, setFromEnv(), and uint32_t.

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

TEST	(	MPMCQueue	,
		mt_prod_cons
	)

Definition at line 670 of file MPMCQueueTest.cpp.

References runMtProdCons().

                              {
  runMtProdCons();
}

TEST	(	MPMCQueue	,
		mt_prod_cons_dynamic
	)

Definition at line 674 of file MPMCQueueTest.cpp.

References runMtProdCons().

                                      {
  runMtProdCons</* Dynamic = */ true>();
}

TEST	(	MPMCQueue	,
		mt_prod_cons_emulated_futex
	)

Definition at line 704 of file MPMCQueueTest.cpp.

References runMtProdConsEmulatedFutex().

                                             {
  runMtProdConsEmulatedFutex();
}

TEST	(	MPMCQueue	,
		mt_prod_cons_emulated_futex_dynamic
	)

Definition at line 708 of file MPMCQueueTest.cpp.

References Atom, and runMtProdConsEmulatedFutex().

                                                     {
  runMtProdConsEmulatedFutex</* Dynamic = */ true>();
}

TEST	(	MPMCQueue	,
		mt_never_fail
	)

Definition at line 774 of file MPMCQueueTest.cpp.

                               {
  std::vector<int> nts{1, 3, 100};
  int n = 100000;
  runMtNeverFail<std::atomic>(nts, n);
}

TEST	(	MPMCQueue	,
		mt_never_fail_emulated_futex
	)

Definition at line 780 of file MPMCQueueTest.cpp.

                                              {
  std::vector<int> nts{1, 3, 100};
  int n = 100000;
  runMtNeverFail<EmulatedFutexAtomic>(nts, n);
}

TEST	(	MPMCQueue	,
		mt_never_fail_deterministic
	)

Definition at line 801 of file MPMCQueueTest.cpp.

References Atom, runMtNeverFailDeterministic(), and seed.

                                             {
  std::vector<int> nts{3, 10};
  long seed = 0; // nowMicro() % 10000;
  int n = 1000;
  runMtNeverFailDeterministic(nts, n, seed);
}

TEST	(	MPMCQueue	,
		mt_never_fail_until_system
	)

Definition at line 868 of file MPMCQueueTest.cpp.

References runMtNeverFailUntilSystem().

                                            {
  std::vector<int> nts{1, 3, 100};
  int n = 100000;
  runMtNeverFailUntilSystem(nts, n);
}

TEST	(	MPMCQueue	,
		mt_never_fail_until_steady
	)

Definition at line 885 of file MPMCQueueTest.cpp.

References runMtNeverFailUntilSteady().

                                            {
  std::vector<int> nts{1, 3, 100};
  int n = 100000;
  runMtNeverFailUntilSteady(nts, n);
}

TEST	(	MPMCQueue	,
		perfect_forwarding
	)

Definition at line 1047 of file MPMCQueueTest.cpp.

                                    {
  runPerfectForwardingTest<std::false_type>();
}

TEST	(	MPMCQueue	,
		perfect_forwarding_relocatable
	)

Definition at line 1051 of file MPMCQueueTest.cpp.

                                                {
  runPerfectForwardingTest<std::true_type>();
}

TEST	(	MPMCQueue	,
		queue_moving
	)

Definition at line 1120 of file MPMCQueueTest.cpp.

References run_queue_moving().

                              {
  run_queue_moving();
}

TEST	(	MPMCQueue	,
		queue_moving_dynamic
	)

Definition at line 1124 of file MPMCQueueTest.cpp.

                                      {
  run_queue_moving<true>();
}

TEST	(	MPMCQueue	,
		explicit_zero_capacity_fail
	)

Definition at line 1128 of file MPMCQueueTest.cpp.

References ASSERT_THROW.

                                             {
  ASSERT_THROW(MPMCQueue<int> cq(0), std::invalid_argument);

  using DynamicMPMCQueueInt = MPMCQueue<int, std::atomic, true>;
  ASSERT_THROW(DynamicMPMCQueueInt cq(0), std::invalid_argument);
}

TEST	(	MPMCQueue	,
		try_read_until
	)

Definition at line 1208 of file MPMCQueueTest.cpp.

                                {
  testTryReadUntil<false>();
}

TEST	(	MPMCQueue	,
		try_read_until_dynamic
	)

Definition at line 1212 of file MPMCQueueTest.cpp.

                                        {
  testTryReadUntil<true>();
}

TEST	(	MPMCQueue	,
		try_write_until
	)

Definition at line 1216 of file MPMCQueueTest.cpp.

                                 {
  testTryWriteUntil<false>();
}

TEST	(	MPMCQueue	,
		try_write_until_dynamic
	)

Definition at line 1220 of file MPMCQueueTest.cpp.

                                         {
  testTryWriteUntil<true>();
}

TEST	(	MPMCQueue	,
		try_write_until_timeout
	)

Definition at line 1232 of file MPMCQueueTest.cpp.

                                         {
  folly::MPMCQueue<int, std::atomic, false> queue(1);
  testTimeout<false>(queue);
}

TEST	(	MPMCQueue	,
		must_fail_try_write_until_dynamic
	)

Definition at line 1237 of file MPMCQueueTest.cpp.

                                                   {
  folly::MPMCQueue<int, std::atomic, true> queue(200, 1, 2);
  testTimeout<true>(queue);
}

template<bool Dynamic>

void testTimeout

(

MPMCQueue< int, std::atomic, Dynamic > &

q

)

Definition at line 1225 of file MPMCQueueTest.cpp.

References now().

                                                        {
  CHECK(q.write(1));
  /* The following must not block forever */
  q.tryWriteUntil(
      std::chrono::system_clock::now() + std::chrono::microseconds(10000), 2);
}

template<bool Dynamic>

void testTryReadUntil

(

)

Definition at line 1136 of file MPMCQueueTest.cpp.

References b, folly::custom_stop_watch< Clock, Duration >::elapsed(), EXPECT_EQ, EXPECT_FALSE, EXPECT_TRUE, folly::custom_stop_watch< Clock, Duration >::getCheckpoint(), i, threads, and folly::detail::distributed_mutex::wait().

                        {
  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

(

)

Definition at line 1171 of file MPMCQueueTest.cpp.

References b, folly::custom_stop_watch< Clock, Duration >::elapsed(), EXPECT_EQ, EXPECT_FALSE, EXPECT_TRUE, folly::custom_stop_watch< Clock, Duration >::getCheckpoint(), i, threads, folly::detail::distributed_mutex::wait(), and x.

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

Variable Documentation

FOLLY_TLS int lc_counts[MAX_LIFECYCLE_EVENT]

static

Definition at line 903 of file MPMCQueueTest.cpp.

Referenced by lc_outstanding(), lc_snap(), lc_step(), Lifecycle< R >::Lifecycle(), Lifecycle< R >::operator=(), and Lifecycle< R >::~Lifecycle().

FOLLY_TLS int lc_prev[MAX_LIFECYCLE_EVENT]

static

Definition at line 904 of file MPMCQueueTest.cpp.

Referenced by lc_snap(), and lc_step().