UnboundedQueueTest.cpp File Reference

#include <folly/concurrency/UnboundedQueue.h>
#include <folly/MPMCQueue.h>
#include <folly/ProducerConsumerQueue.h>
#include <folly/portability/GTest.h>
#include <glog/logging.h>
#include <atomic>
#include <thread>

Go to the source code of this file.

Classes
class	MPMC< T >
class	PCQ< T >
struct	IntArray< M >

Typedefs
template<typename T , bool MayBlock>
using	USPSC = folly::USPSCQueue< T, MayBlock >
template<typename T , bool MayBlock>
using	UMPSC = folly::UMPSCQueue< T, MayBlock >
template<typename T , bool MayBlock>
using	USPMC = folly::USPMCQueue< T, MayBlock >
template<typename T , bool MayBlock>
using	UMPMC = folly::UMPMCQueue< T, MayBlock >
template<typename T , bool ignore>
using	FMPMC = MPMC< T >
template<typename T , bool ignore>
using	FPCQ = PCQ< T >

Functions
	DEFINE_bool (bench, false,"run benchmark")
	DEFINE_int32 (reps, 10,"number of reps")
	DEFINE_int32 (ops, 1000000,"number of operations per rep")
	DEFINE_int64 (capacity, 256 *1024,"capacity")
template<template< typename, bool > class Q, bool MayBlock>
void	basic_test ()
	TEST (UnboundedQueue, basic)
template<template< typename, bool > class Q, bool MayBlock>
void	timeout_test ()
	TEST (UnboundedQueue, timeout)
template<template< typename, bool > class Q, bool MayBlock>
void	peek_test ()
	TEST (UnboundedQueue, peek)
	TEST (UnboundedQueue, cleanup_on_destruction)
template<typename ProdFunc , typename ConsFunc , typename EndFunc >
uint64_t	run_once (int nprod, int ncons, const ProdFunc &prodFn, const ConsFunc &consFn, const EndFunc &endFn)
template<bool SingleProducer, bool SingleConsumer, bool MayBlock>
void	enq_deq_test (const int nprod, const int ncons)
	TEST (UnboundedQueue, enq_deq)
template<typename RepFunc >
uint64_t	runBench (const std::string &name, int ops, const RepFunc &repFn)
template<template< typename, bool > class Q, typename T , int Op>
uint64_t	bench (const int nprod, const int ncons, const std::string &name)
void	dottedLine ()
template<typename T >
void	type_benches (const int np, const int nc, const std::string &name)
void	benches (const int np, const int nc)
	TEST (UnboundedQueue, bench)

Typedef Documentation

template<typename T , bool ignore>

using FMPMC = MPMC<T>

Definition at line 425 of file UnboundedQueueTest.cpp.

template<typename T , bool ignore>

using FPCQ = PCQ<T>

Definition at line 456 of file UnboundedQueueTest.cpp.

template<typename T , bool MayBlock>

using UMPMC = folly::UMPMCQueue<T, MayBlock>

Definition at line 42 of file UnboundedQueueTest.cpp.

template<typename T , bool MayBlock>

using UMPSC = folly::UMPSCQueue<T, MayBlock>

Definition at line 36 of file UnboundedQueueTest.cpp.

template<typename T , bool MayBlock>

using USPMC = folly::USPMCQueue<T, MayBlock>

Definition at line 39 of file UnboundedQueueTest.cpp.

template<typename T , bool MayBlock>

using USPSC = folly::USPSCQueue<T, MayBlock>

Definition at line 33 of file UnboundedQueueTest.cpp.

Function Documentation

template<template< typename, bool > class Q, bool MayBlock>

void basic_test

(

)

Definition at line 45 of file UnboundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_FALSE, and ASSERT_TRUE.

                  {
  Q<int, MayBlock> q;
  ASSERT_TRUE(q.empty());
  ASSERT_EQ(q.size(), 0);
  int v = -1;
  ASSERT_FALSE(q.try_dequeue(v));

  q.enqueue(1);
  ASSERT_FALSE(q.empty());
  ASSERT_EQ(q.size(), 1);

  q.enqueue(2);
  ASSERT_EQ(q.size(), 2);
  ASSERT_FALSE(q.empty());

  ASSERT_TRUE(q.try_dequeue(v));
  ASSERT_EQ(v, 1);
  ASSERT_FALSE(q.empty());
  ASSERT_EQ(q.size(), 1);

  ASSERT_TRUE(q.try_dequeue(v));
  ASSERT_EQ(v, 2);
  ASSERT_TRUE(q.empty());
  ASSERT_EQ(q.size(), 0);
}

template<template< typename, bool > class Q, typename T , int Op>

uint64_t bench	(	const int	nprod,
		const int	ncons,
		const std::string &	name
	)

Definition at line 349 of file UnboundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_TRUE, i, ops, run_once(), runBench(), sum(), T, uint64_t, UNLIKELY, and v.

                                                                        {
  int ops = FLAGS_ops;
  auto repFn = [&] {
    Q<T, Op == 3 || Op == 4 || Op == 5> q;
    std::atomic<uint64_t> sum(0);
    auto prod = [&](int tid) {
      for (int i = tid; i < ops; i += nprod) {
        q.enqueue(i);
      }
    };
    auto cons = [&](int tid) {
      uint64_t mysum = 0;
      for (int i = tid; i < ops; i += ncons) {
        T v;
        if (Op == 0 || Op == 3) {
          while (UNLIKELY(!q.try_dequeue(v))) {
            /* keep trying */;
          }
        } else if (Op == 1 || Op == 4) {
          auto duration = std::chrono::microseconds(1000);
          while (UNLIKELY(!q.try_dequeue_for(v, duration))) {
            /* keep trying */;
          }
        } else {
          ASSERT_TRUE(Op == 2 || Op == 5);
          q.dequeue(v);
        }
        if (nprod == 1 && ncons == 1) {
          DCHECK_EQ(int(v), i);
        }
        mysum += v;
      }
      sum.fetch_add(mysum);
    };
    auto endfn = [&] {
      uint64_t expected = ops;
      expected *= ops - 1;
      expected /= 2;
      ASSERT_EQ(sum.load(), expected);
    };
    return run_once(nprod, ncons, prod, cons, endfn);
  };
  return runBench(name, ops, repFn);
}

void benches	(	const int	np,
		const int	nc
	)

Definition at line 526 of file UnboundedQueueTest.cpp.

Referenced by TEST().

                                         {
  std::cout << "====================== " << std::setw(2) << np << " prod"
            << "  " << std::setw(2) << nc << " cons"
            << " ======================" << std::endl;
  type_benches<uint32_t>(np, nc, "=== uint32_t ======");
  // Benchmarks for other element sizes can be added as follows:
  //   type_benches<IntArray<4>>(np, nc, "=== IntArray<4> ===");
}

DEFINE_bool	(	bench	,
		false	,
		"run benchmark"
	)

DEFINE_int32	(	reps	,
		10	,
		"number of reps"
	)

DEFINE_int32	(	ops	,
		1000000	,
		"number of operations per rep"
	)

DEFINE_int64	(	capacity	,
		256 *	1024,
		"capacity"
	)

void dottedLine

(

)

Definition at line 472 of file UnboundedQueueTest.cpp.

Referenced by TEST(), and type_benches().

                  {
  std::cout << ".............................................................."
            << std::endl;
}

template<bool SingleProducer, bool SingleConsumer, bool MayBlock>

void enq_deq_test	(	const int	nprod,
		const int	ncons
	)

Definition at line 216 of file UnboundedQueueTest.cpp.

References ASSERT_EQ, folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::dequeue(), folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::enqueue(), i, ops, run_once(), sum(), folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::try_dequeue(), folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::try_dequeue_for(), folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::try_peek(), uint64_t, and v.

                                                    {
  if (SingleProducer) {
    ASSERT_EQ(nprod, 1);
  }
  if (SingleConsumer) {
    ASSERT_EQ(ncons, 1);
  }

  int ops = 1000;
  folly::UnboundedQueue<int, SingleProducer, SingleConsumer, MayBlock, 4> q;
  std::atomic<uint64_t> sum(0);

  auto prod = [&](int tid) {
    for (int i = tid; i < ops; i += nprod) {
      q.enqueue(i);
    }
  };

  auto cons = [&](int tid) {
    uint64_t mysum = 0;
    for (int i = tid; i < ops; i += ncons) {
      int v = -1;
      int vpeek = -1;

      if (SingleConsumer) {
        while (true) {
          auto res = q.try_peek();
          if (res) {
            vpeek = *res;
            break;
          }
        }
      }
      if ((i % 3) == 0) {
        while (!q.try_dequeue(v)) {
          /* keep trying */;
        }
      } else if ((i % 3) == 1) {
        auto duration = std::chrono::milliseconds(1);
        while (!q.try_dequeue_for(v, duration)) {
          /* keep trying */;
        }
      } else {
        q.dequeue(v);
      }
      if (nprod == 1 && ncons == 1) {
        ASSERT_EQ(v, i);
      }
      if (SingleConsumer) {
        ASSERT_EQ(v, vpeek);
      }
      mysum += v;
    }
    sum.fetch_add(mysum);
  };

  auto endfn = [&] {
    uint64_t expected = ops;
    expected *= ops - 1;
    expected /= 2;
    ASSERT_EQ(sum.load(), expected);
  };
  run_once(nprod, ncons, prod, cons, endfn);
}

template<template< typename, bool > class Q, bool MayBlock>

void peek_test

(

)

Definition at line 107 of file UnboundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_FALSE, ASSERT_TRUE, i, and v.

                 {
  Q<int, MayBlock> q;
  auto res = q.try_peek();
  ASSERT_FALSE(res);
  for (int i = 0; i < 1000; ++i) {
    q.enqueue(i);
  }
  for (int i = 0; i < 700; ++i) {
    int v;
    q.dequeue(v);
  }
  res = q.try_peek();
  ASSERT_TRUE(res);
  ASSERT_EQ(*res, 700);
}

template<typename ProdFunc , typename ConsFunc , typename EndFunc >

uint64_t run_once ( int  nprod, int  ncons, const ProdFunc &  prodFn, const ConsFunc &  consFn, const EndFunc &  endFn  )

inline

Definition at line 158 of file UnboundedQueueTest.cpp.

References count, i, now(), and start.

Referenced by bench(), and enq_deq_test().

                          {
  std::atomic<bool> start{false};
  std::atomic<int> ready{0};

  /* producers */
  std::vector<std::thread> prodThr(nprod);
  for (int tid = 0; tid < nprod; ++tid) {
    prodThr[tid] = std::thread([&, tid] {
      ++ready;
      while (!start.load()) {
        /* spin */;
      }
      prodFn(tid);
    });
  }

  /* consumers */
  std::vector<std::thread> consThr(ncons);
  for (int tid = 0; tid < ncons; ++tid) {
    consThr[tid] = std::thread([&, tid] {
      ++ready;
      while (!start.load()) {
        /* spin */;
      }
      consFn(tid);
    });
  }

  /* wait for all producers and consumers to be ready */
  while (ready.load() < (nprod + ncons)) {
    /* spin */;
  }

  /* begin time measurement */
  auto tbegin = std::chrono::steady_clock::now();
  start.store(true);

  /* wait for completion */
  for (int i = 0; i < nprod; ++i) {
    prodThr[i].join();
  }
  for (int i = 0; i < ncons; ++i) {
    consThr[i].join();
  }

  /* end time measurement */
  auto tend = std::chrono::steady_clock::now();
  endFn();
  return std::chrono::duration_cast<std::chrono::nanoseconds>(tend - tbegin)
      .count();
}

template<typename RepFunc >

uint64_t runBench	(	const std::string &	name,
		int	ops,
		const RepFunc &	repFn
	)

Definition at line 317 of file UnboundedQueueTest.cpp.

References max, min, name, string, sum(), T, uint64_t, and folly::unit.

Referenced by bench().

                                                                        {
  int reps = FLAGS_reps;
  uint64_t min = UINTMAX_MAX;
  uint64_t max = 0;
  uint64_t sum = 0;

  repFn(); // sometimes first run is outlier
  for (int r = 0; r < reps; ++r) {
    uint64_t dur = repFn();
    sum += dur;
    min = std::min(min, dur);
    max = std::max(max, dur);
    // if each rep takes too long run at least 3 reps
    const uint64_t minute = 60000000000UL;
    if (sum > minute && r >= 2) {
      reps = r + 1;
      break;
    }
  }

  const std::string unit = " ns";
  uint64_t avg = sum / reps;
  uint64_t res = min;
  std::cout << name;
  std::cout << "   " << std::setw(4) << max / ops << unit;
  std::cout << "   " << std::setw(4) << avg / ops << unit;
  std::cout << "   " << std::setw(4) << res / ops << unit;
  std::cout << std::endl;
  return res;
}

TEST	(	UnboundedQueue	,
		basic
	)

Definition at line 71 of file UnboundedQueueTest.cpp.

                            {
  basic_test<USPSC, false>();
  basic_test<UMPSC, false>();
  basic_test<USPMC, false>();
  basic_test<UMPMC, false>();
  basic_test<USPSC, true>();
  basic_test<UMPSC, true>();
  basic_test<USPMC, true>();
  basic_test<UMPMC, true>();
}

TEST	(	UnboundedQueue	,
		timeout
	)

Definition at line 95 of file UnboundedQueueTest.cpp.

                              {
  timeout_test<USPSC, false>();
  timeout_test<UMPSC, false>();
  timeout_test<USPMC, false>();
  timeout_test<UMPMC, false>();
  timeout_test<USPSC, true>();
  timeout_test<UMPSC, true>();
  timeout_test<USPMC, true>();
  timeout_test<UMPMC, true>();
}

TEST	(	UnboundedQueue	,
		peek
	)

Definition at line 123 of file UnboundedQueueTest.cpp.

                           {
  peek_test<USPSC, false>();
  peek_test<UMPSC, false>();
  peek_test<USPSC, true>();
  peek_test<UMPSC, true>();
}

TEST	(	UnboundedQueue	,
		cleanup_on_destruction
	)

Definition at line 130 of file UnboundedQueueTest.cpp.

References count, folly::UnboundedQueue< T, SingleProducer, SingleConsumer, MayBlock, LgSegmentSize, LgAlign, Atom >::enqueue(), folly::exchange(), EXPECT_EQ, testing::gmock_matchers_test::foo, i, folly::gen::move, and folly::pushmi::__adl::noexcept().

                                             {
  struct Foo {
    int* p_{nullptr};
    explicit Foo(int* p) : p_(p) {}
    Foo(Foo&& o) noexcept : p_(std::exchange(o.p_, nullptr)) {}
    ~Foo() {
      if (p_) {
        ++(*p_);
      }
    }
    Foo& operator=(Foo&& o) noexcept {
      p_ = std::exchange(o.p_, nullptr);
      return *this;
    }
  };
  int count = 0;
  int num = 3;
  {
    folly::UMPMCQueue<Foo, false> q;
    for (int i = 0; i < num; ++i) {
      Foo foo(&count);
      q.enqueue(std::move(foo));
    }
  }
  EXPECT_EQ(count, num);
}

TEST	(	UnboundedQueue	,
		enq_deq
	)

Definition at line 281 of file UnboundedQueueTest.cpp.

                              {
  /* SPSC */
  enq_deq_test<true, true, false>(1, 1);
  enq_deq_test<true, true, true>(1, 1);
  /* MPSC */
  enq_deq_test<false, true, false>(1, 1);
  enq_deq_test<false, true, true>(1, 1);
  enq_deq_test<false, true, false>(2, 1);
  enq_deq_test<false, true, true>(2, 1);
  enq_deq_test<false, true, false>(10, 1);
  enq_deq_test<false, true, true>(10, 1);
  /* SPMC */
  enq_deq_test<true, false, false>(1, 1);
  enq_deq_test<true, false, true>(1, 1);
  enq_deq_test<true, false, false>(1, 2);
  enq_deq_test<true, false, true>(1, 2);
  enq_deq_test<true, false, false>(1, 10);
  enq_deq_test<true, false, true>(1, 10);
  /* MPMC */
  enq_deq_test<false, false, false>(1, 1);
  enq_deq_test<false, false, true>(1, 1);
  enq_deq_test<false, false, false>(2, 1);
  enq_deq_test<false, false, true>(2, 1);
  enq_deq_test<false, false, false>(10, 1);
  enq_deq_test<false, false, true>(10, 1);
  enq_deq_test<false, false, false>(1, 2);
  enq_deq_test<false, false, true>(1, 2);
  enq_deq_test<false, false, false>(1, 10);
  enq_deq_test<false, false, true>(1, 10);
  enq_deq_test<false, false, false>(2, 2);
  enq_deq_test<false, false, true>(2, 2);
  enq_deq_test<false, false, false>(10, 10);
  enq_deq_test<false, false, true>(10, 10);
}

TEST	(	UnboundedQueue	,
		bench
	)

Definition at line 535 of file UnboundedQueueTest.cpp.

References benches(), and dottedLine().

                            {
  if (!FLAGS_bench) {
    return;
  }
  std::cout << "=============================================================="
            << std::endl;
  std::cout << std::setw(2) << FLAGS_reps << " reps of " << std::setw(8)
            << FLAGS_ops << " handoffs\n";
  dottedLine();
  std::cout << "$ numactl -N 1 $dir/unbounded_queue_test --bench\n";
  dottedLine();
  std::cout << "Using capacity " << FLAGS_capacity
            << " for folly::ProducerConsumerQueue and\n"
            << "folly::MPMCQueue\n";
  std::cout << "=============================================================="
            << std::endl;
  std::cout << "Test name                         Max time  Avg time  Min time"
            << std::endl;

  for (int nc : {1, 2, 4, 8, 16, 32}) {
    int np = 1;
    benches(np, nc);
  }

  for (int np : {1, 2, 4, 8, 16, 32}) {
    int nc = 1;
    benches(np, nc);
  }

  for (int np : {2, 4, 8, 16, 32}) {
    for (int nc : {2, 4, 8, 16, 32}) {
      benches(np, nc);
    }
  }
}

template<template< typename, bool > class Q, bool MayBlock>

void timeout_test

(

)

Definition at line 83 of file UnboundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_FALSE, ASSERT_TRUE, now(), and v.

                    {
  Q<int, MayBlock> q;
  int v;
  ASSERT_FALSE(q.try_dequeue_until(
      v, std::chrono::steady_clock::now() + std::chrono::microseconds(1)));
  ASSERT_FALSE(q.try_dequeue_for(v, std::chrono::microseconds(1)));
  q.enqueue(10);
  ASSERT_TRUE(q.try_dequeue_until(
      v, std::chrono::steady_clock::now() + std::chrono::microseconds(1)));
  ASSERT_EQ(v, 10);
}

template<typename T >

void type_benches	(	const int	np,
		const int	nc,
		const std::string &	name
	)

Definition at line 478 of file UnboundedQueueTest.cpp.

References dottedLine().

                                                                     {
  std::cout << name
            << "===========================================" << std::endl;
  if (np == 1 && nc == 1) {
    bench<USPSC, T, 0>(1, 1, "Unbounded SPSC try   spin only  ");
    bench<USPSC, T, 1>(1, 1, "Unbounded SPSC timed spin only  ");
    bench<USPSC, T, 2>(1, 1, "Unbounded SPSC wait  spin only  ");
    bench<USPSC, T, 3>(1, 1, "Unbounded SPSC try   may block  ");
    bench<USPSC, T, 4>(1, 1, "Unbounded SPSC timed may block  ");
    bench<USPSC, T, 5>(1, 1, "Unbounded SPSC wait  may block  ");
    dottedLine();
  }
  if (nc == 1) {
    bench<UMPSC, T, 0>(np, 1, "Unbounded MPSC try   spin only  ");
    bench<UMPSC, T, 1>(np, 1, "Unbounded MPSC timed spin only  ");
    bench<UMPSC, T, 2>(np, 1, "Unbounded MPSC wait  spin only  ");
    bench<UMPSC, T, 3>(np, 1, "Unbounded MPSC try   may block  ");
    bench<UMPSC, T, 4>(np, 1, "Unbounded MPSC timed may block  ");
    bench<UMPSC, T, 5>(np, 1, "Unbounded MPSC wait  may block  ");
    dottedLine();
  }
  if (np == 1) {
    bench<USPMC, T, 0>(1, nc, "Unbounded SPMC try   spin only  ");
    bench<USPMC, T, 1>(1, nc, "Unbounded SPMC timed spin only  ");
    bench<USPMC, T, 2>(1, nc, "Unbounded SPMC wait  spin only  ");
    bench<USPMC, T, 3>(1, nc, "Unbounded SPMC try   may block  ");
    bench<USPMC, T, 4>(1, nc, "Unbounded SPMC timed may block  ");
    bench<USPMC, T, 5>(1, nc, "Unbounded SPMC wait  may block  ");
    dottedLine();
  }
  bench<UMPMC, T, 0>(np, nc, "Unbounded MPMC try   spin only  ");
  bench<UMPMC, T, 1>(np, nc, "Unbounded MPMC timed spin only  ");
  bench<UMPMC, T, 2>(np, nc, "Unbounded MPMC wait  spin only  ");
  bench<UMPMC, T, 3>(np, nc, "Unbounded MPMC try   may block  ");
  bench<UMPMC, T, 4>(np, nc, "Unbounded MPMC timed may block  ");
  bench<UMPMC, T, 5>(np, nc, "Unbounded MPMC wait  may block  ");
  dottedLine();
  if (np == 1 && nc == 1) {
    bench<FPCQ, T, 0>(1, 1, "folly::PCQ  read                ");
    dottedLine();
  }
  bench<FMPMC, T, 3>(np, nc, "folly::MPMC  read               ");
  bench<FMPMC, T, 4>(np, nc, "folly::MPMC  tryReadUntil       ");
  bench<FMPMC, T, 5>(np, nc, "folly::MPMC  blockingRead       ");
  std::cout << "=============================================================="
            << std::endl;
}