DynamicBoundedQueueTest.cpp File Reference

#include <folly/concurrency/DynamicBoundedQueue.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, typename WeightFn >
using	DSPSC = folly::DSPSCQueue< T, MayBlock, 8, 7, WeightFn >
template<typename T , bool MayBlock, typename WeightFn >
using	DMPSC = folly::DMPSCQueue< T, MayBlock, 8, 7, WeightFn >
template<typename T , bool MayBlock, typename WeightFn >
using	DSPMC = folly::DSPMCQueue< T, MayBlock, 8, 7, WeightFn >
template<typename T , bool MayBlock, typename WeightFn >
using	DMPMC = folly::DMPMCQueue< T, MayBlock, 8, 7, WeightFn >
template<typename T , bool , typename >
using	FMPMC = MPMC< T >
template<typename T , bool , typename >
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, 1000000,"capacity")
template<template< typename, bool, typename > class Q, bool MayBlock>
void	basic_test ()
	TEST (DynamicBoundedQueue, basic)
template<template< typename, bool, typename > class Q, bool MayBlock>
void	move_test ()
	TEST (DynamicBoundedQueue, move)
template<template< typename, bool, typename > class Q, bool MayBlock>
void	capacity_test ()
	TEST (DynamicBoundedQueue, capacity)
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 (DynamicBoundedQueue, enq_deq)
template<typename RepFunc >
uint64_t	runBench (const std::string &name, int ops, const RepFunc &repFn)
template<template< typename, bool, typename > 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 (DynamicBoundedQueue, bench)

Typedef Documentation

template<typename T , bool MayBlock, typename WeightFn >

using DMPMC = folly::DMPMCQueue<T, MayBlock, 8, 7, WeightFn>

Definition at line 42 of file DynamicBoundedQueueTest.cpp.

template<typename T , bool MayBlock, typename WeightFn >

using DMPSC = folly::DMPSCQueue<T, MayBlock, 8, 7, WeightFn>

Definition at line 36 of file DynamicBoundedQueueTest.cpp.

template<typename T , bool MayBlock, typename WeightFn >

using DSPMC = folly::DSPMCQueue<T, MayBlock, 8, 7, WeightFn>

Definition at line 39 of file DynamicBoundedQueueTest.cpp.

template<typename T , bool MayBlock, typename WeightFn >

using DSPSC = folly::DSPSCQueue<T, MayBlock, 8, 7, WeightFn>

Definition at line 33 of file DynamicBoundedQueueTest.cpp.

template<typename T , bool , typename >

using FMPMC = MPMC<T>

Definition at line 474 of file DynamicBoundedQueueTest.cpp.

template<typename T , bool , typename >

using FPCQ = PCQ<T>

Definition at line 515 of file DynamicBoundedQueueTest.cpp.

Function Documentation

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

void basic_test

(

)

Definition at line 45 of file DynamicBoundedQueueTest.cpp.

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

                  {
  auto dur = std::chrono::microseconds(100);
  auto deadline = std::chrono::steady_clock::now() + dur;

  struct CustomWeightFn {
    uint64_t operator()(int val) {
      return val * 100;
    }
  };

  Q<int, MayBlock, CustomWeightFn> q(10000);

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

  q.enqueue(1);
  ASSERT_TRUE(q.try_enqueue(2));
  ASSERT_TRUE(q.try_enqueue_until(3, deadline));
  ASSERT_TRUE(q.try_enqueue_for(4, dur));

  ASSERT_EQ(q.size(), 4);
  ASSERT_EQ(q.weight(), 1000);
  ASSERT_FALSE(q.empty());

  q.dequeue(v);
  ASSERT_EQ(v, 1);
  ASSERT_TRUE(q.try_dequeue(v));
  ASSERT_EQ(v, 2);
  ASSERT_TRUE(q.try_dequeue_until(v, deadline));
  ASSERT_EQ(v, 3);
  ASSERT_TRUE(q.try_dequeue_for(v, dur));
  ASSERT_EQ(v, 4);

  ASSERT_TRUE(q.empty());
  ASSERT_EQ(q.size(), 0);
  ASSERT_EQ(q.weight(), 0);
}

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

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

Definition at line 372 of file DynamicBoundedQueueTest.cpp.

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

                                                                        {
  int ops = FLAGS_ops;
  auto repFn = [&] {
    Q<T, Op == 3 || Op == 4 || Op == 5, folly::DefaultWeightFn<T>> q(
        FLAGS_capacity);
    std::atomic<uint64_t> sum(0);
    auto prod = [&](int tid) {
      for (int i = tid; i < ops; i += nprod) {
        if (Op == 0 || Op == 3) {
          while (!q.try_enqueue(i)) {
            /* keep trying */;
          }
        } else if (Op == 1 || Op == 4) {
          while (!q.try_enqueue_for(i, std::chrono::microseconds(1000))) {
            /* keep trying */;
          }
        } else {
          q.enqueue(i);
        }
      }
    };
    auto cons = [&](int tid) {
      uint64_t mysum = 0;
      T v = -1;
      for (int i = tid; i < ops; i += ncons) {
        if (Op == 0 || Op == 3) {
          while (!q.try_dequeue(v)) {
            /* keep trying */;
          }
        } else if (Op == 1 || Op == 4) {
          while (!q.try_dequeue_for(v, std::chrono::microseconds(1000))) {
            /* keep trying */;
          }
        } else {
          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 585 of file DynamicBoundedQueueTest.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> ===");
}

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

void capacity_test

(

)

Definition at line 142 of file DynamicBoundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_FALSE, ASSERT_TRUE, uint64_t, v, and val.

                     {
  struct CustomWeightFn {
    uint64_t operator()(int val) {
      return val;
    }
  };

  Q<int, MayBlock, CustomWeightFn> q(1000);
  ASSERT_EQ(q.weight(), 0);
  int v;
  q.enqueue(100);
  ASSERT_EQ(q.weight(), 100);
  q.enqueue(300);
  ASSERT_EQ(q.weight(), 400);
  ASSERT_FALSE(q.try_enqueue(1200));
  q.reset_capacity(2000); // reset capacityy to 2000
  ASSERT_TRUE(q.try_enqueue(1200));
  ASSERT_EQ(q.weight(), 1600);
  ASSERT_EQ(q.size(), 3);
  q.reset_capacity(1000); // reset capacity back to 1000
  ASSERT_FALSE(q.try_enqueue(100));
  q.dequeue(v);
  ASSERT_EQ(v, 100);
  ASSERT_EQ(q.weight(), 1500);
  q.dequeue(v);
  ASSERT_EQ(v, 300);
  ASSERT_EQ(q.weight(), 1200);
}

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	,
		1000000	,
		"capacity"
	)

void dottedLine

(

)

Definition at line 531 of file DynamicBoundedQueueTest.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 241 of file DynamicBoundedQueueTest.cpp.

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

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

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

  auto prod = [&](int tid) {
    for (int i = tid; i < ops; i += nprod) {
      if ((i % 3) == 0) {
        while (!q.try_enqueue(i)) {
          /* keep trying */;
        }
      } else if ((i % 3) == 1) {
        auto dur = std::chrono::microseconds(100);
        while (!q.try_enqueue_for(i, dur)) {
          /* keep trying */;
        }
      } else {
        q.enqueue(i);
      }
    }
  };

  auto cons = [&](int tid) {
    uint64_t mysum = 0;
    for (int i = tid; i < ops; i += ncons) {
      int v;
      if ((i % 3) == 0) {
        while (!q.try_dequeue(v)) {
          /* keep trying */;
        }
      } else if ((i % 3) == 1) {
        auto dur = std::chrono::microseconds(100);
        while (!q.try_dequeue_for(v, dur)) {
          /* keep trying */;
        }
      } else {
        q.dequeue(v);
      }
      if (nprod == 1 && ncons == 1) {
        ASSERT_EQ(v, i);
      }
      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, typename > class Q, bool MayBlock>

void move_test

(

)

Definition at line 97 of file DynamicBoundedQueueTest.cpp.

References ASSERT_EQ, ASSERT_TRUE, folly::gen::move, folly::pushmi::__adl::noexcept(), now(), uint64_t, and v.

                 {
  struct Foo {
    int v_;
    explicit Foo(int v) noexcept : v_(v) {}
    Foo(const Foo&) = delete;
    Foo& operator=(const Foo&) = delete;
    Foo(Foo&& other) noexcept : v_(other.v_) {}
    Foo& operator=(Foo&& other) noexcept {
      v_ = other.v_;
      return *this;
    }
  };

  struct CustomWeightFn {
    uint64_t operator()(Foo&&) {
      return 10;
    }
  };

  auto dur = std::chrono::microseconds(100);
  auto deadline = std::chrono::steady_clock::now() + dur;

  Q<Foo, MayBlock, CustomWeightFn> q(100);
  Foo v(1);
  q.enqueue(std::move(v));
  ASSERT_TRUE(q.try_enqueue(std::move(v)));
  ASSERT_TRUE(q.try_enqueue_until(std::move(v), deadline));
  ASSERT_TRUE(q.try_enqueue_for(std::move(v), dur));

  ASSERT_EQ(q.size(), 4);
  ASSERT_EQ(q.weight(), 40);
}

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 183 of file DynamicBoundedQueueTest.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 340 of file DynamicBoundedQueueTest.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 2 reps
    const uint64_t minute = 60000000000ULL;
    if (sum > minute && r >= 1) {
      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	(	DynamicBoundedQueue	,
		basic
	)

Definition at line 85 of file DynamicBoundedQueueTest.cpp.

                                 {
  basic_test<DSPSC, false>();
  basic_test<DMPSC, false>();
  basic_test<DSPMC, false>();
  basic_test<DMPMC, false>();
  basic_test<DSPSC, true>();
  basic_test<DMPSC, true>();
  basic_test<DSPMC, true>();
  basic_test<DMPMC, true>();
}

TEST	(	DynamicBoundedQueue	,
		move
	)

Definition at line 130 of file DynamicBoundedQueueTest.cpp.

                                {
  move_test<DSPSC, false>();
  move_test<DMPSC, false>();
  move_test<DSPMC, false>();
  move_test<DMPMC, false>();
  move_test<DSPSC, true>();
  move_test<DMPSC, true>();
  move_test<DSPMC, true>();
  move_test<DMPMC, true>();
}

TEST	(	DynamicBoundedQueue	,
		capacity
	)

Definition at line 171 of file DynamicBoundedQueueTest.cpp.

                                    {
  capacity_test<DSPSC, false>();
  capacity_test<DMPSC, false>();
  capacity_test<DSPMC, false>();
  capacity_test<DMPMC, false>();
  capacity_test<DSPSC, true>();
  capacity_test<DMPSC, true>();
  capacity_test<DSPMC, true>();
  capacity_test<DMPMC, true>();
}

TEST	(	DynamicBoundedQueue	,
		enq_deq
	)

Definition at line 304 of file DynamicBoundedQueueTest.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	(	DynamicBoundedQueue	,
		bench
	)

Definition at line 594 of file DynamicBoundedQueueTest.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 << "Using capacity " << FLAGS_capacity << " for all queues\n";
  std::cout << "=============================================================="
            << std::endl;
  std::cout << "Test name                         Max time  Avg time  Min time"
            << std::endl;

  for (int np : {1, 8, 32}) {
    for (int nc : {1, 8, 32}) {
      benches(np, nc);
    }
  }
}

template<typename T >

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

Definition at line 537 of file DynamicBoundedQueueTest.cpp.

References dottedLine().

                                                                     {
  std::cout << name
            << "===========================================" << std::endl;
  if (np == 1 && nc == 1) {
    bench<DSPSC, T, 0>(1, 1, "DSPSC try   spin only           ");
    bench<DSPSC, T, 1>(1, 1, "DSPSC timed spin only           ");
    bench<DSPSC, T, 2>(1, 1, "DSPSC wait  spin only           ");
    bench<DSPSC, T, 3>(1, 1, "DSPSC try   may block           ");
    bench<DSPSC, T, 4>(1, 1, "DSPSC timed may block           ");
    bench<DSPSC, T, 5>(1, 1, "DSPSC wait  may block           ");
    dottedLine();
  }
  if (nc == 1) {
    bench<DMPSC, T, 0>(np, 1, "DMPSC try   spin only           ");
    bench<DMPSC, T, 1>(np, 1, "DMPSC timed spin only           ");
    bench<DMPSC, T, 2>(np, 1, "DMPSC wait  spin only           ");
    bench<DMPSC, T, 3>(np, 1, "DMPSC try   may block           ");
    bench<DMPSC, T, 4>(np, 1, "DMPSC timed may block           ");
    bench<DMPSC, T, 5>(np, 1, "DMPSC wait  may block           ");
    dottedLine();
  }
  if (np == 1) {
    bench<DSPMC, T, 0>(1, nc, "DSPMC try   spin only           ");
    bench<DSPMC, T, 1>(1, nc, "DSPMC timed spin only           ");
    bench<DSPMC, T, 2>(1, nc, "DSPMC wait  spin only           ");
    bench<DSPMC, T, 3>(1, nc, "DSPMC try   may block           ");
    bench<DSPMC, T, 4>(1, nc, "DSPMC timed may block           ");
    bench<DSPMC, T, 5>(1, nc, "DSPMC wait  may block           ");
    dottedLine();
  }
  bench<DMPMC, T, 0>(np, nc, "DMPMC try   spin only           ");
  bench<DMPMC, T, 1>(np, nc, "DMPMC timed spin only           ");
  bench<DMPMC, T, 2>(np, nc, "DMPMC wait  spin only           ");
  bench<DMPMC, T, 3>(np, nc, "DMPMC try   may block           ");
  bench<DMPMC, T, 4>(np, nc, "DMPMC timed may block           ");
  bench<DMPMC, T, 5>(np, nc, "DMPMC 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;
}