Extended Utility Codel (#475)

Author: TheJStone

core/utils/codel.h

@@ -3,13 +3,13 @@
 
 #include <cmath>
 #include <deque>
+#include <functional>
 
 #include <glog/logging.h>
 
 #include "time.h"
 #include "queue.h"
 
-
 namespace bess {
 namespace utils {
 // Codel(Controlled Delay Management) is an Queue controller based on this
@@ -50,6 +50,7 @@ class Codel final: public Queue<T> {
         queue_(),
         drop_func_(drop_func) { }
 
+  // deconstructor that drops all objects still left in the internal queue.  
   virtual ~Codel() {
     Wrapper w;
     while (!queue_.empty()) {
@@ -134,8 +135,15 @@ class Codel final: public Queue<T> {
     }
     return i;
   }
-
-  size_t Capacity() override { return queue_.max_size(); }
+  // the underlying queue is deque which is a dynamically sized queue with a max size 
+  // determined by system limit. Therefore, the capacity is a specified value used to
+  // limit the queue or if no value is specified, the queue's system limit.
+  size_t Capacity() override {
+    if (max_size_ != 0) {
+      return max_size_;
+    }
+    return queue_.max_size();
+  }
 
   bool Empty() override { return queue_.empty(); }
 
@@ -147,6 +155,21 @@ class Codel final: public Queue<T> {
   }
 
   size_t Size() override { return queue_.size(); }
+  
+  // The undelying queue is deque which is a dynamically sized queue with a max size 
+  // determined by system limits. Therefore, the resize method will error if the new_capacity
+  // is outside of the queue's system limits or otherwise, only change the imposed limit on
+  // the capacity of the queue.
+  int Resize(size_t new_capacity) override {
+    if (new_capacity <= Size()) {
+      return -1;
+    } else if (new_capacity >= queue_.max_size()) {
+      return -1;
+    }
+
+    max_size_ = new_capacity;
+    return 0;
+  }
 
  private:
   // Calls the drop_func on the object if the drop function exists

core/utils/codel_test.cc

@@ -8,6 +8,7 @@
 namespace {
 
 using bess::utils::Codel;
+using bess::utils::Queue;
 void integer_drop(int* ptr) {
   delete ptr;
 }
@@ -226,7 +227,7 @@ TEST(CodelTest, MultiPushPop) {
   EXPECT_EQ(c.Size(), n);
 
   int** output = new int*[n]; 
-  c.Pop(output, n);
+  ASSERT_EQ(c.Pop(output, n), n);
   for (int i = 0; i < n; i++) {
     ASSERT_EQ(output[i], vals[i]);
   }

core/utils/llqueue_test.cc

@@ -5,6 +5,7 @@
 namespace {
 
 using bess::utils::LockLessQueue;
+using bess::utils::Queue;
 
 // Simple test to make sure one can get back out object
 TEST(LLQueueTest, SingleInputOutput) {
@@ -53,6 +54,39 @@ TEST(LLQueueTest, MultiInputOutput) {
   }
 }
 
+// simple test to make sure that the queue is resized properly
+TEST(LLQueueTest, Resize) {
+  LockLessQueue<int*> q(8);
+
+  int n = 6;
+  int* vals1[n];
+  int* vals2[n];
+  for (int i = 0; i < n; i++) {
+    vals1[i] = new int();
+    vals2[i] = new int();
+  }
+  ASSERT_EQ(q.Push(vals1, n), n);
+  EXPECT_EQ(q.Size(), n);
+
+  ASSERT_EQ(q.Resize(16), 0);
+  ASSERT_EQ(q.Capacity(), 16);
+  ASSERT_EQ(q.Push(vals2, n), n);
+  ASSERT_EQ(q.Size(), 2 * n);
+
+  int** output = new int*[2*n]; 
+  ASSERT_EQ(q.Pop(output, 2 * n), 2*n);
+  for (int i = 0; i < n; i++) {
+    ASSERT_EQ(output[i], vals1[i]);
+    ASSERT_EQ(output[i + n], vals2[i]);
+  }
+
+  for (int i = 0; i < n; i++) {
+    delete vals1[i];
+    delete vals2[i];
+  }
+  delete[] output;
+}
+
 // simple test to make sure that multiple objects can be enqueued and dequeued
 // at the same time
 TEST(LLQueueTest, MultiPushPop) {
@@ -66,7 +100,7 @@ TEST(LLQueueTest, MultiPushPop) {
   EXPECT_EQ(q.Size(), n);
 
   int** output = new int*[n]; 
-  q.Pop(output, n);
+  ASSERT_EQ(q.Pop(output, n), n);
   for (int i = 0; i < n; i++) {
     ASSERT_EQ(output[i], vals[i]);
   }

core/utils/lock_less_queue.h

@@ -73,6 +73,33 @@ class LockLessQueue final : public Queue<T> {
   bool Empty() override { return llring_empty(ring_); }
 
   bool Full() override { return llring_full(ring_); }
+ 
+  int Resize(size_t new_capacity) override {
+    if (new_capacity <= Size() || (new_capacity & (new_capacity - 1))) {
+      return -1;
+    }
+
+    int err;
+    size_t ring_sz = llring_bytes_with_slots(new_capacity);
+    llring* new_ring = reinterpret_cast<struct llring*>(malloc(ring_sz));
+    CHECK(new_ring);
+    err = llring_init(new_ring, new_capacity, ring_->common.sp_enqueue,
+                      ring_->common.sc_dequeue);
+    if (err != 0) {
+      free(new_ring);
+      return err;
+    }
+
+    void* obj;
+    while (llring_dequeue(ring_, reinterpret_cast<void**>(&obj)) == 0) {
+      llring_enqueue(new_ring, obj);
+    }
+
+    free(ring_);
+    ring_ = new_ring;
+    capacity_ = new_capacity;
+    return 0;
+  }
 
  private:
   struct llring* ring_;  // class's ring buffer

core/utils/queue.h

@@ -39,6 +39,10 @@ class Queue {
 
   // Returns true if full and false otherwise
   virtual bool Full() = 0;
+
+  // Resizes the queue to the specified new capacity which must be larger than
+  // the current size. Returns 0 on success.
+  virtual int Resize(size_t) = 0;
 };
 
 }  // namespace utils