More Simulation Framework features

webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc

Index: webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc
diff --git a/webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc b/webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc
index ad1f5fd5bf0f3f4bb7f7f2d0367430d430923a66..5f1beb94b0db50efdac5dba1e556b351ae184669 100644
--- a/webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc
+++ b/webrtc/modules/remote_bitrate_estimator/test/bwe_test_framework.cc
@@ -20,9 +20,10 @@ namespace bwe {
 
 class DelayCapHelper {
  public:
+  // Max delay = 0 stands for +infinite.
   DelayCapHelper() : max_delay_us_(0), delay_stats_() {}
 
-  void SetMaxDelay(int max_delay_ms) {
+  void SetMaxDelayMs(int64_t max_delay_ms) {
     BWE_TEST_LOGGING_ENABLE(false);
     BWE_TEST_LOGGING_LOG1("Max Delay", "%d ms", static_cast<int>(max_delay_ms));
     assert(max_delay_ms >= 0);
@@ -51,48 +52,33 @@ const FlowIds CreateFlowIds(const int *flow_ids_array, size_t num_flow_ids) {
   return flow_ids;
 }
 
-class RateCounter {
- public:
-  RateCounter()
-      : kWindowSizeUs(1000000),
-        packets_per_second_(0),
-        bytes_per_second_(0),
-        last_accumulated_us_(0),
-        window_() {}
-
-  void UpdateRates(int64_t send_time_us, uint32_t payload_size) {
-    packets_per_second_++;
-    bytes_per_second_ += payload_size;
-    last_accumulated_us_ = send_time_us;
-    window_.push_back(std::make_pair(send_time_us, payload_size));
-    while (!window_.empty()) {
-      const TimeSizePair& packet = window_.front();
-      if (packet.first > (last_accumulated_us_ - kWindowSizeUs)) {
-        break;
-      }
-      assert(packets_per_second_ >= 1);
-      assert(bytes_per_second_ >= packet.second);
-      packets_per_second_--;
-      bytes_per_second_ -= packet.second;
-      window_.pop_front();
-    }
+const FlowIds CreateFlowIdRange(int initial_value, int last_value) {
+  int size = last_value - initial_value + 1;
+  assert(size > 0);
+  int* flow_ids_array = new int[size];
+  for (int i = initial_value; i <= last_value; ++i) {
+    flow_ids_array[i - initial_value] = i;
   }
-
-  uint32_t bits_per_second() const {
-    return bytes_per_second_ * 8;
+  return CreateFlowIds(flow_ids_array, size);
+}
+
+void RateCounter::UpdateRates(int64_t send_time_us, uint32_t payload_size) {
+  packets_per_second_++;
+  bytes_per_second_ += payload_size;
+  last_accumulated_us_ = send_time_us;
+  window_.push_back(std::make_pair(send_time_us, payload_size));
+  while (!window_.empty()) {
+    const TimeSizePair& packet = window_.front();
+    if (packet.first > (last_accumulated_us_ - kWindowSizeUs)) {
+      break;
+    }
+    assert(packets_per_second_ >= 1);
+    assert(bytes_per_second_ >= packet.second);
+    packets_per_second_--;
+    bytes_per_second_ -= packet.second;
+    window_.pop_front();
   }
-
-  uint32_t packets_per_second() const { return packets_per_second_; }
-
- private:
-  typedef std::pair<int64_t, uint32_t> TimeSizePair;
-
-  const int64_t kWindowSizeUs;
-  uint32_t packets_per_second_;
-  uint32_t bytes_per_second_;
-  int64_t last_accumulated_us_;
-  std::list<TimeSizePair> window_;
-};
+}
 
 Random::Random(uint32_t seed)
     : a_(0x531FDB97 ^ seed),
@@ -107,6 +93,11 @@ float Random::Rand() {
   return result;
 }
 
+int Random::Rand(int low, int high) {
+  float uniform = Rand() * (high - low + 1) + low;
+  return static_cast<int>(uniform);
+}
+
 int Random::Gaussian(int mean, int standard_deviation) {
   // Creating a Normal distribution variable from two independent uniform
   // variables based on the Box-Muller transform, which is defined on the
@@ -121,6 +112,11 @@ int Random::Gaussian(int mean, int standard_deviation) {
       sqrt(-2 * log(u1)) * cos(2 * kPi * u2));
 }
 
+int Random::Exponential(float lambda) {
+  float uniform = Rand();
+  return static_cast<int>(-log(uniform) / lambda);
+}
+
 Packet::Packet()
     : flow_id_(0), creation_time_us_(-1), send_time_us_(-1), payload_size_(0) {
 }
@@ -209,7 +205,9 @@ bool IsTimeSorted(const Packets& packets) {
 PacketProcessor::PacketProcessor(PacketProcessorListener* listener,
                                  int flow_id,
                                  ProcessorType type)
-    : listener_(listener), flow_ids_(&flow_id, &flow_id + 1) {
+    : rate_counter_(new RateCounter()),
+      listener_(listener),
+      flow_ids_(&flow_id, &flow_id + 1) {
   if (listener_) {
     listener_->AddPacketProcessor(this, type);
   }
@@ -218,7 +216,9 @@ PacketProcessor::PacketProcessor(PacketProcessorListener* listener,
 PacketProcessor::PacketProcessor(PacketProcessorListener* listener,
                                  const FlowIds& flow_ids,
                                  ProcessorType type)
-    : listener_(listener), flow_ids_(flow_ids) {
+    : rate_counter_(new RateCounter()),
+      listener_(listener),
+      flow_ids_(flow_ids) {
   if (listener_) {
     listener_->AddPacketProcessor(this, type);
   }
@@ -230,14 +230,22 @@ PacketProcessor::~PacketProcessor() {
   }
 }
 
+uint32_t PacketProcessor::packets_per_second() const {
+  return rate_counter_->packets_per_second();
+}
+
+uint32_t PacketProcessor::bits_per_second() const {
+  return rate_counter_->bits_per_second();
+}

[stefan-webrtc, 2015/07/02 11:03:41]

Good suggestion to move this to the PacketProcessors!

+
 RateCounterFilter::RateCounterFilter(PacketProcessorListener* listener,
                                      int flow_id,
                                      const char* name)
     : PacketProcessor(listener, flow_id, kRegular),
-      rate_counter_(new RateCounter()),
       packets_per_second_stats_(),
       kbps_stats_(),
-      name_() {
+      name_(),
+      start_plotting_time_ms_(0) {
   std::stringstream ss;
   ss << name << "_" << flow_id;
   name_ = ss.str();
@@ -247,10 +255,10 @@ RateCounterFilter::RateCounterFilter(PacketProcessorListener* listener,
                                      const FlowIds& flow_ids,
                                      const char* name)
     : PacketProcessor(listener, flow_ids, kRegular),
-      rate_counter_(new RateCounter()),
       packets_per_second_stats_(),
       kbps_stats_(),
-      name_() {
+      name_(),
+      start_plotting_time_ms_(0) {
   std::stringstream ss;
   ss << name << "_";
   for (int flow_id : flow_ids) {
@@ -259,17 +267,18 @@ RateCounterFilter::RateCounterFilter(PacketProcessorListener* listener,
   name_ = ss.str();
 }
 
-RateCounterFilter::~RateCounterFilter() {
-  LogStats();
+RateCounterFilter::RateCounterFilter(PacketProcessorListener* listener,
+                                     const FlowIds& flow_ids,
+                                     const char* name,
+                                     int64_t start_plotting_time_ms)
+    : RateCounterFilter(listener, flow_ids, name) {
+  start_plotting_time_ms_ = start_plotting_time_ms;
 }
 
-uint32_t RateCounterFilter::packets_per_second() const {
-  return rate_counter_->packets_per_second();
+RateCounterFilter::~RateCounterFilter() {
+  LogStats();
 }
 
-uint32_t RateCounterFilter::bits_per_second() const {
-  return rate_counter_->bits_per_second();
-}
 
 void RateCounterFilter::LogStats() {
   BWE_TEST_LOGGING_CONTEXT("RateCounterFilter");
@@ -282,9 +291,14 @@ Stats<double> RateCounterFilter::GetBitrateStats() const {
 }
 
 void RateCounterFilter::Plot(int64_t timestamp_ms) {
+  uint32_t plot_kbps = 0;
+  if (timestamp_ms >= start_plotting_time_ms_) {
+    plot_kbps = rate_counter_->bits_per_second() / 1000.0;
+  }
   BWE_TEST_LOGGING_CONTEXT(name_.c_str());
-  BWE_TEST_LOGGING_PLOT(0, "Throughput_#1", timestamp_ms,
-                        rate_counter_->bits_per_second() / 1000.0);
+  BWE_TEST_LOGGING_PLOT(0, "Throughput_#1", timestamp_ms, plot_kbps);
+  // Silencing unused variable compiling error.

[stefan-webrtc, 2015/07/02 11:03:41]

Remove this comment, RTC_UNUSED is enough

[magalhaesc, 2015/07/02 17:06:18]

Done.

+  RTC_UNUSED(plot_kbps);
 }
 
 void RateCounterFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
@@ -330,30 +344,32 @@ void LossFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
   }
 }
 
+const int64_t kDefaultOneWayDelayUs = 0;
+
 DelayFilter::DelayFilter(PacketProcessorListener* listener, int flow_id)
     : PacketProcessor(listener, flow_id, kRegular),
-      delay_us_(0),
+      one_way_delay_us_(kDefaultOneWayDelayUs),
       last_send_time_us_(0) {
 }
 
 DelayFilter::DelayFilter(PacketProcessorListener* listener,
                          const FlowIds& flow_ids)
     : PacketProcessor(listener, flow_ids, kRegular),
-      delay_us_(0),
+      one_way_delay_us_(kDefaultOneWayDelayUs),
       last_send_time_us_(0) {
 }
 
-void DelayFilter::SetDelayMs(int64_t delay_ms) {
+void DelayFilter::SetOneWayDelayMs(int64_t one_way_delay_ms) {
   BWE_TEST_LOGGING_ENABLE(false);
-  BWE_TEST_LOGGING_LOG1("Delay", "%d ms", static_cast<int>(delay_ms));
-  assert(delay_ms >= 0);
-  delay_us_ = delay_ms * 1000;
+  BWE_TEST_LOGGING_LOG1("Delay", "%d ms", static_cast<int>(one_way_delay_ms));
+  assert(one_way_delay_ms >= 0);
+  one_way_delay_us_ = one_way_delay_ms * 1000;
 }
 
 void DelayFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
   assert(in_out);
   for (Packet* packet : *in_out) {
-    int64_t new_send_time_us = packet->send_time_us() + delay_us_;
+    int64_t new_send_time_us = packet->send_time_us() + one_way_delay_us_;
     last_send_time_us_ = std::max(last_send_time_us_, new_send_time_us);
     packet->set_send_time_us(last_send_time_us_);
   }
@@ -431,27 +447,52 @@ void ReorderFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
   }
 }
 
+const uint32_t kDefaultKbps = 1200;
+
 ChokeFilter::ChokeFilter(PacketProcessorListener* listener, int flow_id)
     : PacketProcessor(listener, flow_id, kRegular),
-      kbps_(1200),
+      capacity_kbps_(kDefaultKbps),
       last_send_time_us_(0),
       delay_cap_helper_(new DelayCapHelper()) {
+  running_flows_.insert(flow_id);
 }
 
 ChokeFilter::ChokeFilter(PacketProcessorListener* listener,
                          const FlowIds& flow_ids)
     : PacketProcessor(listener, flow_ids, kRegular),
-      kbps_(1200),
+      capacity_kbps_(kDefaultKbps),
       last_send_time_us_(0),
-      delay_cap_helper_(new DelayCapHelper()) {
+      delay_cap_helper_(new DelayCapHelper()),
+      running_flows_(flow_ids) {
+}
+
+void ChokeFilter::PauseFlow(int flow_id) {
+  running_flows_.erase(flow_id);
+}
+
+void ChokeFilter::ResumeFlow(int flow_id) {
+  running_flows_.insert(flow_id);
 }
 
 ChokeFilter::~ChokeFilter() {}
 
-void ChokeFilter::SetCapacity(uint32_t kbps) {
+void ChokeFilter::SetCapacityKbps(uint32_t kbps) {
   BWE_TEST_LOGGING_ENABLE(false);
   BWE_TEST_LOGGING_LOG1("BitrateChoke", "%d kbps", kbps);
-  kbps_ = kbps;
+  capacity_kbps_ = kbps;
+}
+
+uint32_t ChokeFilter::TotalAvailableKbps() {
+  return capacity_kbps_;
+}
+
+uint32_t ChokeFilter::AvailablePerFlowKbps(int flow_id) {

[stefan-webrtc, 2015/07/02 11:03:41]

Or just a getter for the total capacity, and then you can have the test compute
the per flow rate based on what flows it has running? Just to avoid having Pause
and Resume on a Choke, which doesn't seem too logical.

[magalhaesc, 2015/07/02 17:06:18]

Right, the per flow one was moved to a new class in metric_recorder

+  uint32_t available_capacity_per_flow_kbps = 0;
+  if (running_flows_.find(flow_id) != running_flows_.end()) {
+    available_capacity_per_flow_kbps =
+        capacity_kbps_ / static_cast<uint32_t>(running_flows_.size());
+  }
+  return available_capacity_per_flow_kbps;
 }
 
 void ChokeFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
@@ -459,9 +500,12 @@ void ChokeFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
   for (PacketsIt it = in_out->begin(); it != in_out->end(); ) {
     int64_t earliest_send_time_us =
         last_send_time_us_ +
-        ((*it)->payload_size() * 8 * 1000 + kbps_ / 2) / kbps_;
+        ((*it)->payload_size() * 8 * 1000 + capacity_kbps_ / 2) /
+            capacity_kbps_;
+
     int64_t new_send_time_us =
         std::max((*it)->send_time_us(), earliest_send_time_us);
+
     if (delay_cap_helper_->ShouldSendPacket(new_send_time_us,
                                             (*it)->send_time_us())) {
       (*it)->set_send_time_us(new_send_time_us);
@@ -474,8 +518,8 @@ void ChokeFilter::RunFor(int64_t /*time_ms*/, Packets* in_out) {
   }
 }
 
-void ChokeFilter::SetMaxDelay(int max_delay_ms) {
-  delay_cap_helper_->SetMaxDelay(max_delay_ms);
+void ChokeFilter::SetMaxDelayMs(int64_t max_delay_ms) {
+  delay_cap_helper_->SetMaxDelayMs(max_delay_ms);
 }
 
 Stats<double> ChokeFilter::GetDelayStats() const {
@@ -592,8 +636,8 @@ void TraceBasedDeliveryFilter::RunFor(int64_t time_ms, Packets* in_out) {
   kbps_stats_.Push(rate_counter_->bits_per_second() / 1000.0);
 }
 
-void TraceBasedDeliveryFilter::SetMaxDelay(int max_delay_ms) {
-  delay_cap_helper_->SetMaxDelay(max_delay_ms);
+void TraceBasedDeliveryFilter::SetMaxDelayMs(int64_t max_delay_ms) {
+  delay_cap_helper_->SetMaxDelayMs(max_delay_ms);
 }
 
 Stats<double> TraceBasedDeliveryFilter::GetDelayStats() const {
@@ -631,10 +675,12 @@ VideoSource::VideoSource(int flow_id,
       frame_period_ms_(1000.0 / fps),
       bits_per_second_(1000 * kbps),
       frame_size_bytes_(bits_per_second_ / 8 / fps),
+      running_(true),
       flow_id_(flow_id),
       next_frame_ms_(first_frame_offset_ms),
       now_ms_(0),
-      prototype_header_() {
+      prototype_header_(),
+      start_plotting_ms_(first_frame_offset_ms) {
   memset(&prototype_header_, 0, sizeof(prototype_header_));
   prototype_header_.ssrc = ssrc;
   prototype_header_.sequenceNumber = 0xf000u;
@@ -651,11 +697,10 @@ uint32_t VideoSource::NextPacketSize(uint32_t frame_size,
 
 void VideoSource::RunFor(int64_t time_ms, Packets* in_out) {
   assert(in_out);
-  std::stringstream ss;
-  ss << "SendEstimate_" << flow_id_ << "#1";
-  BWE_TEST_LOGGING_PLOT(0, ss.str(), now_ms_, bits_per_second_ / 1000);
+
   now_ms_ += time_ms;
   Packets new_packets;
+
   while (now_ms_ >= next_frame_ms_) {
     prototype_header_.timestamp = kTimestampBase +
         static_cast<uint32_t>(next_frame_ms_ * 90.0);
@@ -665,26 +710,61 @@ void VideoSource::RunFor(int64_t time_ms, Packets* in_out) {
     // but the payload size is capped, so if the whole frame doesn't fit in
     // one packet, we will see a number of equally sized packets followed by
     // one smaller at the tail.
+
     int64_t send_time_us = next_frame_ms_ * 1000.0;
-    uint32_t frame_size = NextFrameSize();
-    uint32_t payload_size = frame_size;
 
-    while (payload_size > 0) {
+    if (running_) {
+      uint32_t frame_size = NextFrameSize();
+      uint32_t payload_size = frame_size;
+
+      while (payload_size > 0) {
+        ++prototype_header_.sequenceNumber;
+        uint32_t size = NextPacketSize(frame_size, payload_size);
+        MediaPacket* new_packet =
+            new MediaPacket(flow_id_, send_time_us, size, prototype_header_);
+        new_packets.push_back(new_packet);
+        new_packet->SetAbsSendTimeMs(next_frame_ms_);
+        new_packet->set_sender_timestamp_us(send_time_us);
+        // dynamics_->set_sending_estimate_kbps(bits_per_second_ / 1000);

[stefan-webrtc, 2015/07/02 11:03:41]

Can this and line 740 be removed?

[magalhaesc, 2015/07/02 17:06:18]

Yes, removed

+        payload_size -= size;
+      }
+    } else {
+      int64_t send_time_us = next_frame_ms_ * 1000.0;
       ++prototype_header_.sequenceNumber;
-      uint32_t size = NextPacketSize(frame_size, payload_size);
+      uint32_t size = 0;
       MediaPacket* new_packet =
           new MediaPacket(flow_id_, send_time_us, size, prototype_header_);

[stefan-webrtc, 2015/07/02 11:03:41]

Why exactly do we have to create empty packets when not running? :)

[magalhaesc, 2015/07/02 17:06:18]

This was for plotting reasons, for not having a jump line once the flow resumes.

This will be removed and the case will be handled directly in the
MetricRecorder.

       new_packets.push_back(new_packet);
       new_packet->SetAbsSendTimeMs(next_frame_ms_);
       new_packet->set_sender_timestamp_us(send_time_us);
-      payload_size -= size;
+      // dynamics_->set_sending_estimate_kbps(bits_per_second_ / 1000);
     }
 
-    next_frame_ms_ += frame_period_ms_;
+    // A variance picked uniformly from {-1, 0, 1}ms is added to the frame

[stefan-webrtc, 2015/07/02 11:03:41]

Space between } and ms

[magalhaesc, 2015/07/02 17:06:18]

Done.

+    // timestamp.
+    next_frame_ms_ +=
+        frame_period_ms_ - 1 + 2 * static_cast<float>(rand()) / RAND_MAX;
   }
+
   in_out->merge(new_packets, DereferencingComparator<Packet>);
 }
 
+void VideoSource::Pause() {
+  running_ = false;
+}
+
+void VideoSource::Resume() {
+  running_ = true;
+}
+
+void AdaptiveVideoSource::SetBitrateBps(int bitrate_bps) {
+  if (!running_) {
+    bitrate_bps = 0;
+  }
+  bits_per_second_ = std::min(bitrate_bps, 2500000);
+  frame_size_bytes_ = (bits_per_second_ / 8 * frame_period_ms_ + 500) / 1000;
+}
+
 AdaptiveVideoSource::AdaptiveVideoSource(int flow_id,
                                          float fps,
                                          uint32_t kbps,
@@ -693,11 +773,6 @@ AdaptiveVideoSource::AdaptiveVideoSource(int flow_id,
     : VideoSource(flow_id, fps, kbps, ssrc, first_frame_offset_ms) {
 }
 
-void AdaptiveVideoSource::SetBitrateBps(int bitrate_bps) {
-  bits_per_second_ = std::min(bitrate_bps, 2500000);
-  frame_size_bytes_ = (bits_per_second_ / 8 * frame_period_ms_ + 500) / 1000;
-}
-
 PeriodicKeyFrameSource::PeriodicKeyFrameSource(int flow_id,
                                                float fps,
                                                uint32_t kbps,