FlatCombiningPriorityQueueTest.cpp File Reference

#include <folly/experimental/FlatCombiningPriorityQueue.h>
#include <folly/Benchmark.h>
#include <folly/portability/GTest.h>
#include <glog/logging.h>
#include <condition_variable>
#include <mutex>
#include <queue>

Go to the source code of this file.

Classes
class	BaselinePQ< T, PriorityQueue, Mutex >

Typedefs
using	FCPQ = folly::FlatCombiningPriorityQueue< int >
using	Baseline = BaselinePQ< int >

Enumerations
enum	Exp { NoFC, FCNonBlock, FCBlock, FCTimed }

Functions
	DEFINE_bool (bench, false,"run benchmark")
	DEFINE_int32 (reps, 10,"number of reps")
	DEFINE_int32 (ops, 100000,"number of operations per rep")
	DEFINE_int32 (size, 64,"initial size of the priority queue")
	DEFINE_int32 (work, 1000,"amount of unrelated work per operation")
void	doWork (int work)
template<typename PriorityQueue , typename Func >
static uint64_t	run_once (PriorityQueue &pq, const Func &fn)
	TEST (FCPriQueue, basic)
	TEST (FCPriQueue, bounded)
	TEST (FCPriQueue, timeout)
	TEST (FCPriQueue, push_pop)
static uint64_t	test (std::string name, Exp exp, uint64_t base)
	TEST (FCPriQueue, bench)

Variables
static std::vector< int >	nthr = {1, 2, 3, 4, 6, 8, 12, 16, 24, 32, 48, 64}
static uint32_t	nthreads

Typedef Documentation

using Baseline = BaselinePQ<int>

Definition at line 108 of file FlatCombiningPriorityQueueTest.cpp.

using FCPQ = folly::FlatCombiningPriorityQueue<int>

Definition at line 107 of file FlatCombiningPriorityQueueTest.cpp.

Enumeration Type Documentation

enum Exp

Enumerator
NoFC
FCNonBlock
FCBlock
FCTimed

Definition at line 272 of file FlatCombiningPriorityQueueTest.cpp.

         {
  NoFC,
  FCNonBlock,
  FCBlock,
  FCTimed,
};

Function Documentation

DEFINE_bool	(	bench	,
		false	,
		"run benchmark"
	)

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

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

DEFINE_int32	(	size	,
		64	,
		"initial size of the priority queue"
	)

DEFINE_int32	(	work	,
		1000	,
		"amount of unrelated work per operation"
	)

void doWork

(

int

work

)

Definition at line 32 of file FlatCombiningPriorityQueueTest.cpp.

References a, folly::doNotOptimizeAway(), i, mutex, Mutex, T, and uint64_t.

Referenced by TEST(), and test().

                      {
  uint64_t a = 0;
  for (int i = work; i > 0; --i) {
    a += i;
  }
  folly::doNotOptimizeAway(a);
}

template<typename PriorityQueue , typename Func >

static uint64_t run_once ( PriorityQueue &  pq, const Func &  fn  )

static

Definition at line 118 of file FlatCombiningPriorityQueueTest.cpp.

References count, i, now(), nthreads, ops, BaselinePQ< T, PriorityQueue, Mutex >::size(), start, folly::pushmi::detail::t, threads, uint32_t, and uint64_t.

Referenced by TEST(), and test().

                                                            {
  int ops = FLAGS_ops;
  int size = FLAGS_size;
  std::atomic<bool> start{false};
  std::atomic<uint32_t> started{0};

  for (int i = 0; i < size; ++i) {
    CHECK(pq.try_push(i * (ops / size)));
  }

  std::vector<std::thread> threads(nthreads);
  for (uint32_t tid = 0; tid < nthreads; ++tid) {
    threads[tid] = std::thread([&, tid] {
      started.fetch_add(1);
      while (!start.load()) {
        /* nothing */;
      }
      fn(tid);
    });
  }

  while (started.load() < nthreads) {
    /* nothing */;
  }
  auto tbegin = std::chrono::steady_clock::now();

  // begin time measurement
  start.store(true);

  for (auto& t : threads) {
    t.join();
  }

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

TEST	(	FCPriQueue	,
		basic
	)

Definition at line 159 of file FlatCombiningPriorityQueueTest.cpp.

References EXPECT_EQ, EXPECT_FALSE, and v.

                        {
  FCPQ pq;
  CHECK(pq.empty());
  CHECK_EQ(pq.size(), 0);
  int v;
  CHECK(!pq.try_pop(v));
  // try_pop() returns an Optional
  EXPECT_FALSE(bool(pq.try_pop()));

  CHECK(pq.try_push(1));
  CHECK(pq.try_push(2));
  CHECK(!pq.empty());
  CHECK_EQ(pq.size(), 2);

  pq.peek(v);
  CHECK_EQ(v, 2); // higher value has higher priority
  CHECK(pq.try_peek(v));
  CHECK_EQ(v, 2);
  CHECK(!pq.empty());
  CHECK_EQ(pq.size(), 2);

  CHECK(pq.try_pop(v));
  CHECK_EQ(v, 2);
  CHECK(!pq.empty());
  CHECK_EQ(pq.size(), 1);

  CHECK(pq.try_pop(v));
  CHECK_EQ(v, 1);
  CHECK(pq.empty());
  CHECK_EQ(pq.size(), 0);

  CHECK(pq.try_push(1));
  CHECK(pq.try_push(2));

  // check successful try_pop()
  EXPECT_EQ(*pq.try_pop(), 2);
  CHECK(!pq.empty());
  CHECK_EQ(pq.size(), 1);

  EXPECT_EQ(*pq.try_pop(), 1);
  CHECK(pq.empty());
  CHECK_EQ(pq.size(), 0);
}

TEST	(	FCPriQueue	,
		bounded
	)

Definition at line 203 of file FlatCombiningPriorityQueueTest.cpp.

References v.

                          {
  FCPQ pq(1);
  CHECK(pq.try_push(1));
  CHECK(!pq.try_push(1));
  CHECK_EQ(pq.size(), 1);
  CHECK(!pq.empty());
  int v;
  CHECK(pq.try_pop(v));
  CHECK_EQ(v, 1);
  CHECK_EQ(pq.size(), 0);
  CHECK(pq.empty());
}

TEST	(	FCPriQueue	,
		timeout
	)

Definition at line 216 of file FlatCombiningPriorityQueueTest.cpp.

References EXPECT_EQ, EXPECT_FALSE, now(), and v.

                          {
  FCPQ pq(1);
  int v;
  CHECK(!pq.try_peek(v));
  CHECK(!pq.try_pop(v));
  pq.push(10);
  CHECK(!pq.try_push(20));

  auto dur = std::chrono::microseconds(1000);
  EXPECT_EQ(*pq.try_pop(), 10);
  CHECK(pq.empty());
  // check try_***_for
  EXPECT_FALSE(bool(pq.try_pop_for(dur)));
  EXPECT_FALSE(bool(pq.try_peek_for(dur)));
  CHECK(pq.try_push_for(10, dur));
  CHECK(!pq.try_push_for(20, dur));
  EXPECT_EQ(*pq.try_peek_for(dur), 10);
  EXPECT_EQ(*pq.try_pop_for(dur), 10);

  CHECK(pq.empty());
  // check try_***_until
  EXPECT_FALSE(bool(pq.try_pop_until(std::chrono::steady_clock::now() + dur)));

  EXPECT_FALSE(bool(pq.try_peek_until(std::chrono::steady_clock::now() + dur)));
  CHECK(pq.try_push_until(10, std::chrono::steady_clock::now() + dur));
  CHECK(!pq.try_push_until(20, std::chrono::steady_clock::now() + dur));
  EXPECT_EQ(*pq.try_peek_until(std::chrono::steady_clock::now() + dur), 10);
  EXPECT_EQ(*pq.try_pop_until(std::chrono::steady_clock::now() + dur), 10);
  CHECK(pq.empty());
}

TEST	(	FCPriQueue	,
		push_pop
	)

Definition at line 247 of file FlatCombiningPriorityQueueTest.cpp.

References doWork(), EXPECT_NE, i, folly::none, now(), nthreads, ops, run_once(), uint32_t, v, and folly::when().

                           {
  int ops = 1000;
  int work = 0;
  std::chrono::steady_clock::time_point when =
      std::chrono::steady_clock::now() + std::chrono::hours(24);
  for (auto n : nthr) {
    nthreads = n;
    FCPQ pq(10000);
    auto fn = [&](uint32_t tid) {
      for (int i = tid; i < ops; i += nthreads) {
        CHECK(pq.try_push(i));
        CHECK(pq.try_push_until(i, when));
        pq.push(i);
        doWork(work);
        int v;
        CHECK(pq.try_pop(v));
        EXPECT_NE(pq.try_pop_until(when), folly::none);
        pq.pop(v);
        doWork(work);
      }
    };
    run_once(pq, fn);
  }
}

static uint64_t test ( std::string  name, Exp  exp, uint64_t  base  )

static

Definition at line 279 of file FlatCombiningPriorityQueueTest.cpp.

References doWork(), EXPECT_NE, FCBlock, FCNonBlock, FCTimed, i, max, min, name, NoFC, folly::none, now(), nthreads, ops, run_once(), sum(), BaselinePQ< T, PriorityQueue, Mutex >::try_pop(), BaselinePQ< T, PriorityQueue, Mutex >::try_push(), uint32_t, uint64_t, v, and folly::when().

Referenced by TEST().

                                                             {
  int ops = FLAGS_ops;
  int work = FLAGS_work;

  uint64_t min = UINTMAX_MAX;
  uint64_t max = 0;
  uint64_t sum = 0;

  for (int r = 0; r < FLAGS_reps; ++r) {
    uint64_t dur;
    switch (exp) {
      case NoFC: {
        Baseline pq;
        auto fn = [&](uint32_t tid) {
          for (int i = tid; i < ops; i += nthreads) {
            CHECK(pq.try_push(i));
            doWork(work);
            int v;
            CHECK(pq.try_pop(v));
            doWork(work);
          }
        };
        dur = run_once(pq, fn);
        break;
      }
      case FCNonBlock: {
        FCPQ pq;
        auto fn = [&](uint32_t tid) {
          for (int i = tid; i < ops; i += nthreads) {
            CHECK(pq.try_push(i));
            doWork(work);
            int v;
            CHECK(pq.try_pop(v));
            doWork(work);
          }
        };
        dur = run_once(pq, fn);
        break;
      }
      case FCBlock: {
        FCPQ pq;
        auto fn = [&](uint32_t tid) {
          for (int i = tid; i < ops; i += nthreads) {
            pq.push(i);
            doWork(work);
            int v;
            pq.pop(v);
            doWork(work);
          }
        };
        dur = run_once(pq, fn);
        break;
      }
      case FCTimed: {
        FCPQ pq;
        auto fn = [&](uint32_t tid) {
          std::chrono::steady_clock::time_point when =
              std::chrono::steady_clock::now() + std::chrono::hours(24);
          for (int i = tid; i < ops; i += nthreads) {
            CHECK(pq.try_push_until(i, when));
            doWork(work);
            EXPECT_NE(pq.try_pop_until(when), folly::none);
            doWork(work);
          }
        };
        dur = run_once(pq, fn);
        break;
      }
      default:
        CHECK(false);
    }

    sum += dur;
    min = std::min(min, dur);
    max = std::max(max, dur);
  }

  uint64_t avg = sum / FLAGS_reps;
  uint64_t res = min;
  std::cout << name;
  std::cout << "   " << std::setw(4) << max / FLAGS_ops << " ns";
  std::cout << "   " << std::setw(4) << avg / FLAGS_ops << " ns";
  std::cout << "   " << std::setw(4) << res / FLAGS_ops << " ns";
  if (base) {
    std::cout << " " << std::setw(3) << 100 * base / res << "%";
  }
  std::cout << std::endl;
  return res;
}

TEST	(	FCPriQueue	,
		bench
	)

Definition at line 369 of file FlatCombiningPriorityQueueTest.cpp.

References FCBlock, FCNonBlock, FCTimed, i, NoFC, nthreads, test(), and uint64_t.

                        {
  if (!FLAGS_bench) {
    return;
  }

  std::cout << "Test_name, Max time, Avg time, Min time, % base min / min"
            << std::endl;
  for (int i : nthr) {
    nthreads = i;
    std::cout << "\n------------------------------------ Number of threads = "
              << i << std::endl;
    uint64_t base = test("baseline                    ", NoFC, 0);
    test("baseline - dup              ", NoFC, base);
    std::cout << "---- fc -------------------------------" << std::endl;
    test("fc non-blocking             ", FCNonBlock, base);
    test("fc non-blocking - dup       ", FCNonBlock, base);
    test("fc timed                    ", FCTimed, base);
    test("fc timed - dup              ", FCTimed, base);
    test("fc blocking                 ", FCBlock, base);
    test("fc blocking - dup           ", FCBlock, base);
  }
}

Variable Documentation

std::vector<int> nthr = {1, 2, 3, 4, 6, 8, 12, 16, 24, 32, 48, 64}

static

Definition at line 113 of file FlatCombiningPriorityQueueTest.cpp.

uint32_t nthreads

static

Definition at line 115 of file FlatCombiningPriorityQueueTest.cpp.

Referenced by run_once(), TEST(), and test().