Update enqueue implementation to match paper.

Fix indentation.

Remove spurious change.

Parse args properly.

Author: squidarth

src/frontend/linkshell.cc

@@ -1,6 +1,7 @@
 /* -*-mode:c++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
 
 #include <getopt.h>
+#include <map>
 
 #include "infinite_packet_queue.hh"
 #include "drop_tail_packet_queue.hh"
@@ -10,6 +11,7 @@
 #include "red_packet_queue.hh"
 #include "link_queue.hh"
 #include "packetshell.cc"
+#include "tokenize.hh"
 
 using namespace std;
 
@@ -33,7 +35,7 @@ void usage_error( const string & program_name )
     throw runtime_error( "invalid arguments" );
 }
 
-unique_ptr<AbstractPacketQueue> get_packet_queue( const string & type, const string & args, const string & program_name )
+unique_ptr<AbstractPacketQueue> get_packet_queue( const string & type, const map<string, string> & args, const string & program_name )
 {
     if ( type == "infinite" ) {
         return unique_ptr<AbstractPacketQueue>( new InfinitePacketQueue( args ) );
@@ -71,6 +73,26 @@ string shell_quote( const string & arg )
     return ret;
 }
 
+map<string, string> parse_queue_args( const string & arg) {
+  map<string, string> argMap = map<string, string>();
+  if (arg.size() == 0) {
+    return argMap;
+  }
+  vector<string> argList = split(arg, ",");
+
+  for (size_t i = 0;i < argList.size();i++) {
+    string s = argList[i];
+    vector<string> argParts = split(s, "=");
+    if (argParts.size() != 2) {
+      throw runtime_error("Queue args passed in wrong format");
+    }
+
+    argMap.insert(pair<string, string>(argParts[0], argParts[1]));
+  }
+
+  return argMap;
+}
+
 int main( int argc, char *argv[] )
 {
     try {
@@ -99,7 +121,7 @@ int main( int argc, char *argv[] )
             { "meter-downlink-delay",       no_argument, nullptr, 'y' },
             { "meter-all",                  no_argument, nullptr, 'z' },
             { "uplink-queue",         required_argument, nullptr, 'q' },
-            { "downlink-queue",         required_argument, nullptr, 'w' },
+            { "downlink-queue",       required_argument, nullptr, 'w' },
             { "uplink-queue-args",    required_argument, nullptr, 'a' },
             { "downlink-queue-args",  required_argument, nullptr, 'b' },
             { 0,                                      0, nullptr, 0 }
@@ -186,11 +208,11 @@ int main( int argc, char *argv[] )
 
         link_shell_app.start_uplink( "[link] ", command,
                                      "Uplink", uplink_filename, uplink_logfile, repeat, meter_uplink, meter_uplink_delay,
-                                     get_packet_queue( uplink_queue_type, uplink_queue_args, argv[ 0 ] ),
+                                     get_packet_queue( uplink_queue_type, parse_queue_args(uplink_queue_args), argv[ 0 ] ),
                                      command_line );
 
         link_shell_app.start_downlink( "Downlink", downlink_filename, downlink_logfile, repeat, meter_downlink, meter_downlink_delay,
-                                       get_packet_queue( downlink_queue_type, downlink_queue_args, argv[ 0 ] ),
+                                       get_packet_queue( downlink_queue_type, parse_queue_args(downlink_queue_args), argv[ 0 ] ),
                                        command_line );
 
         return link_shell_app.wait_for_exit();

src/packet/Makefile.am

@@ -7,6 +7,6 @@ libpacket_a_SOURCES = packetshell.hh packetshell.cc queued_packet.hh \
                       abstract_packet_queue.hh dropping_packet_queue.hh dropping_packet_queue.cc infinite_packet_queue.hh \
                       drop_tail_packet_queue.hh drop_head_packet_queue.hh \
                       codel_packet_queue.cc codel_packet_queue.hh \
-		       red_packet_queue.cc red_packet_queue.hh \
+		        red_packet_queue.cc red_packet_queue.hh \
                       pie_packet_queue.cc pie_packet_queue.hh \
                       bindworkaround.hh

src/packet/codel_packet_queue.cc

@@ -5,10 +5,10 @@
 
 using namespace std;
 
-CODELPacketQueue::CODELPacketQueue( const string & args )
+CODELPacketQueue::CODELPacketQueue( const map<string, string> & args )
   : DroppingPacketQueue(args),
-    target_ ( get_arg( args, "target") ),
-    interval_ ( get_arg( args, "interval") ),
+    target_ ( get_int_arg( args, "target") ),
+    interval_ ( get_int_arg( args, "interval") ),
     first_above_time_ ( 0 ),
     drop_next_( 0 ),
     count_ ( 0 ),

src/packet/codel_packet_queue.hh

@@ -45,7 +45,7 @@ private:
     uint64_t control_law ( uint64_t t, uint32_t count );
 
 public:
-    CODELPacketQueue( const std::string & args );
+    CODELPacketQueue( const std::map<std::string, std::string> & args );
 
     void enqueue( QueuedPacket && p ) override;
 

src/packet/dropping_packet_queue.cc

@@ -8,15 +8,31 @@
 
 using namespace std;
 
-DroppingPacketQueue::DroppingPacketQueue( const string & args )
-    : packet_limit_( get_arg( args, "packets" ) ),
-      byte_limit_( get_arg( args, "bytes" ) )
+DroppingPacketQueue::DroppingPacketQueue( const map<string, string> & args )
+    : packet_limit_( get_int_arg( args, "packets" ) ),
+      byte_limit_( get_int_arg( args, "bytes" ) )
 {
     if ( packet_limit_ == 0 and byte_limit_ == 0 ) {
         throw runtime_error( "Dropping queue must have a byte or packet limit." );
     }
 }
 
+unsigned int DroppingPacketQueue::get_int_arg(const map<string, string> & args, const string & name) {
+  if (args.count(name) > 0) {
+    return myatoi(args.at(name));
+  } else {
+    return 0;
+  }
+}
+
+double DroppingPacketQueue::get_float_arg(const map<string, string> & args, const string & name) {
+  if (args.count(name) > 0) {
+    return myatof(args.at(name));
+  } else {
+    return 0;
+  }
+}
+
 QueuedPacket DroppingPacketQueue::dequeue( void )
 {
     assert( not internal_queue_.empty() );
@@ -98,66 +114,3 @@ string DroppingPacketQueue::to_string( void ) const
 
     return ret;
 }
-
-string DroppingPacketQueue::parse_number_arg(const string & args, const string & name, bool isfloat)
-{
-    auto offset = args.find( name );
-    if ( offset == string::npos ) {
-        return ""; /* default value */
-    } else {
-        /* extract the value */
-
-        /* advance by length of name */
-        offset += name.size();
-
-        /* make sure next char is "=" */
-        if ( args.substr( offset, 1 ) != "=" ) {
-            throw runtime_error( "could not parse queue arguments: " + args );
-        }
-
-        /* advance by length of "=" */
-        offset++;
-
-        /* find the first non-valid character */
-        string validchars;
-        if (isfloat) {
-            validchars = "0123456789.";
-        } else {
-            validchars = "0123456789";
-        }
-        auto offset2 = args.substr( offset ).find_first_not_of( validchars );
-
-        auto digit_string = args.substr( offset ).substr( 0, offset2 );
-
-        if ( digit_string.empty() ) {
-            throw runtime_error( "could not parse queue arguments: " + args );
-        }
-        return digit_string;
-    }
-}
-
-unsigned int DroppingPacketQueue::get_arg( const string & args, const string & name)
-{
-    auto offset = args.find( name );
-    if ( offset == string::npos ) {
-        return 0; /* default value */
-    } else {
-        auto digit_string = parse_number_arg(args, name, false);
-
-        return myatoi( digit_string );
-    }
-}
-
-double DroppingPacketQueue::get_float_arg( const string & args, const string & name)
-{
-    auto offset = args.find( name );
-    if ( offset == string::npos ) {
-        return 0; /* default value */
-    } else {
-        auto digit_string = parse_number_arg(args, name, true);
-        return myatof( digit_string );
-    }
-
-}
-
-

src/packet/dropping_packet_queue.hh

@@ -5,6 +5,7 @@
 
 #include <queue>
 #include <cassert>
+#include <map>
 
 #include "abstract_packet_queue.hh"
 #include "exception.hh"
@@ -31,7 +32,7 @@ protected:
                     const unsigned int size_in_packets ) const;
 
 public:
-    DroppingPacketQueue( const std::string & args );
+    DroppingPacketQueue( const std::map<std::string, std::string> & args );
 
     virtual void enqueue( QueuedPacket && p ) = 0;
 
@@ -42,12 +43,13 @@ public:
     std::string to_string( void ) const override;
 
     static std::string parse_number_arg(const std::string & args, const std::string & name, bool isfloat);
+
     unsigned int size_bytes( void ) const override;
     unsigned int size_packets( void ) const override;
 
+    static unsigned int get_int_arg( const std::map<std::string, std::string> & args, const std::string & name );
     static unsigned int get_arg( const std::string & args, const std::string & name );
-
-    static double get_float_arg( const std::string & args, const std::string & name );
+    static double get_float_arg( const std::map<std::string, std::string> & args, const std::string & name );
 };
 
 #endif /* DROPPING_PACKET_QUEUE_HH */ 

src/packet/infinite_packet_queue.hh

@@ -9,6 +9,7 @@
 #include "queued_packet.hh"
 #include "abstract_packet_queue.hh"
 #include "exception.hh"
+#include <map>
 
 class InfinitePacketQueue : public AbstractPacketQueue
 {
@@ -17,7 +18,7 @@ private:
     int queue_size_in_bytes_ = 0, queue_size_in_packets_ = 0;
 
 public:
-    InfinitePacketQueue( const std::string & args )
+    InfinitePacketQueue( const std::map<std::string, std::string> & args )
     {
         if ( not args.empty() ) {
             throw std::runtime_error( "InfinitePacketQueue does not take arguments." );

src/packet/pie_packet_queue.cc

@@ -8,10 +8,10 @@ using namespace std;
 
 #define DQ_COUNT_INVALID   (uint32_t)-1
 
-PIEPacketQueue::PIEPacketQueue( const string & args )
+PIEPacketQueue::PIEPacketQueue( const map<string, string> & args )
   : DroppingPacketQueue(args),
-    qdelay_ref_ ( get_arg( args, "qdelay_ref" ) ),
-    max_burst_ ( get_arg( args, "max_burst" ) ),
+    qdelay_ref_ ( get_int_arg( args, "qdelay_ref" ) ),
+    max_burst_ ( get_int_arg( args, "max_burst" ) ),
     alpha_ ( 0.125 ),
     beta_ ( 1.25 ),
     t_update_ ( 30 ),

src/packet/pie_packet_queue.hh

@@ -52,7 +52,7 @@ private:
     void calculate_drop_prob ( void );
 
 public:
-    PIEPacketQueue( const std::string & args );
+    PIEPacketQueue( const std::map<std::string, std::string> & args );
 
     void enqueue( QueuedPacket && p ) override;
 

src/packet/red_packet_queue.cc

@@ -4,19 +4,24 @@
 
 using namespace std;
 
-REDPacketQueue::REDPacketQueue( const string & args)
+REDPacketQueue::REDPacketQueue( const map<string, string> & args)
   : DroppingPacketQueue(args),
     wq_(get_float_arg(args, "wq")),
     min_thresh_(get_float_arg(args, "minthresh")),
     max_thresh_(get_float_arg(args, "maxthresh")),
-    time_at_zero_q_(0)
+    transmission_time_(get_int_arg(args, "transmission_time")),
+    time_at_zero_q_(0),
+    prng_( random_device()() ),
+    drop_dist_ (0, 1),
+    current_random_val_(0),
+    count_(0)
 {
   if (packet_limit_ == 0) {
     throw runtime_error( "RED queue must have packet limit." );
   }
 
-  if ( wq_ == 0.0 || min_thresh_ == 0.0 || max_thresh_ == 0.0 ) {
-    throw runtime_error( "RED queue must have wq, minthresh, and maxthresh arguments." );
+  if ( wq_ == 0.0 || min_thresh_ == 0.0 || max_thresh_ == 0.0 || transmission_time_ == 0 ) {
+    throw runtime_error( "RED queue must have wq, minthresh, maxthresh, and transmission_time arguments." );
   }
 
 }
@@ -33,32 +38,60 @@ QueuedPacket REDPacketQueue::dequeue( void )
 
 void REDPacketQueue::enqueue( QueuedPacket && p )
 {
-    int s = 4;
     auto instantaneous_queue_size = size_packets();
 
     /* If the queue is empty, decay the weighted average by
      * the amount of time that the queue has been
      * empty. */
     if (size_packets() == 0) {
-        weighted_average_ = powf((1 - wq_), (timestamp() - time_at_zero_q_)/s) * weighted_average_;
+        weighted_average_ = powf((1 - wq_), (timestamp() - time_at_zero_q_)/transmission_time_) * weighted_average_;
     } else {
         weighted_average_ = (instantaneous_queue_size * wq_ ) + (1- wq_) * weighted_average_;
     }
 
+
+    /* ratio is how full the queue is percentage-wise */
     auto ratio = (weighted_average_)/packet_limit_;
-    std::default_random_engine generator (0);
-    std::uniform_real_distribution<double> distribution (min_thresh_, max_thresh_);
-    double threshold = distribution(generator);
-    if (ratio > max_thresh_) {
-      ratio = 1;
-    }
-    if (ratio < min_thresh_) {
-      ratio = 0;
+    bool accept_packet = true;
+
+    /*
+     * Logic for enqueuing packets:
+     *  - If ratio is between the min and max threshold,
+     *  begin counting the number of packets that have
+     *  passed through the queue. We compute a drop
+     *  probability based on `ratio`, and if the
+     *  number of packets in the range exceeds
+     *  some random number divided by that drop probability,
+     *  drop the packet, and reset teh count.
+     *  - If ratio is greater than the max_thresh_, drop
+     * the packet.
+     *  - If it is under max thresh, don't
+     *  drop the packet.
+     *  - See: http://www.icir.org/floyd/papers/early.twocolumn.pdf
+     *  for more details
+     */
+    if (ratio >= min_thresh_ && ratio <= max_thresh_) {
+      count_++;
+      float drop_probability = (ratio - min_thresh_)/(max_thresh_ - min_thresh_);
+
+      if (count_ > 0 && count_ >= current_random_val_/drop_probability ) {
+        accept_packet = false;
+        count_ = 0;
+      }
+
+      if (count_ == 0) {
+        current_random_val_ = drop_dist_(prng_);
+      }
+    } else if (ratio > max_thresh_ ) {
+      accept_packet = false;
+      count_ = 0;
+    } else {
+      count_ = -1;
     }
 
-    if ( (threshold > ratio) && good_with( size_bytes() + p.contents.size(),
+    if ( accept_packet && good_with( size_bytes() + p.contents.size(),
                     size_packets() + 1 ) ) {
-        accept( std::move( p ) );
+        accept( move( p ) );
     }
 
     assert( good() );