RelaxedConcurrentPriorityQueueTest.cpp File Reference

#include <thread>
#include <boost/thread.hpp>
#include <folly/Random.h>
#include <folly/SpinLock.h>
#include <folly/experimental/FlatCombiningPriorityQueue.h>
#include <folly/experimental/RelaxedConcurrentPriorityQueue.h>
#include <folly/portability/GFlags.h>
#include <folly/portability/GTest.h>
#include <folly/test/DeterministicSchedule.h>
#include <glog/logging.h>

Go to the source code of this file.

Classes
class	Queue< E >
class	GlobalLockPQ< T >

Typedefs
using	DSched = folly::test::DeterministicSchedule
using	FCPQ = folly::FlatCombiningPriorityQueue< int >

Functions
	DEFINE_bool (bench, false,"run benchmark")
	DEFINE_int32 (reps, 1,"number of reps")
	DEFINE_int64 (ops, 32,"number of operations per rep")
	DEFINE_int64 (elems, 64,"number of elements")
template<class PriorityQueue >
void	basicOpsTest ()
	TEST (CPQ, BasicOpsTest)
template<typename Func >
static uint64_t	run_once (const Func &fn)
	execute the function for nthreads More...
template<class PriorityQueue >
void	singleThreadTest ()
	TEST (CPQ, SingleThrStrictImplTest)
	TEST (CPQ, SingleThrRelaxedImplTest)
template<class PriorityQueue >
void	concurrentPopforSharedBuffer ()
	TEST (CPQ, ConcurrentPopStrictImplTest)
	TEST (CPQ, ConcurrentPopRelaxedImplTest)
template<class PriorityQueue >
void	concurrentPush ()
	TEST (CPQ, ConcurrentPushStrictImplTest)
	TEST (CPQ, ConcurrentPushRelaxedImplTest)
template<class PriorityQueue >
void	concurrentOps (int ops)
template<class PriorityQueue >
void	concurrentSizeTest (int ops)
	TEST (CPQ, ConcurrentMixedStrictImplTest)
	TEST (CPQ, ConcurrentMixedRelaxedImplTest)
	TEST (CPQ, StrictImplSizeTest)
	TEST (CPQ, RelaxedImplSizeTest)
template<class PriorityQueue >
void	multiPusherPopper (int PushThr, int PopThr)
	TEST (CPQ, PusherPopperBlockingTest)
	TEST (CPQ, PusherPopperSpinningTest)
template<class PriorityQueue >
void	blockingFirst ()
template<class PriorityQueue >
void	concurrentBlocking ()
	TEST (CPQ, PopBlockingTest)
	TEST (CPQ, MixedBlockingTest)
template<class PriorityQueue , template< typename > class Atom = std::atomic>
static void	DSchedMixedTest ()
	TEST (CPQ, DSchedMixedStrictTest)
	TEST (CPQ, DSchedMixedRelaxedTest)
template<class PriorityQueue >
static uint64_t	producer_consumer_test (std::string name, uint32_t PushThr, uint32_t PopThr, uint64_t initial_size)
template<class PriorityQueue >
static uint64_t	throughtput_test (std::string name, uint64_t initial_size)
template<class PriorityQueue >
static void	accuracy_test (std::string name, uint64_t initial_size, uint32_t top_percent)
	TEST (CPQ, ThroughtputBench)
	TEST (CPQ, ProducerConsumerBench)
	TEST (CPQ, Accuracy)

Variables
static std::vector< int >	nthr = {1, 2, 4, 8}
static uint32_t	nthreads
static std::vector< int >	sizes = {0, 1024}

Typedef Documentation

using DSched = folly::test::DeterministicSchedule

Definition at line 625 of file RelaxedConcurrentPriorityQueueTest.cpp.

using FCPQ = folly::FlatCombiningPriorityQueue<int>

Definition at line 1052 of file RelaxedConcurrentPriorityQueueTest.cpp.

Function Documentation

template<class PriorityQueue >

static void accuracy_test ( std::string  name, uint64_t  initial_size, uint32_t  top_percent  )

static

Definition at line 1004 of file RelaxedConcurrentPriorityQueueTest.cpp.

References counter, folly::pushmi::operators::filter, i, folly::Random::rand32(), rng, stop(), uint64_t, and val.

                                                                           {
  int avg = 0;
  int reps = 15;
  int valid = initial_size / top_percent;
  if (valid < 1) {
    return;
  }
  int target = initial_size - valid;
  for (int r = 0; r < reps; ++r) {
    PriorityQueue pq;
    std::unordered_set<int> filter;
    folly::Random::DefaultGenerator rng;
    rng.seed(initial_size + r);

    // initialize the queue according to initial_size
    // eliminate repeated priorities
    for (uint64_t i = 0; i < initial_size; i++) {
      int val;
      do {
        val = folly::Random::rand32(rng) % initial_size;
      } while (filter.find(val) != filter.end());
      filter.insert(val);
      pq.push(val);
    }

    int counter = 0;
    int stop = valid;
    for (uint64_t i = 0; i < initial_size; i++) {
      int val;
      pq.pop(val);
      if (val >= target) {
        stop--;
      }
      if (stop > 0 && val < target) {
        counter++;
      }
      if (stop == 0) {
        break;
      }
    }
    avg += counter;
  }
  avg /= reps;
  std::cout << std::setw(16) << name << "  ";
  std::cout << "Lower priority popped: " << avg;
  std::cout << std::endl;
}

template<class PriorityQueue >

void basicOpsTest

(

)

Definition at line 40 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, EXPECT_FALSE, and EXPECT_TRUE.

                    {
  int res;
  PriorityQueue pq;

  EXPECT_TRUE(pq.empty());
  EXPECT_EQ(pq.size(), 0);
  pq.push(1);
  pq.push(2);
  pq.push(3);

  EXPECT_FALSE(pq.empty());
  EXPECT_EQ(pq.size(), 3);
  pq.pop(res);
  EXPECT_EQ(res, 3);
  pq.pop(res);
  EXPECT_EQ(res, 2);
  pq.pop(res);
  EXPECT_EQ(res, 1);
  EXPECT_TRUE(pq.empty());
  EXPECT_EQ(pq.size(), 0);

  pq.push(3);
  pq.push(2);
  pq.push(1);

  pq.pop(res);
  EXPECT_EQ(res, 3);
  pq.pop(res);
  EXPECT_EQ(res, 2);
  pq.pop(res);
  EXPECT_EQ(res, 1);
  EXPECT_TRUE(pq.empty());
  EXPECT_EQ(pq.size(), 0);
}

template<class PriorityQueue >

void blockingFirst

(

)

Definition at line 523 of file RelaxedConcurrentPriorityQueueTest.cpp.

References b, c, EXPECT_EQ, folly::pushmi::detail::t, and val.

                     {
  PriorityQueue pq;
  int nPop = 16;

  boost::barrier b{static_cast<unsigned int>(nPop + 1)};
  std::atomic<int> finished{0};

  std::vector<std::thread> threads_pop(nPop);
  for (int tid = 0; tid < nPop; ++tid) {
    threads_pop[tid] = std::thread([&] {
      int val;
      b.wait();
      pq.pop(val);
      finished.fetch_add(1, std::memory_order_acq_rel);
    });
  }

  b.wait();
  int c = 0;
  // push to Mound one by one
  // the popping threads should wake up one by one
  do {
    pq.push(1);
    c++;
    while (finished.load(std::memory_order_acquire) != c)
      ;
    EXPECT_EQ(finished.load(std::memory_order_acquire), c);
  } while (c < nPop);

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

template<class PriorityQueue >

void concurrentBlocking

(

)

Definition at line 558 of file RelaxedConcurrentPriorityQueueTest.cpp.

References b, EXPECT_EQ, folly::pushmi::detail::t, uint32_t, and val.

Referenced by TEST().

                          {
  uint32_t nThrs = 16;

  for (int iter = 0; iter < FLAGS_reps * 10; iter++) {
    PriorityQueue pq;
    boost::barrier b{static_cast<unsigned int>(nThrs + nThrs + 1)};
    std::atomic<uint32_t> finished{0};
    std::vector<std::thread> threads_pop(nThrs);
    for (uint32_t tid = 0; tid < nThrs; ++tid) {
      threads_pop[tid] = std::thread([&] {
        b.wait();
        int val;
        pq.pop(val);
        finished.fetch_add(1, std::memory_order_acq_rel);
      });
    }

    std::vector<std::thread> threads_push(nThrs);
    for (uint32_t tid = 0; tid < nThrs; ++tid) {
      threads_push[tid] = std::thread([&, tid] {
        b.wait();
        pq.push(tid);
        while (finished.load(std::memory_order_acquire) != nThrs)
          ;
        EXPECT_EQ(finished.load(std::memory_order_acquire), nThrs);
      });
    }

    b.wait();
    for (auto& t : threads_pop) {
      t.join();
    }
    for (auto& t : threads_push) {
      t.join();
    }
  }
}

template<class PriorityQueue >

void concurrentOps

(

int

ops

)

initialize the queue

operations

clean up work

Definition at line 291 of file RelaxedConcurrentPriorityQueueTest.cpp.

References i, nthreads, ops, folly::Random::rand32(), rng, run_once(), folly::pushmi::detail::t, uint32_t, uint64_t, and val.

                            {
  for (int t : nthr) {
    PriorityQueue pq;
    std::atomic<uint64_t> counter_push(0);
    std::atomic<uint64_t> counter_pop(0);
    nthreads = t;

    boost::barrier rb0{nthreads};
    boost::barrier rb1{nthreads};
    boost::barrier rb2{nthreads};
    std::atomic<int> to_end(0);

    auto fn = [&](uint32_t tid) {
      folly::Random::DefaultGenerator rng;
      rng.seed(tid);
      uint64_t local_push = 0;
      uint64_t local_pop = 0;
      int res;

      for (int i = tid; i < FLAGS_elems; i += nthreads) {
        int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
        local_push++;
        pq.push(val);
      }
      rb0.wait();

      for (int i = 0; i < ops; i++) {
        if (ops % 2 == 0) {
          int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
          local_push++;
          pq.push(val);
        } else {
          pq.pop(res);
          local_pop++;
        }
      }
      rb1.wait();
      // collecting the ops info for checking purpose
      counter_push.fetch_add(local_push, std::memory_order_seq_cst);
      counter_pop.fetch_add(local_pop, std::memory_order_seq_cst);
      rb2.wait();
      uint64_t r = counter_push.load(std::memory_order_seq_cst) -
          counter_pop.load(std::memory_order_seq_cst);
      while (true) {
        uint64_t index = to_end.fetch_add(1, std::memory_order_acq_rel);
        if (index < r) {
          pq.pop(res);
        } else {
          break;
        }
      }
    };
    run_once(fn);
    // the total push and pop ops should be the same
  }
}

template<class PriorityQueue >

void concurrentPopforSharedBuffer

(

)

concurrent pop should made the queue empty with executing the eaqual elements pop function

Definition at line 171 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, i, nthreads, folly::Random::rand32(), rng, run_once(), folly::pushmi::detail::t, uint32_t, uint64_t, and val.

Referenced by TEST().

                                    {
  for (int t : nthr) {
    PriorityQueue pq;

    folly::Random::DefaultGenerator rng;
    rng.seed(FLAGS_elems);
    uint64_t check_sum = 0;
    // random push
    for (int i = 0; i < FLAGS_elems; i++) {
      int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
      pq.push(val);
      check_sum += val;
    }

    std::atomic<uint64_t> pop_sum(0);
    std::atomic<int> to_end(0);

    nthreads = t;
    auto fn = [&](uint32_t tid) {
      int val = tid;
      while (true) {
        int index = to_end.fetch_add(1, std::memory_order_acq_rel);
        if (index < FLAGS_elems) {
          pq.pop(val);
        } else {
          break;
        }
        pop_sum.fetch_add(val, std::memory_order_acq_rel);
      }
    };
    run_once(fn);
    // check the sum of returned values of successful pop
    EXPECT_EQ(pop_sum, check_sum);
  }
}

template<class PriorityQueue >

void concurrentPush

(

)

executing fixed number of push, counting the element number & total value.

Definition at line 246 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, i, nthreads, folly::Random::rand32(), rng, run_once(), folly::pushmi::detail::t, uint32_t, uint64_t, and val.

                      {
  for (int t : nthr) {
    PriorityQueue pq;
    std::atomic<uint64_t> counter_sum(0);
    nthreads = t;

    auto fn = [&](uint32_t tid) {
      folly::Random::DefaultGenerator rng;
      rng.seed(tid);
      uint64_t local_sum = 0;
      for (int i = tid; i < FLAGS_elems; i += nthreads) {
        int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
        local_sum += val;
        pq.push(val);
      }
      counter_sum.fetch_add(local_sum, std::memory_order_acq_rel);
    };
    run_once(fn);

    // check the total number of elements
    uint64_t actual_sum = 0;
    for (int i = 0; i < FLAGS_elems; i++) {
      int res = 0;
      pq.pop(res);
      actual_sum += res;
    }
    EXPECT_EQ(actual_sum, counter_sum);
  }
}

template<class PriorityQueue >

void concurrentSizeTest

(

int

ops

)

initialize the queue

operations

Definition at line 352 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, EXPECT_TRUE, i, nthreads, ops, folly::Random::rand32(), rng, run_once(), folly::pushmi::detail::t, uint32_t, uint64_t, and val.

                                 {
  for (int t : nthr) {
    PriorityQueue pq;
    nthreads = t;
    EXPECT_TRUE(pq.empty());
    auto fn = [&](uint32_t tid) {
      folly::Random::DefaultGenerator rng;
      rng.seed(tid);
      uint64_t local_push = 0;
      int res;

      for (int i = tid; i < FLAGS_elems; i += nthreads) {
        int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
        local_push++;
        pq.push(val);
      }

      for (int i = 0; i < ops; i++) {
        int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
        pq.push(val);
        pq.pop(res);
      }
    };
    run_once(fn);
    // the total push and pop ops should be the same
    EXPECT_EQ(pq.size(), FLAGS_elems);
  }
}

DEFINE_bool	(	bench	,
		false	,
		"run benchmark"
	)

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

DEFINE_int64	(	ops	,
		32	,
		"number of operations per rep"
	)

DEFINE_int64	(	elems	,
		64	,
		"number of elements"
	)

template<class PriorityQueue , template< typename > class Atom = std::atomic>

static void DSchedMixedTest ( )

static

Definition at line 630 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, i, folly::test::DeterministicSchedule::join(), folly::Random::rand32(), rng, folly::pushmi::detail::t, folly::test::DeterministicSchedule::thread(), threads, uint64_t, and val.

Referenced by TEST().

                              {
  for (int r = 0; r < FLAGS_reps; r++) {
    // the thread number is  32
    int thr = 8;
    PriorityQueue pq;
    folly::Random::DefaultGenerator rng;
    rng.seed(thr);
    uint64_t pre_sum = 0;
    uint64_t pre_size = 0;
    for (int i = 0; i < FLAGS_elems; i++) {
      int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
      pq.push(val);
      pre_sum += val;
    }
    pre_size = FLAGS_elems;
    Atom<uint64_t> atom_push_sum{0};
    Atom<uint64_t> atom_pop_sum{0};
    std::vector<std::thread> threads(thr);
    for (int tid = 0; tid < thr; ++tid) {
      threads[tid] = DSched::thread([&]() {
        folly::Random::DefaultGenerator tl_rng;
        tl_rng.seed(thr);
        uint64_t pop_sum = 0;
        uint64_t push_sum = 0;
        for (int i = 0; i < FLAGS_ops; i++) {
          int val = folly::Random::rand32(tl_rng) % FLAGS_elems + 1;
          pq.push(val);
          push_sum += val;
          pq.pop(val);
          pop_sum += val;
        }
        atom_push_sum.fetch_add(push_sum, std::memory_order_acq_rel);
        atom_pop_sum.fetch_add(pop_sum, std::memory_order_acq_rel);
      });
    }

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

    // It checks the number of elements remain in Mound
    while (pre_size > 0) {
      pre_size--;
      int val = -1;
      pq.pop(val);
      atom_pop_sum += val;
    }
    // Check the accumulation of popped and pushed priorities
    EXPECT_EQ(atom_pop_sum, pre_sum + atom_push_sum);
  }
}

template<class PriorityQueue >

void multiPusherPopper	(	int	PushThr,
		int	PopThr
	)

Definition at line 428 of file RelaxedConcurrentPriorityQueueTest.cpp.

References EXPECT_EQ, i, ops, folly::Random::rand32(), folly::pushmi::detail::t, uint32_t, and val.

Referenced by TEST().

                                                {
  int ops = FLAGS_ops;
  uint32_t total_threads = PushThr + PopThr;

  PriorityQueue pq;
  std::atomic<uint64_t> sum_push(0);
  std::atomic<uint64_t> sum_pop(0);

  auto fn_popthr = [&](uint32_t tid) {
    for (int i = tid; i < ops; i += PopThr) {
      int val;
      pq.pop(val);
      sum_pop.fetch_add(val, std::memory_order_acq_rel);
    }
  };

  auto fn_pushthr = [&](uint32_t tid) {
    folly::Random::DefaultGenerator rng_t;
    rng_t.seed(tid);
    for (int i = tid; i < ops; i += PushThr) {
      int val = folly::Random::rand32(rng_t);
      pq.push(val);
      sum_push.fetch_add(val, std::memory_order_acq_rel);
    }
  };
  boost::barrier barrier_start{total_threads + 1};

  std::vector<std::thread> threads_push(PushThr);
  for (int tid = 0; tid < PushThr; ++tid) {
    threads_push[tid] = std::thread([&, tid] {
      barrier_start.wait();
      fn_pushthr(tid);
    });
  }
  std::vector<std::thread> threads_pop(PopThr);
  for (int tid = 0; tid < PopThr; ++tid) {
    threads_pop[tid] = std::thread([&, tid] {
      barrier_start.wait();
      fn_popthr(tid);
    });
  }

  barrier_start.wait(); // start the execution
  // begin time measurement
  for (auto& t : threads_push) {
    t.join();
  }
  for (auto& t : threads_pop) {
    t.join();
  }
  EXPECT_EQ(sum_pop, sum_push);
}

template<class PriorityQueue >

static uint64_t producer_consumer_test ( std::string  name, uint32_t  PushThr, uint32_t  PopThr, uint64_t  initial_size  )

static

Definition at line 868 of file RelaxedConcurrentPriorityQueueTest.cpp.

References count, i, max, min, name, now(), ops, folly::Random::rand32(), rng, sum(), folly::pushmi::detail::t, uint32_t, uint64_t, and val.

Referenced by TEST().

                           {
  int ops = 1 << 18;
  int reps = 15;
  if (name.find("RCPQ") != std::string::npos) {
    ops <<= 3;
  }
  uint64_t min = UINTMAX_MAX;
  uint64_t max = 0;
  uint64_t sum = 0;
  uint32_t total_threads = PushThr + PopThr;

  for (int r = 0; r < reps; ++r) {
    uint64_t dur;
    PriorityQueue pq;

    folly::Random::DefaultGenerator rng;
    rng.seed(initial_size);
    // initialize the queue according to initial_size
    for (uint64_t i = 0; i < initial_size; i++) {
      int val = folly::Random::rand32(rng) % ops;
      pq.push(val);
    }

    auto fn_popthr = [&](uint32_t tid) {
      for (int i = tid; i < ops; i += PopThr) {
        int val;
        pq.pop(val);
      }
    };

    auto fn_pushthr = [&](uint32_t tid) {
      folly::Random::DefaultGenerator rng_t;
      rng_t.seed(tid);
      for (int i = tid; i < ops; i += PushThr) {
        int val = folly::Random::rand32(rng_t) % ops;
        pq.push(val);
      }
    };
    boost::barrier barrier_start{total_threads + 1};
    std::vector<std::thread> threads_push(PushThr);
    for (uint32_t tid = 0; tid < PushThr; ++tid) {
      threads_push[tid] = std::thread([&, tid] {
        barrier_start.wait();
        fn_pushthr(tid);
      });
    }
    std::vector<std::thread> threads_pop(PopThr);
    for (uint32_t tid = 0; tid < PopThr; ++tid) {
      threads_pop[tid] = std::thread([&, tid] {
        barrier_start.wait();
        fn_popthr(tid);
      });
    }
    barrier_start.wait(); // start the execution
    // begin time measurement
    auto tbegin = std::chrono::steady_clock::now();
    for (auto& t : threads_push) {
      t.join();
    }
    for (auto& t : threads_pop) {
      t.join();
    }
    // end time measurement
    auto tend = std::chrono::steady_clock::now();
    dur = std::chrono::duration_cast<std::chrono::nanoseconds>(tend - tbegin)
              .count();
    sum += dur;
    min = std::min(min, dur);
    max = std::max(max, dur);
  }
  uint64_t avg = sum / reps;
  std::cout << std::setw(12) << name;
  std::cout << "   " << std::setw(8) << max / ops << " ns";
  std::cout << "   " << std::setw(8) << avg / ops << " ns";
  std::cout << "   " << std::setw(8) << min / ops << " ns";
  std::cout << std::endl;
  return min;
}

template<typename Func >

static uint64_t run_once ( const Func &  fn )

static

execute the function for nthreads

Definition at line 94 of file RelaxedConcurrentPriorityQueueTest.cpp.

References count, now(), nthreads, folly::pushmi::detail::t, threads, uint32_t, and uint64_t.

Referenced by concurrentOps(), concurrentPopforSharedBuffer(), concurrentPush(), concurrentSizeTest(), and throughtput_test().

                                         {
  boost::barrier barrier_start{nthreads + 1};
  std::vector<std::thread> threads(nthreads);
  for (uint32_t tid = 0; tid < nthreads; ++tid) {
    threads[tid] = std::thread([&, tid] {
      barrier_start.wait();
      fn(tid);
    });
  }

  barrier_start.wait(); // start the execution
  auto tbegin = std::chrono::steady_clock::now();
  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;
}

template<class PriorityQueue >

void singleThreadTest

(

)

Definition at line 119 of file RelaxedConcurrentPriorityQueueTest.cpp.

References counter, EXPECT_EQ, i, folly::Random::rand32(), rng, sum(), uint64_t, and val.

                        {
  PriorityQueue pq;

  folly::Random::DefaultGenerator rng;
  rng.seed(FLAGS_elems);
  uint64_t expect_sum = 0;
  // random push
  for (int i = 0; i < FLAGS_elems; i++) {
    int val = folly::Random::rand32(rng) % FLAGS_elems + 1;
    expect_sum += val;
    pq.push(val);
  }

  int val = 0;
  int counter = 0;
  uint64_t sum = 0;
  while (counter < FLAGS_elems) {
    pq.pop(val);
    sum += val;
    counter++;
  }

  EXPECT_EQ(sum, expect_sum);
}

TEST	(	CPQ	,
		BasicOpsTest
	)

Definition at line 75 of file RelaxedConcurrentPriorityQueueTest.cpp.

                        {
  // Spinning
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true>>();
  // Strict
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 0>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 0, 1>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 0, 3>>();
  // Relaxed
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 1, 1>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 2, 1>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, false, true, 3, 3>>();
  // Blocking
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, true, true>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, true, true, 0>>();
  basicOpsTest<RelaxedConcurrentPriorityQueue<int, true, true, 2>>();
}

TEST	(	CPQ	,
		SingleThrStrictImplTest
	)

Definition at line 144 of file RelaxedConcurrentPriorityQueueTest.cpp.

                                   {
  // spinning
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 0>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 0, 1>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 0, 8>>();
  // blocking
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, true, false, 0>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, true, false, 0, 1>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, true, false, 0, 8>>();
}

TEST	(	CPQ	,
		SingleThrRelaxedImplTest
	)

Definition at line 155 of file RelaxedConcurrentPriorityQueueTest.cpp.

                                    {
  // spinning
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 1>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 8>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 8, 2>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 8, 8>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 2, 128>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, false, false, 100, 8>>();
  // blocking
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, true, false, 1>>();
  singleThreadTest<RelaxedConcurrentPriorityQueue<int, true, false, 8>>();
}

TEST	(	CPQ	,
		ConcurrentPopStrictImplTest
	)

Definition at line 207 of file RelaxedConcurrentPriorityQueueTest.cpp.

References concurrentPopforSharedBuffer().

                                       {
  // spinning
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 0>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 0, 1>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 0, 128>>();
  // blocking
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 0>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 0, 1>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 0, 128>>();
}

TEST	(	CPQ	,
		ConcurrentPopRelaxedImplTest
	)

Definition at line 224 of file RelaxedConcurrentPriorityQueueTest.cpp.

References concurrentPopforSharedBuffer().

                                        {
  // spinning
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 1, 8>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 8, 2>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, false, false, 128, 2>>();
  // blocking
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 1>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 128>>();
  concurrentPopforSharedBuffer<
      RelaxedConcurrentPriorityQueue<int, true, false, 8, 128>>();
}

TEST	(	CPQ	,
		ConcurrentPushStrictImplTest
	)

Definition at line 276 of file RelaxedConcurrentPriorityQueueTest.cpp.

                                        {
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 0>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 0, 8>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 0, 128>>();
}

TEST	(	CPQ	,
		ConcurrentPushRelaxedImplTest
	)

Definition at line 282 of file RelaxedConcurrentPriorityQueueTest.cpp.

                                         {
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 1, 8>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 2, 128>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 128, 8>>();
  concurrentPush<RelaxedConcurrentPriorityQueue<int, false, false, 8, 8>>();
}

TEST	(	CPQ	,
		ConcurrentMixedStrictImplTest
	)

Definition at line 385 of file RelaxedConcurrentPriorityQueueTest.cpp.

References folly::size().

                                         {
  for (auto size : sizes) {
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 0>>(size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 0, 1>>(
        size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 0, 32>>(
        size);
  }
}

TEST	(	CPQ	,
		ConcurrentMixedRelaxedImplTest
	)

Definition at line 395 of file RelaxedConcurrentPriorityQueueTest.cpp.

References folly::size().

                                          {
  for (auto size : sizes) {
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false>>(size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 1, 32>>(
        size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 32, 1>>(
        size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, false, false, 8, 8>>(
        size);
    concurrentOps<RelaxedConcurrentPriorityQueue<int, true, false, 8, 8>>(size);
  }
}

TEST	(	CPQ	,
		StrictImplSizeTest
	)

Definition at line 408 of file RelaxedConcurrentPriorityQueueTest.cpp.

References folly::size().

                              {
  for (auto size : sizes) {
    concurrentSizeTest<RelaxedConcurrentPriorityQueue<int, false, true, 0>>(
        size);
    concurrentSizeTest<RelaxedConcurrentPriorityQueue<int, true, true, 0>>(
        size);
  }
}

TEST	(	CPQ	,
		RelaxedImplSizeTest
	)

Definition at line 417 of file RelaxedConcurrentPriorityQueueTest.cpp.

References folly::size().

                               {
  for (auto size : sizes) {
    concurrentSizeTest<RelaxedConcurrentPriorityQueue<int, false, true>>(size);
    concurrentSizeTest<RelaxedConcurrentPriorityQueue<int, true, true, 2, 8>>(
        size);
    concurrentSizeTest<RelaxedConcurrentPriorityQueue<int, false, true, 8, 2>>(
        size);
  }
}

TEST	(	CPQ	,
		PusherPopperBlockingTest
	)

Definition at line 481 of file RelaxedConcurrentPriorityQueueTest.cpp.

References i, and multiPusherPopper().

                                    {
  for (auto i : nthr) {
    for (auto j : nthr) {
      // Original
      multiPusherPopper<RelaxedConcurrentPriorityQueue<int, true, false, 0, 1>>(
          i, j);
      // Relaxed
      multiPusherPopper<RelaxedConcurrentPriorityQueue<int, true, false, 1, 8>>(
          i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, true, false, 8, 128>>(i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, true, false, 128, 8>>(i, j);
      multiPusherPopper<RelaxedConcurrentPriorityQueue<int, true, false>>(i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, true, false, 16, 16>>(i, j);
    }
  }
}

TEST	(	CPQ	,
		PusherPopperSpinningTest
	)

Definition at line 501 of file RelaxedConcurrentPriorityQueueTest.cpp.

References i, and multiPusherPopper().

                                    {
  for (auto i : nthr) {
    for (auto j : nthr) {
      // Original
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, false, false, 0, 1>>(i, j);
      // Relaxed
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, false, false, 1, 8>>(i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, false, false, 8, 128>>(i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, false, false, 128, 8>>(i, j);
      multiPusherPopper<RelaxedConcurrentPriorityQueue<int, false, false>>(
          i, j);
      multiPusherPopper<
          RelaxedConcurrentPriorityQueue<int, false, false, 16, 16>>(i, j);
    }
  }
}

TEST	(	CPQ	,
		PopBlockingTest
	)

Definition at line 596 of file RelaxedConcurrentPriorityQueueTest.cpp.

                           {
  // strict
  blockingFirst<RelaxedConcurrentPriorityQueue<int, true, false, 0, 1>>();
  blockingFirst<RelaxedConcurrentPriorityQueue<int, true, false, 0, 16>>();
  // relaxed
  blockingFirst<RelaxedConcurrentPriorityQueue<int, true, false>>();
  blockingFirst<RelaxedConcurrentPriorityQueue<int, true, false, 8, 8>>();
  blockingFirst<RelaxedConcurrentPriorityQueue<int, true, false, 16, 1>>();
  // Spinning
  blockingFirst<RelaxedConcurrentPriorityQueue<int, false, false>>();
  blockingFirst<RelaxedConcurrentPriorityQueue<int, false, false, 8, 8>>();
  blockingFirst<RelaxedConcurrentPriorityQueue<int, false, false, 16, 1>>();
}

TEST	(	CPQ	,
		MixedBlockingTest
	)

Definition at line 610 of file RelaxedConcurrentPriorityQueueTest.cpp.

References concurrentBlocking().

                             {
  // strict
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, true, false, 0, 1>>();
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, true, false, 0, 16>>();
  // relaxed
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, true, false>>();
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, true, false, 8, 8>>();
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, true, false, 16, 1>>();
  // Spinning
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, false, false>>();
  concurrentBlocking<RelaxedConcurrentPriorityQueue<int, false, false, 8, 8>>();
  concurrentBlocking<
      RelaxedConcurrentPriorityQueue<int, false, false, 16, 1>>();
}

TEST	(	CPQ	,
		DSchedMixedStrictTest
	)

Definition at line 682 of file RelaxedConcurrentPriorityQueueTest.cpp.

References DSchedMixedTest(), and folly::test::DeterministicSchedule::uniform().

                                 {
  DSched sched(DSched::uniform(0));
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          0,
          25,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          0,
          25,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          0,
          1,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          0,
          1,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          0,
          128,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          0,
          128,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
}

TEST	(	CPQ	,
		DSchedMixedRelaxedTest
	)

Definition at line 746 of file RelaxedConcurrentPriorityQueueTest.cpp.

References DSchedMixedTest(), and folly::test::DeterministicSchedule::uniform().

                                  {
  DSched sched(DSched::uniform(0));
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          16,
          25,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          16,
          25,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          1,
          16,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          1,
          16,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          16,
          1,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          16,
          1,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          false,
          false,
          16,
          16,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
  DSchedMixedTest<
      RelaxedConcurrentPriorityQueue<
          int,
          true,
          false,
          16,
          16,
          DeterministicMutex,
          DeterministicAtomic>,
      DeterministicAtomic>();
}

TEST	(	CPQ	,
		ThroughtputBench
	)

Definition at line 1054 of file RelaxedConcurrentPriorityQueueTest.cpp.

References i, nthreads, nthrs, and s.

                            {
  if (!FLAGS_bench) {
    return;
  }
  std::vector<int> test_sizes = {64, 512, 65536};
  std::vector<int> nthrs = {1, 2, 4, 8, 12, 14, 16, 28, 32, 56};

  std::cout << "Threads have equal chance to push and pop. \n"
            << "The bench caculates the avg execution time for\n"
            << "one operation (push OR pop).\n"
            << "GL : std::priority_queue protected by global lock\n"
            << "FL : flatcombinning priority queue\n"
            << "RCPQ: the relaxed concurrent priority queue\n"
            << std::endl;
  std::cout << "\nTest_name, Max time, Avg time, Min time" << std::endl;
  for (auto s : test_sizes) {
    std::cout << "\n ------ Initial size: " << s << " ------" << std::endl;
    for (int i : nthrs) {
      nthreads = i;
      std::cout << "Thread number: " << i << std::endl;
      throughtput_test<GlobalLockPQ<int>>("GL", s);
      throughtput_test<FCPQ>("FC", s);
      throughtput_test<RelaxedConcurrentPriorityQueue<int>>("RCPQ", s);
    }
  }
}

TEST	(	CPQ	,
		ProducerConsumerBench
	)

Definition at line 1081 of file RelaxedConcurrentPriorityQueueTest.cpp.

References m, nthrs, producer_consumer_test(), and s.

                                 {
  if (!FLAGS_bench) {
    return;
  }
  std::vector<int> test_sizes = {0, 512, 65536};
  std::vector<int> nthrs = {1, 2, 4, 8, 12, 16, 24};

  std::cout << "<Producer, Consumer> pattern \n"
            << "The bench caculates the avg execution time for\n"
            << "push AND pop pair(two operations).\n"
            << "GL : std::priority_queue protected by global lock\n"
            << "FL : flatcombinning priority queue\n"
            << "RCPQ SPN: RCPQ spinning\n"
            << "RCPQ BLK: RCPQ blocking\n"
            << std::endl;
  for (int s : test_sizes) {
    std::cout << "\n ------ Scalability ------" << std::endl;
    for (int m : nthrs) {
      for (int n : nthrs) {
        if (m != n) {
          continue;
        }
        std::cout << "<" << m << " , " << n << "> , size = " << s << ":"
                  << std::endl;
        producer_consumer_test<GlobalLockPQ<int>>("GL", m, n, s);
        producer_consumer_test<FCPQ>("FC", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, false, false>>(
            "RCPQ SPN", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, true, false>>(
            "RCPQ BLK", m, n, s);
      }
    }
    std::cout << "\n ------ Unbalanced(Producer<Consumer) ------" << std::endl;
    for (int m : nthrs) {
      for (int n : nthrs) {
        if (m > 4 || n - 4 <= m) {
          continue;
        }
        std::cout << "<" << m << " , " << n << "> , size = " << s << ":"
                  << std::endl;
        producer_consumer_test<GlobalLockPQ<int>>("GL", m, n, s);
        producer_consumer_test<FCPQ>("FC", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, false, false>>(
            "RCPQ SPN", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, true, false>>(
            "RCPQ BLK", m, n, s);
      }
    }

    std::cout << "\n ------ Unbalanced(Producer>Consumer) ------" << std::endl;
    for (int m : nthrs) {
      for (int n : nthrs) {
        if (m <= 8 || n > m - 4 || n % 4 != 0) {
          continue;
        }
        std::cout << "<" << m << " , " << n << "> , size = " << s << ":"
                  << std::endl;
        producer_consumer_test<GlobalLockPQ<int>>("GL", m, n, s);
        producer_consumer_test<FCPQ>("FC", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, false, false>>(
            "RCPQ SPN", m, n, s);
        producer_consumer_test<
            RelaxedConcurrentPriorityQueue<int, true, false>>(
            "RCPQ BLK", m, n, s);
      }
    }
  }
}

TEST	(	CPQ	,
		Accuracy
	)

Definition at line 1155 of file RelaxedConcurrentPriorityQueueTest.cpp.

References s.

                    {
  if (!FLAGS_bench) {
    return;
  }
  std::vector<int> test_sizes = {512, 65536, 1 << 20};
  std::vector<int> rates = {1000, 100, 10};
  for (auto s : test_sizes) {
    for (auto p : rates) {
      std::cout << "\n------ Size: " << s << "  Get top: " << 100. / p << "%"
                << " (Num: " << s / p << ")"
                << " ------" << std::endl;
      accuracy_test<Queue<int>>("FIFO Q", s, p);
      accuracy_test<RelaxedConcurrentPriorityQueue<int, false, false, 0>>(
          "RCPQ(strict)", s, p);
      accuracy_test<RelaxedConcurrentPriorityQueue<int, false, false, 2>>(
          "RCPQ(batch=2)", s, p);
      accuracy_test<RelaxedConcurrentPriorityQueue<int, false, false, 8>>(
          "RCPQ(batch=8)", s, p);
      accuracy_test<RelaxedConcurrentPriorityQueue<int, false, false, 16>>(
          "RCPQ(batch=16)", s, p);
      accuracy_test<RelaxedConcurrentPriorityQueue<int, false, false, 50>>(
          "RCPQ(batch=50)", s, p);
    }
  }
}

template<class PriorityQueue >

static uint64_t throughtput_test ( std::string  name, uint64_t  initial_size  )

static

Definition at line 952 of file RelaxedConcurrentPriorityQueueTest.cpp.

References counter, i, max, min, name, nthreads, ops, folly::Random::rand32(), rng, run_once(), sum(), uint32_t, uint64_t, and val.

                                                                        {
  int ops = 1 << 18;
  int reps = 15;
  uint64_t min = UINTMAX_MAX;
  uint64_t max = 0;
  uint64_t sum = 0;

  for (int r = 0; r < reps; ++r) {
    uint64_t dur;
    PriorityQueue pq;

    folly::Random::DefaultGenerator rng;
    rng.seed(initial_size);
    // initialize the queue according to initial_size
    for (uint64_t i = 0; i < initial_size; i++) {
      int val = folly::Random::rand32(rng) % (ops + 1);
      pq.push(val);
    }

    auto fn = [&](uint32_t tid) {
      folly::Random::DefaultGenerator rng_tl;
      rng_tl.seed(tid);
      uint32_t counter = 0;
      for (int i = tid; i < ops; i += nthreads) {
        int val;
        counter++;
        if (counter % 2) {
          val = folly::Random::rand32(rng_tl) % (ops + 1);
          pq.push(val);
        } else {
          pq.pop(val);
        }
      }
    };

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

  uint64_t avg = sum / reps;
  std::cout << std::setw(12) << name;
  std::cout << "   " << std::setw(8) << max / ops << " ns";
  std::cout << "   " << std::setw(8) << avg / ops << " ns";
  std::cout << "   " << std::setw(8) << min / ops << " ns";
  std::cout << std::endl;
  return min;
}

Variable Documentation

std::vector<int> nthr = {1, 2, 4, 8}

static

Definition at line 35 of file RelaxedConcurrentPriorityQueueTest.cpp.

uint32_t nthreads

static

Definition at line 37 of file RelaxedConcurrentPriorityQueueTest.cpp.

Referenced by concurrentOps(), concurrentPopforSharedBuffer(), concurrentPush(), concurrentSizeTest(), run_once(), TEST(), and throughtput_test().

std::vector<int> sizes = {0, 1024}

static

Definition at line 383 of file RelaxedConcurrentPriorityQueueTest.cpp.