Merge pull request #3282 from tchaikov/pq-test

common/PrioritizedQueue: add basic tests for it

Reviewed-by: Samuel Just <sjust@redhat.com>

src/common/PrioritizedQueue.h

@@ -49,13 +49,14 @@ class PrioritizedQueue {
   int64_t max_tokens_per_subqueue;
   int64_t min_cost;
 
+  typedef std::list<std::pair<unsigned, T> > ListPairs;
   template <class F>
   static unsigned filter_list_pairs(
-    list<pair<unsigned, T> > *l, F f,
-    list<T> *out) {
+    ListPairs *l, F f,
+    std::list<T> *out) {
     unsigned ret = 0;
     if (out) {
-      for (typename list<pair<unsigned, T> >::reverse_iterator i = l->rbegin();
+      for (typename ListPairs::reverse_iterator i = l->rbegin();
 	   i != l->rend();
 	   ++i) {
 	if (f(i->second)) {
@@ -63,7 +64,7 @@ class PrioritizedQueue {
 	}
       }
     }
-    for (typename list<pair<unsigned, T> >::iterator i = l->begin();
+    for (typename ListPairs::iterator i = l->begin();
 	 i != l->end();
       ) {
       if (f(i->second)) {
@@ -78,10 +79,11 @@ class PrioritizedQueue {
 
   struct SubQueue {
   private:
-    map<K, list<pair<unsigned, T> > > q;
+    typedef std::map<K, ListPairs> Classes;
+    Classes q;
     unsigned tokens, max_tokens;
     int64_t size;
-    typename map<K, list<pair<unsigned, T> > >::iterator cur;
+    typename Classes::iterator cur;
   public:
     SubQueue(const SubQueue &other)
       : q(other.q),
@@ -114,18 +116,18 @@ class PrioritizedQueue {
 	tokens = 0;
     }
     void enqueue(K cl, unsigned cost, T item) {
-      q[cl].push_back(make_pair(cost, item));
+      q[cl].push_back(std::make_pair(cost, item));
       if (cur == q.end())
 	cur = q.begin();
       size++;
     }
     void enqueue_front(K cl, unsigned cost, T item) {
-      q[cl].push_front(make_pair(cost, item));
+      q[cl].push_front(std::make_pair(cost, item));
       if (cur == q.end())
 	cur = q.begin();
       size++;
     }
-    pair<unsigned, T> front() const {
+    std::pair<unsigned, T> front() const {
       assert(!(q.empty()));
       assert(cur != q.end());
       return cur->second.front();
@@ -150,8 +152,8 @@ class PrioritizedQueue {
       return q.empty();
     }
     template <class F>
-    void remove_by_filter(F f, list<T> *out) {
-      for (typename map<K, list<pair<unsigned, T> > >::iterator i = q.begin();
+    void remove_by_filter(F f, std::list<T> *out) {
+      for (typename Classes::iterator i = q.begin();
 	   i != q.end();
 	   ) {
 	size -= filter_list_pairs(&(i->second), f, out);
@@ -166,15 +168,15 @@ class PrioritizedQueue {
       if (cur == q.end())
 	cur = q.begin();
     }
-    void remove_by_class(K k, list<T> *out) {
-      typename map<K, list<pair<unsigned, T> > >::iterator i = q.find(k);
+    void remove_by_class(K k, std::list<T> *out) {
+      typename Classes::iterator i = q.find(k);
       if (i == q.end())
 	return;
       size -= i->second.size();
       if (i == cur)
 	++cur;
       if (out) {
-	for (typename list<pair<unsigned, T> >::reverse_iterator j =
+	for (typename ListPairs::reverse_iterator j =
 	       i->second.rbegin();
 	     j != i->second.rend();
 	     ++j) {
@@ -195,11 +197,13 @@ class PrioritizedQueue {
 	f->dump_int("first_item_cost", front().first);
     }
   };
-  map<unsigned, SubQueue> high_queue;
-  map<unsigned, SubQueue> queue;
+
+  typedef std::map<unsigned, SubQueue> SubQueues;
+  SubQueues high_queue;
+  SubQueues queue;
 
   SubQueue *create_queue(unsigned priority) {
-    typename map<unsigned, SubQueue>::iterator p = queue.find(priority);
+    typename SubQueues::iterator p = queue.find(priority);
     if (p != queue.end())
       return &p->second;
     total_priority += priority;
@@ -218,7 +222,7 @@ class PrioritizedQueue {
   void distribute_tokens(unsigned cost) {
     if (total_priority == 0)
       return;
-    for (typename map<unsigned, SubQueue>::iterator i = queue.begin();
+    for (typename SubQueues::iterator i = queue.begin();
 	 i != queue.end();
 	 ++i) {
       i->second.put_tokens(((i->first * cost) / total_priority) + 1);
@@ -234,13 +238,13 @@ public:
 
   unsigned length() const {
     unsigned total = 0;
-    for (typename map<unsigned, SubQueue>::const_iterator i = queue.begin();
+    for (typename SubQueues::const_iterator i = queue.begin();
 	 i != queue.end();
 	 ++i) {
       assert(i->second.length());
       total += i->second.length();
     }
-    for (typename map<unsigned, SubQueue>::const_iterator i = high_queue.begin();
+    for (typename SubQueues::const_iterator i = high_queue.begin();
 	 i != high_queue.end();
 	 ++i) {
       assert(i->second.length());
@@ -250,8 +254,8 @@ public:
   }
 
   template <class F>
-  void remove_by_filter(F f, list<T> *removed = 0) {
-    for (typename map<unsigned, SubQueue>::iterator i = queue.begin();
+  void remove_by_filter(F f, std::list<T> *removed = 0) {
+    for (typename SubQueues::iterator i = queue.begin();
 	 i != queue.end();
 	 ) {
       unsigned priority = i->first;
@@ -264,7 +268,7 @@ public:
 	++i;
       }
     }
-    for (typename map<unsigned, SubQueue>::iterator i = high_queue.begin();
+    for (typename SubQueues::iterator i = high_queue.begin();
 	 i != high_queue.end();
 	 ) {
       i->second.remove_by_filter(f, removed);
@@ -276,8 +280,8 @@ public:
     }
   }
 
-  void remove_by_class(K k, list<T> *out = 0) {
-    for (typename map<unsigned, SubQueue>::iterator i = queue.begin();
+  void remove_by_class(K k, std::list<T> *out = 0) {
+    for (typename SubQueues::iterator i = queue.begin();
 	 i != queue.end();
 	 ) {
       i->second.remove_by_class(k, out);
@@ -289,7 +293,7 @@ public:
 	++i;
       }
     }
-    for (typename map<unsigned, SubQueue>::iterator i = high_queue.begin();
+    for (typename SubQueues::iterator i = high_queue.begin();
 	 i != high_queue.end();
 	 ) {
       i->second.remove_by_class(k, out);
@@ -345,7 +349,7 @@ public:
     // if there are multiple buckets/subqueues with sufficient tokens,
     // we behave like a strict priority queue among all subqueues that
     // are eligible to run.
-    for (typename map<unsigned, SubQueue>::iterator i = queue.begin();
+    for (typename SubQueues::iterator i = queue.begin();
 	 i != queue.end();
 	 ++i) {
       assert(!(i->second.empty()));
@@ -377,7 +381,7 @@ public:
     f->dump_int("max_tokens_per_subqueue", max_tokens_per_subqueue);
     f->dump_int("min_cost", min_cost);
     f->open_array_section("high_queues");
-    for (typename map<unsigned, SubQueue>::const_iterator p = high_queue.begin();
+    for (typename SubQueues::const_iterator p = high_queue.begin();
 	 p != high_queue.end();
 	 ++p) {
       f->open_object_section("subqueue");
@@ -387,7 +391,7 @@ public:
     }
     f->close_section();
     f->open_array_section("queues");
-    for (typename map<unsigned, SubQueue>::const_iterator p = queue.begin();
+    for (typename SubQueues::const_iterator p = queue.begin();
 	 p != queue.end();
 	 ++p) {
       f->open_object_section("subqueue");

src/test/Makefile.am

@@ -151,6 +151,12 @@ unittest_histogram_CXXFLAGS = $(UNITTEST_CXXFLAGS)
 unittest_histogram_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL)
 check_PROGRAMS += unittest_histogram
 
+unittest_prioritized_queue_SOURCES = test/common/test_prioritized_queue.cc
+unittest_prioritized_queue_CXXFLAGS = $(UNITTEST_CXXFLAGS)
+unittest_prioritized_queue_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL)
+check_PROGRAMS += unittest_prioritized_queue
+
+
 unittest_str_map_SOURCES = test/common/test_str_map.cc
 unittest_str_map_CXXFLAGS = $(UNITTEST_CXXFLAGS)
 unittest_str_map_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL)

src/test/common/test_prioritized_queue.cc

@@ -0,0 +1,248 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "gtest/gtest.h"
+#include "common/PrioritizedQueue.h"
+
+#include <numeric>
+#include <vector>
+
+
+using std::vector;
+
+class PrioritizedQueueTest : public testing::Test
+{
+protected:
+  typedef int Klass;
+  typedef unsigned Item;
+  typedef PrioritizedQueue<Item, Klass> PQ;
+  enum { item_size  = 100, };
+  vector<Item> items;
+
+  virtual void SetUp() {
+    for (int i = 0; i < item_size; i++) {
+      items.push_back(Item(i));
+    }
+    random_shuffle(items.begin(), items.end());
+  }
+  virtual void TearDown() {
+    items.clear();
+  }
+};
+
+TEST_F(PrioritizedQueueTest, capacity) {
+  const unsigned min_cost  = 10;
+  const unsigned max_tokens_per_subqueue = 50;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+  EXPECT_TRUE(pq.empty());
+  EXPECT_EQ(0u, pq.length());
+
+  pq.enqueue_strict(Klass(1), 0, Item(0));
+  EXPECT_FALSE(pq.empty());
+  EXPECT_EQ(1u, pq.length());
+
+  for (int i = 0; i < 3; i++) {
+    pq.enqueue(Klass(1), 0, 10, Item(0));
+  }
+  for (unsigned i = 4; i > 0; i--) {
+    EXPECT_FALSE(pq.empty());
+    EXPECT_EQ(i, pq.length());
+    pq.dequeue();
+  }
+  EXPECT_TRUE(pq.empty());
+  EXPECT_EQ(0u, pq.length());
+}
+
+TEST_F(PrioritizedQueueTest, strict_pq) {
+  const unsigned min_cost = 1;
+  const unsigned max_tokens_per_subqueue = 50;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+  // 0 .. item_size-1
+  for (unsigned i = 0; i < item_size; i++) {
+    unsigned priority = items[i];
+    pq.enqueue_strict(Klass(0), priority, items[i]);
+  }
+  // item_size-1 .. 0
+  for (unsigned i = item_size; i > 0; i--) {
+    Item item = pq.dequeue();
+    unsigned priority = item;
+    EXPECT_EQ(i - 1, priority);
+  }
+}
+
+TEST_F(PrioritizedQueueTest, lowest_among_eligible_otherwise_highest) {
+  // to minimize the effect of `distribute_tokens()`
+  // all eligible items will be assigned with cost of min_cost
+  const unsigned min_cost = 0;
+  const unsigned max_tokens_per_subqueue = 100;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+
+#define ITEM_TO_COST(item_) (item_ % 5 ? min_cost : max_tokens_per_subqueue)
+  unsigned num_low_cost = 0, num_high_cost = 0;
+  for (int i = 0; i < item_size; i++) {
+    const Item& item = items[i];
+    unsigned cost = ITEM_TO_COST(item);
+    unsigned priority = item;
+    if (cost == min_cost) {
+      num_low_cost++;
+    } else {
+      num_high_cost++;
+    }
+    pq.enqueue(Klass(0), priority, cost, item);
+  }
+  // the token in all buckets is 0 at the beginning, so dequeue() should pick
+  // the first one with the highest priority.
+  unsigned highest_priority;
+  {
+    Item item = pq.dequeue();
+    unsigned cost = ITEM_TO_COST(item);
+    unsigned priority = item;
+    if (cost == min_cost) {
+      num_low_cost--;
+    } else {
+      num_high_cost--;
+    }
+    EXPECT_EQ(item_size - 1u, priority);
+    highest_priority = priority;
+  }
+  unsigned lowest_priority = 0;
+  for (unsigned i = 0; i < num_low_cost; i++) {
+    Item item = pq.dequeue();
+    unsigned cost = ITEM_TO_COST(item);
+    unsigned priority = item;
+    EXPECT_EQ(min_cost, cost);
+    EXPECT_GT(priority, lowest_priority);
+    lowest_priority = priority;
+  }
+  for (unsigned i = 0; i < num_high_cost; i++) {
+    Item item = pq.dequeue();
+    unsigned cost = ITEM_TO_COST(item);
+    unsigned priority = item;
+    EXPECT_EQ(max_tokens_per_subqueue, cost);
+    EXPECT_LT(priority, highest_priority);
+    highest_priority = priority;
+  }
+#undef ITEM_TO_COST
+}
+
+static const unsigned num_classes = 4;
+// just a determinitic number
+#define ITEM_TO_CLASS(item_) Klass((item_ + 43) % num_classes)
+
+TEST_F(PrioritizedQueueTest, fairness_by_class) {
+  // dequeue should be fair to all classes in a certain bucket
+  const unsigned min_cost = 1;
+  const unsigned max_tokens_per_subqueue = 50;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+
+  for (int i = 0; i < item_size; i++) {
+    const Item& item = items[i];
+    Klass k = ITEM_TO_CLASS(item);
+    unsigned priority = 0;
+    unsigned cost = 1;
+    pq.enqueue(k, priority, cost, item);
+  }
+  // just sample first 1/2 of the items
+  // if i pick too small a dataset, the result won't be statisitcally
+  // significant. if the sample dataset is too large, it will converge to the
+  // distribution of the full set.
+  vector<unsigned> num_picked_in_class(num_classes, 0u);
+  for (int i = 0; i < item_size / 2; i++) {
+    Item item = pq.dequeue();
+    Klass k = ITEM_TO_CLASS(item);
+    num_picked_in_class[k]++;
+  }
+  unsigned total = std::accumulate(num_picked_in_class.begin(),
+				   num_picked_in_class.end(),
+				   0);
+  float avg = float(total) / num_classes;
+  for (unsigned i = 0; i < num_classes; i++) {
+    EXPECT_NEAR(avg, num_picked_in_class[i], 0.5);
+  }
+}
+
+template <typename T>
+struct Greater {
+  const T rhs;
+  Greater(const T& v) : rhs(v)
+  {}
+  bool operator()(const T& lhs) const {
+    return lhs > rhs;
+  }
+};
+
+TEST_F(PrioritizedQueueTest, remove_by_filter) {
+  const unsigned min_cost = 1;
+  const unsigned max_tokens_per_subqueue = 50;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+
+  const Greater<Item> pred(item_size/2);
+  unsigned num_to_remove = 0;
+  for (unsigned i = 0; i < item_size; i++) {
+    const Item& item = items[i];
+    pq.enqueue(Klass(1), 0, 10, item);
+    if (pred(item)) {
+      num_to_remove++;
+    }
+  }
+  std::list<Item> removed;
+  pq.remove_by_filter(pred, &removed);
+
+  // see if the removed items are expected ones.
+  for (std::list<Item>::iterator it = removed.begin();
+       it != removed.end();
+       ++it) {
+    const Item& item = *it;
+    EXPECT_TRUE(pred(item));
+    items.erase(remove(items.begin(), items.end(), item), items.end());
+  }
+  EXPECT_EQ(num_to_remove, removed.size());
+  EXPECT_EQ(item_size - num_to_remove, pq.length());
+  EXPECT_EQ(item_size - num_to_remove, items.size());
+  // see if the remainder are expeceted also.
+  while (!pq.empty()) {
+    const Item item = pq.dequeue();
+    EXPECT_FALSE(pred(item));
+    items.erase(remove(items.begin(), items.end(), item), items.end());
+  }
+  EXPECT_TRUE(items.empty());
+}
+
+TEST_F(PrioritizedQueueTest, remove_by_class) {
+  const unsigned min_cost = 1;
+  const unsigned max_tokens_per_subqueue = 50;
+  PQ pq(max_tokens_per_subqueue, min_cost);
+  const Klass class_to_remove(2);
+  unsigned num_to_remove = 0;
+  for (int i = 0; i < item_size; i++) {
+    const Item& item = items[i];
+    Klass k = ITEM_TO_CLASS(item);
+    pq.enqueue(k, 0, 0, item);
+    if (k == class_to_remove) {
+      num_to_remove++;
+    }
+  }
+  std::list<Item> removed;
+  pq.remove_by_class(class_to_remove, &removed);
+
+  // see if the removed items are expected ones.
+  for (std::list<Item>::iterator it = removed.begin();
+       it != removed.end();
+       ++it) {
+    const Item& item = *it;
+    Klass k = ITEM_TO_CLASS(item);
+    EXPECT_EQ(class_to_remove, k);
+    items.erase(remove(items.begin(), items.end(), item), items.end());
+  }
+  EXPECT_EQ(num_to_remove, removed.size());
+  EXPECT_EQ(item_size - num_to_remove, pq.length());
+  EXPECT_EQ(item_size - num_to_remove, items.size());
+  // see if the remainder are expeceted also.
+  while (!pq.empty()) {
+    const Item item = pq.dequeue();
+    Klass k = ITEM_TO_CLASS(item);
+    EXPECT_NE(class_to_remove, k);
+    items.erase(remove(items.begin(), items.end(), item), items.end());
+  }
+  EXPECT_TRUE(items.empty());
+}