Integrated Codel into DRR

Author: TheJStone

bessctl/conf/testing/module_tests/drr.py

@@ -4,11 +4,14 @@
 from port import *
 
 #ensures that given infinite input that the module does not crash.
-def crash_test():
-    return [DRR(), 1, 1]
+def crash_test(queue_type, queue_dict):
+    if queue_type == 1:
+        return [DRR(type=queue_type, codel=queue_dict), 1, 1]
+    else:
+        return [DRR(type=queue_type, ring=queue_dict), 1, 1]
 
-# tests to make that inividual packets gets through the module
-def basic_output_test():
+# tests to make that individual packets gets through the module
+def basic_output_test(queue_type, queue_dict):
 
     # produces the number of duplicate packets specified by num_pkts and uses the provided
     # the specifications in the packet along with dummy ports.
@@ -19,16 +22,28 @@ def gen_packet_list(protocol, input_ip, output_ip, num_pkts):
             packet_list.append({'input_port': 0, 'input_packet': cur_pkt,
                                     'output_port': 0, "output_packet": cur_pkt})
         return packet_list
-
-    single_basic = DRR(num_flows=2, max_flow_queue_size= 100)
+    
+    if queue_type== 1:
+        single_basic = DRR(num_flows=2, max_flow_queue_size= 100, type=queue_type, 
+            codel=queue_dict);
+    else:
+        single_basic = DRR(num_flows=2, max_flow_queue_size= 100, type=queue_type, 
+            ring=queue_dict);
+        
     monitor_task(single_basic, 0)
 
     out = []
     single_packet= gen_packet_list(scapy.TCP, '22.22.22.22', '22.22.22.22', 1)
     out.append([single_basic,  # test this module
                                1, 1,  # it has one input port and one output port
                                single_packet])
-    batch_basic = DRR(num_flows=4, max_flow_queue_size= 100)
+    if queue_type== 1:
+        batch_basic = DRR(num_flows=4, max_flow_queue_size= 100, type=queue_type, 
+            codel=queue_dict);
+    else:
+        batch_basic = DRR(num_flows=4, max_flow_queue_size= 100, type=queue_type, 
+            ring=queue_dict);
+    
     monitor_task(batch_basic, 0)
     packet_list = gen_packet_list(scapy.TCP, '22.22.22.1', '22.22.22.1', 2)
     packet_list += gen_packet_list(scapy.TCP, '22.22.11.1', '22.22.11.1', 2)
@@ -44,7 +59,7 @@ def fairness_test():
     # Takes the number of flows n, the quantum to give drr, the list packet rates for each flow 
     # and the packet rate for the module. runs this setup for five seconds and tests that 
     # throughput for each flow had a jaine fairness of atleast .95.
-    def fairness_n_flow_test(n, quantum, rates, module_rate):
+    def fairness_n_flow_test(n, quantum, rates, module_rate, queue_type, queue_dict):
         err = bess.reset_all()
 
         packets = []
@@ -65,7 +80,11 @@ def fairness_n_flow_test(n, quantum, rates, module_rate):
 
         me_out = Measure() 
         snk = Sink()
-        q = DRR(num_flows= n+1, quantum=quantum) 
+        if queue_type == 1:
+            q = DRR(num_flows= n+1, quantum=quantum, type=queue_type, codel=queue_dict)
+        else:
+            q = DRR(num_flows= n+1, quantum=quantum, type=queue_type, ring=queue_dict)
+
         me_in -> q -> me_out -> exm
 
         measure_out = []
@@ -96,12 +115,16 @@ def fairness_n_flow_test(n, quantum, rates, module_rate):
         else:
             fair = f(me_out)/square_sum
         assert abs(.99 - fair) <=.05
-
-    fairness_n_flow_test(2, 1000, [80000, 20000], 30000)
-    fairness_n_flow_test(5, 1000, [110000, 200000, 70000, 60000, 40000], 150000)
+    llqueue_dict = {} # all default values
+    codel_dict = {} # all default values
+    fairness_n_flow_test(2, 1000, [80000, 20000], 30000, 0, llqueue_dict)
+    fairness_n_flow_test(2, 1000, [80000, 20000], 30000, 1, codel_dict)
+    fairness_n_flow_test(5, 1000, [110000, 200000, 70000, 60000, 40000], 150000, 0, llqueue_dict)
+    fairness_n_flow_test(5, 1000, [110000, 200000, 70000, 60000, 40000], 150000, 1, codel_dict)
 
     ten_flows =  [210000, 120000, 130000, 160000, 100000, 105000, 90000, 70000, 60000, 40000]
-    fairness_n_flow_test(10, 1000, ten_flows , 300000)
+    fairness_n_flow_test(10, 1000, ten_flows , 300000, 0, llqueue_dict)
+    fairness_n_flow_test(10, 1000, ten_flows , 300000, 1, codel_dict)
 
     # hund_flows= []
     # cur = 200000
@@ -110,6 +133,10 @@ def fairness_n_flow_test(n, quantum, rates, module_rate):
         # cur -= 1600
     # fairness_n_flow_test(100, 1000, hund_flows, 300000)
 
-OUTPUT_TEST_INPUTS += basic_output_test()
+llqueue_dict = {} # all default values llqueue queue type: 0
+codel_dict = {} # all default values codel queue type: 1
+OUTPUT_TEST_INPUTS += basic_output_test(0, llqueue_dict)
+OUTPUT_TEST_INPUTS += basic_output_test(1, codel_dict)
 CUSTOM_TEST_FUNCTIONS.append(fairness_test)
-CRASH_TEST_INPUTS.append(crash_test())
+CRASH_TEST_INPUTS.append(crash_test(0, llqueue_dict))
+CRASH_TEST_INPUTS.append(crash_test(1, codel_dict))

core/modules/drr.cc

@@ -8,6 +8,11 @@
 #include "../utils/ether.h"
 #include "../utils/ip.h"
 #include "../utils/udp.h"
+#include "../utils/common.h"
+#include "../utils/codel.h"
+
+using bess::utils::Codel;
+using bess::pb::DRRArg;
 
 uint32_t RoundToPowerTwo(uint32_t v) {
   v--;
@@ -31,7 +36,8 @@ DRR::DRR()
       max_queue_size_(kFlowQueueMax),
       max_number_flows_(kDefaultNumFlows),
       flow_ring_(nullptr),
-      current_flow_(nullptr) {}
+      current_flow_(nullptr),
+      generator_(NULL) {}
 
 DRR::~DRR() {
   for (auto it = flows_.begin(); it != flows_.end();) {
@@ -41,7 +47,7 @@ DRR::~DRR() {
   std::free(flow_ring_);
 }
 
-CommandResponse DRR::Init(const bess::pb::DRRArg& arg) {
+CommandResponse DRR::Init(const DRRArg& arg) {
   CommandResponse err;
   task_id_t tid;
 
@@ -70,11 +76,37 @@ CommandResponse DRR::Init(const bess::pb::DRRArg& arg) {
   }
 
   int err_num = 0;
-  flow_ring_ = AddQueue(max_number_flows_, &err_num);
+  flow_ring_ = new LockLessQueue<Flow*>(max_number_flows_, true, true);
   if (err_num != 0) {
     return CommandFailure(-err_num);
   }
 
+  /* get flow queue generator */
+  if (arg.type() == DRRArg::CODEL) {
+    const DRRArg::Codel& codel = arg.codel();
+    uint64_t target;
+    uint64_t window;
+    if (!(target = codel.target())) {
+      target = Codel<bess::Packet*>::kDefaultTarget;
+    }
+
+    if (!(window = codel.window())) {
+      window = Codel<bess::Packet*>::kDefaultWindow;
+    }
+
+    generator_ =
+        Codel<bess::Packet*>::Factory(bess::Packet::Free, max_queue_size_, target, window);
+  } else {
+    const DRRArg::Llring& ring = arg.ring();
+    size_t size;
+    if (!(size = ring.size())) {
+      size = LockLessQueue<bess::Packet*>::kDefaultRingSize;
+    }
+
+    generator_ = LockLessQueue<bess::Packet*>::Factory(
+        size, !ring.multiple_producer(), !ring.multiple_consumer());
+  }
+
   return CommandSuccess();
 }
 
@@ -101,7 +133,7 @@ void DRR::ProcessBatch(bess::PacketBatch* batch) {
     // if the Flow doesn't exist create one
     // and add the packet to the new Flow
     if (it == nullptr) {
-      if (llring_full(flow_ring_)) {
+      if (flow_ring_->Full()) {
         bess::Packet::Free(pkt);
       } else {
         AddNewFlow(pkt, id, &err);
@@ -140,7 +172,7 @@ struct task_result DRR::RunTask(void*) {
 uint32_t DRR::GetNextBatch(bess::PacketBatch* batch, int* err) {
   Flow* f;
   uint32_t total_bytes = 0;
-  uint32_t count = llring_count(flow_ring_);
+  uint32_t count = flow_ring_->Size();
   if (current_flow_) {
     count++;
   }
@@ -155,33 +187,30 @@ uint32_t DRR::GetNextBatch(bess::PacketBatch* batch, int* err) {
       if (batch_size == batch->cnt()) {
         break;
       } else {
-        count = llring_count(flow_ring_);
+        count = flow_ring_->Size();
         batch_size = batch->cnt();
       }
     }
     count--;
 
     f = GetNextFlow(err);
-    if (*err != 0) {
+    if (*err) {
       return total_bytes;
     } else if (f == nullptr) {
       continue;
     }
 
-    uint32_t bytes = GetNextPackets(batch, f, err);
+    uint32_t bytes = GetNextPackets(batch, f);
     total_bytes += bytes;
-    if (*err != 0) {
-      return total_bytes;
-    }
 
-    if (llring_empty(f->queue) && !f->next_packet) {
+    if (f->queue->Empty() && !f->next_packet) {
       f->deficit = 0;
     }
 
     // if the flow doesn't have any more packets to give, reenqueue it
     if (!f->next_packet || f->next_packet->total_len() > f->deficit) {
-      *err = llring_enqueue(flow_ring_, f);
-      if (*err != 0) {
+      *err = flow_ring_->Push(f);
+      if (*err) {
         return total_bytes;
       }
     } else {
@@ -198,18 +227,18 @@ DRR::Flow* DRR::GetNextFlow(int* err) {
   double now = get_epoch_time();
 
   if (!current_flow_) {
-    *err = llring_dequeue(flow_ring_, reinterpret_cast<void**>(&f));
-    if (*err < 0) {
+    *err = flow_ring_->Pop(f);
+    if (*err) {
       return nullptr;
     }
 
-    if (llring_empty(f->queue) && !f->next_packet) {
+    if (f->queue->Empty() && !f->next_packet) {
       // if the flow expired, remove it
       if (now - f->timer > kTtl) {
         RemoveFlow(f);
       } else {
-        *err = llring_enqueue(flow_ring_, f);
-        if (*err < 0) {
+        *err = flow_ring_->Push(f);
+        if (*err) {
           return nullptr;
         }
       }
@@ -224,15 +253,15 @@ DRR::Flow* DRR::GetNextFlow(int* err) {
   return f;
 }
 
-uint32_t DRR::GetNextPackets(bess::PacketBatch* batch, Flow* f, int* err) {
+uint32_t DRR::GetNextPackets(bess::PacketBatch* batch, Flow* f) {
   uint32_t total_bytes = 0;
   bess::Packet* pkt;
 
-  while (!batch->full() && (!llring_empty(f->queue) || f->next_packet)) {
+  while (!batch->full() && (!f->queue->Empty() || f->next_packet)) {
     // makes sure there isn't already a packet at the front
     if (!f->next_packet) {
-      *err = llring_dequeue(f->queue, reinterpret_cast<void**>(&pkt));
-      if (*err < 0) {
+      int err = f->queue->Pop(pkt);
+      if (err) {
         return total_bytes;
       }
     } else {
@@ -274,21 +303,21 @@ void DRR::AddNewFlow(bess::Packet* pkt, FlowId id, int* err) {
   Flow* f = new Flow(id);
 
   // TODO(joshua) do proper error checking
-  f->queue = AddQueue(static_cast<int>(kFlowQueueSize), err);
+  f->queue = generator_();
 
-  if (*err != 0) {
+  if (*err) {
     return;
   }
 
   flows_.Insert(id, f);
 
   Enqueue(f, pkt, err);
-  if (*err != 0) {
+  if (*err) {
     return;
   }
 
   // puts flow in round robin
-  *err = llring_enqueue(flow_ring_, f);
+  *err = flow_ring_->Push(f);
 }
 
 void DRR::RemoveFlow(Flow* f) {
@@ -299,77 +328,34 @@ void DRR::RemoveFlow(Flow* f) {
   delete f;
 }
 
-llring* DRR::AddQueue(uint32_t slots, int* err) {
-  int bytes = llring_bytes_with_slots(slots);
-  int ret;
-
-  llring* queue = static_cast<llring*>(aligned_alloc(alignof(llring), bytes));
-  if (!queue) {
-    *err = -ENOMEM;
-    return nullptr;
-  }
-
-  ret = llring_init(queue, slots, 1, 1);
-  if (ret) {
-    std::free(queue);
-    *err = -EINVAL;
-    return nullptr;
-  }
-  return queue;
-}
 
 void DRR::Enqueue(Flow* f, bess::Packet* newpkt, int* err) {
   // if the queue is full. drop the packet.
-  if (llring_count(f->queue) >= max_queue_size_) {
+  if (f->queue->Size() >= max_queue_size_) {
     bess::Packet::Free(newpkt);
     return;
   }
 
   // creates a new queue if there is not enough space for the new packet
   // in the old queue
-  if (llring_full(f->queue)) {
+  if (f->queue->Full()) {
     uint32_t slots =
-        RoundToPowerTwo(llring_count(f->queue) * kQueueGrowthFactor);
-    f->queue = ResizeQueue(f->queue, slots, err);
-    if (*err != 0) {
+        RoundToPowerTwo(f->queue->Size() * kQueueGrowthFactor);
+    *err = f->queue->Resize(slots);
+    if (*err) {
       bess::Packet::Free(newpkt);
       return;
     }
   }
 
-  *err = llring_enqueue(f->queue, reinterpret_cast<void*>(newpkt));
-  if (*err == 0) {
+  *err = f->queue->Push(newpkt);
+  if (!*err) {
     f->timer = get_epoch_time();
   } else {
     bess::Packet::Free(newpkt);
   }
 }
 
-llring* DRR::ResizeQueue(llring* old_queue, uint32_t new_size, int* err) {
-  llring* new_queue = AddQueue(new_size, err);
-  if (*err != 0) {
-    return nullptr;
-  }
-
-  // migrates packets from the old queue
-  if (old_queue) {
-    bess::Packet* pkt;
-
-    while (llring_dequeue(old_queue, reinterpret_cast<void**>(&pkt)) == 0) {
-      *err = llring_enqueue(new_queue, pkt);
-      if (*err == -LLRING_ERR_NOBUF) {
-        bess::Packet::Free(pkt);
-        *err = 0;
-      } else if (*err != 0) {
-        std::free(new_queue);
-        return nullptr;
-      }
-    }
-
-    std::free(old_queue);
-  }
-  return new_queue;
-}
 
 CommandResponse DRR::SetQuantumSize(uint32_t size) {
   if (size == 0) {