Use |probe_cluster_id| to cluster packets.

webrtc/modules/congestion_controller/delay_based_bwe.cc

 /*
- *  Copyright (c) 2013 The WebRTC project authors. All Rights Reserved.
+ *  Copyright (c) 2016 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
-#include "webrtc/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h"
+#include "webrtc/modules/congestion_controller/delay_based_bwe.h"
 
 #include <math.h>
 
 #include <algorithm>
 
 #include "webrtc/base/checks.h"
 #include "webrtc/base/constructormagic.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/modules/pacing/paced_sender.h"
 #include "webrtc/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h"
 #include "webrtc/system_wrappers/include/critical_section_wrapper.h"
 #include "webrtc/typedefs.h"
 
-namespace webrtc {
-
+namespace {
 enum {
   kTimestampGroupLengthMs = 5,
   kAbsSendTimeFraction = 18,
   kAbsSendTimeInterArrivalUpshift = 8,
   kInterArrivalShift = kAbsSendTimeFraction + kAbsSendTimeInterArrivalUpshift,
   kInitialProbingIntervalMs = 2000,
   kMinClusterSize = 4,
   kMaxProbePackets = 15,
   kExpectedNumberOfProbes = 3
 };
 
-static const double kTimestampToMs = 1000.0 /
-    static_cast<double>(1 << kInterArrivalShift);
+static const double kTimestampToMs =
+    1000.0 / static_cast<double>(1 << kInterArrivalShift);
 
-template<typename K, typename V>
+template <typename K, typename V>
 std::vector<K> Keys(const std::map<K, V>& map) {
   std::vector<K> keys;
   keys.reserve(map.size());
   for (typename std::map<K, V>::const_iterator it = map.begin();
-      it != map.end(); ++it) {
+       it != map.end(); ++it) {
     keys.push_back(it->first);
   }
   return keys;
 }
 
 uint32_t ConvertMsTo24Bits(int64_t time_ms) {
   uint32_t time_24_bits =
       static_cast<uint32_t>(
           ((static_cast<uint64_t>(time_ms) << kAbsSendTimeFraction) + 500) /
           1000) &
       0x00FFFFFF;
   return time_24_bits;
 }
+}  // namespace
 
-bool RemoteBitrateEstimatorAbsSendTime::IsWithinClusterBounds(
-    int send_delta_ms,
-    const Cluster& cluster_aggregate) {
-    if (cluster_aggregate.count == 0)
-      return true;
-    float cluster_mean = cluster_aggregate.send_mean_ms /
-                         static_cast<float>(cluster_aggregate.count);
-    return fabs(static_cast<float>(send_delta_ms) - cluster_mean) < 2.5f;
-  }
+namespace webrtc {
 
-  void RemoteBitrateEstimatorAbsSendTime::AddCluster(
-      std::list<Cluster>* clusters,
-      Cluster* cluster) {
-    cluster->send_mean_ms /= static_cast<float>(cluster->count);
-    cluster->recv_mean_ms /= static_cast<float>(cluster->count);
-    cluster->mean_size /= cluster->count;
-    clusters->push_back(*cluster);
-  }
-
-  RemoteBitrateEstimatorAbsSendTime::RemoteBitrateEstimatorAbsSendTime(
-      RemoteBitrateObserver* observer)
-      : observer_(observer),
-        inter_arrival_(),
-        estimator_(),
-        detector_(OverUseDetectorOptions()),
-        incoming_bitrate_(kBitrateWindowMs, 8000),
-        incoming_bitrate_initialized_(false),
-        total_probes_received_(0),
-        first_packet_time_ms_(-1),
-        last_update_ms_(-1),
-        ssrcs_() {
-    RTC_DCHECK(observer_);
-    LOG(LS_INFO) << "RemoteBitrateEstimatorAbsSendTime: Instantiating.";
-    network_thread_.DetachFromThread();
+void DelayBasedBwe::AddCluster(std::list<Cluster>* clusters, Cluster* cluster) {
+  cluster->send_mean_ms /= static_cast<float>(cluster->count);
+  cluster->recv_mean_ms /= static_cast<float>(cluster->count);
+  cluster->mean_size /= cluster->count;
+  clusters->push_back(*cluster);
 }
 
-void RemoteBitrateEstimatorAbsSendTime::ComputeClusters(
-    std::list<Cluster>* clusters) const {
+DelayBasedBwe::DelayBasedBwe(RemoteBitrateObserver* observer)
+    : observer_(observer),
+      inter_arrival_(),
+      estimator_(),
+      detector_(OverUseDetectorOptions()),
+      incoming_bitrate_(kBitrateWindowMs, 8000),
+      total_probes_received_(0),
+      first_packet_time_ms_(-1),
+      last_update_ms_(-1),
+      ssrcs_() {
+  RTC_DCHECK(observer_);
+  // NOTE! The BitrateEstimatorTest relies on this EXACT log line.
+  LOG(LS_INFO) << "RemoteBitrateEstimatorAbsSendTime: Instantiating.";
+  network_thread_.DetachFromThread();
+}
+
+void DelayBasedBwe::ComputeClusters(std::list<Cluster>* clusters) const {
   Cluster current;
   int64_t prev_send_time = -1;
   int64_t prev_recv_time = -1;
+  int last_probe_cluster_id = -1;
   for (std::list<Probe>::const_iterator it = probes_.begin();
-       it != probes_.end();
-       ++it) {
+       it != probes_.end(); ++it) {
+    if (last_probe_cluster_id == -1)
+      last_probe_cluster_id = it->cluster_id;
     if (prev_send_time >= 0) {
       int send_delta_ms = it->send_time_ms - prev_send_time;
       int recv_delta_ms = it->recv_time_ms - prev_recv_time;
       if (send_delta_ms >= 1 && recv_delta_ms >= 1) {
         ++current.num_above_min_delta;
       }
-      if (!IsWithinClusterBounds(send_delta_ms, current)) {
+      if (it->cluster_id != last_probe_cluster_id) {
         if (current.count >= kMinClusterSize)
           AddCluster(clusters, &current);
         current = Cluster();
       }
       current.send_mean_ms += send_delta_ms;
       current.recv_mean_ms += recv_delta_ms;
       current.mean_size += it->payload_size;
       ++current.count;
+      last_probe_cluster_id = it->cluster_id;
     }
     prev_send_time = it->send_time_ms;
     prev_recv_time = it->recv_time_ms;
   }
   if (current.count >= kMinClusterSize)
     AddCluster(clusters, &current);
 }
 
-std::list<Cluster>::const_iterator
-RemoteBitrateEstimatorAbsSendTime::FindBestProbe(
+std::list<DelayBasedBwe::Cluster>::const_iterator DelayBasedBwe::FindBestProbe(
     const std::list<Cluster>& clusters) const {
   int highest_probe_bitrate_bps = 0;
   std::list<Cluster>::const_iterator best_it = clusters.end();
   for (std::list<Cluster>::const_iterator it = clusters.begin();
-       it != clusters.end();
-       ++it) {
+       it != clusters.end(); ++it) {
     if (it->send_mean_ms == 0 || it->recv_mean_ms == 0)
       continue;
     int send_bitrate_bps = it->mean_size * 8 * 1000 / it->send_mean_ms;
     int recv_bitrate_bps = it->mean_size * 8 * 1000 / it->recv_mean_ms;
     if (it->num_above_min_delta > it->count / 2 &&
         (it->recv_mean_ms - it->send_mean_ms <= 2.0f &&
          it->send_mean_ms - it->recv_mean_ms <= 5.0f)) {
       int probe_bitrate_bps =
           std::min(it->GetSendBitrateBps(), it->GetRecvBitrateBps());
       if (probe_bitrate_bps > highest_probe_bitrate_bps) {
         highest_probe_bitrate_bps = probe_bitrate_bps;
         best_it = it;
       }
     } else {
       LOG(LS_INFO) << "Probe failed, sent at " << send_bitrate_bps
                    << " bps, received at " << recv_bitrate_bps
                    << " bps. Mean send delta: " << it->send_mean_ms
                    << " ms, mean recv delta: " << it->recv_mean_ms
                    << " ms, num probes: " << it->count;
       break;
     }
   }
   return best_it;
 }
 
-RemoteBitrateEstimatorAbsSendTime::ProbeResult
-RemoteBitrateEstimatorAbsSendTime::ProcessClusters(int64_t now_ms) {
+DelayBasedBwe::ProbeResult DelayBasedBwe::ProcessClusters(int64_t now_ms) {
   std::list<Cluster> clusters;
   ComputeClusters(&clusters);
   if (clusters.empty()) {
     // If we reach the max number of probe packets and still have no clusters,
     // we will remove the oldest one.
     if (probes_.size() >= kMaxProbePackets)
       probes_.pop_front();
     return ProbeResult::kNoUpdate;
   }
 
@@ skipping 15 matching lines @@
     }
   }
 
   // Not probing and received non-probe packet, or finished with current set
   // of probes.
   if (clusters.size() >= kExpectedNumberOfProbes)
     probes_.clear();
   return ProbeResult::kNoUpdate;
 }
 
-bool RemoteBitrateEstimatorAbsSendTime::IsBitrateImproving(
-    int new_bitrate_bps) const {
+bool DelayBasedBwe::IsBitrateImproving(int new_bitrate_bps) const {
   bool initial_probe = !remote_rate_.ValidEstimate() && new_bitrate_bps > 0;
   bool bitrate_above_estimate =
       remote_rate_.ValidEstimate() &&
       new_bitrate_bps > static_cast<int>(remote_rate_.LatestEstimate());
   return initial_probe || bitrate_above_estimate;
 }
 
-void RemoteBitrateEstimatorAbsSendTime::IncomingPacketFeedbackVector(
+void DelayBasedBwe::IncomingPacketFeedbackVector(
     const std::vector<PacketInfo>& packet_feedback_vector) {
   RTC_DCHECK(network_thread_.CalledOnValidThread());
   for (const auto& packet_info : packet_feedback_vector) {
     IncomingPacketInfo(packet_info.arrival_time_ms,
                        ConvertMsTo24Bits(packet_info.send_time_ms),
-                       packet_info.payload_size, 0, packet_info.was_paced);
+                       packet_info.payload_size, 0, packet_info.was_paced,
+                       packet_info.probe_cluster_id);
   }
 }
 
-void RemoteBitrateEstimatorAbsSendTime::IncomingPacket(int64_t arrival_time_ms,
-                                                       size_t payload_size,
-                                                       const RTPHeader& header,
-                                                       bool was_paced) {
+void DelayBasedBwe::IncomingPacket(int64_t arrival_time_ms,
+                                   size_t payload_size,
+                                   const RTPHeader& header,
+                                   bool was_paced) {
   RTC_DCHECK(network_thread_.CalledOnValidThread());
   if (!header.extension.hasAbsoluteSendTime) {
-    LOG(LS_WARNING) << "RemoteBitrateEstimatorAbsSendTimeImpl: Incoming packet "
+    // NOTE! The BitrateEstimatorTest relies on this EXACT log line.
+    LOG(LS_WARNING) << "RemoteBitrateEstimatorAbsSendTime: Incoming packet "
                        "is missing absolute send time extension!";
     return;
   }
   IncomingPacketInfo(arrival_time_ms, header.extension.absoluteSendTime,
-                     payload_size, header.ssrc, was_paced);
+                     payload_size, header.ssrc, was_paced,
+                     PacketInfo::kNotAProbe);
 }
 
-void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo(
-    int64_t arrival_time_ms,
-    uint32_t send_time_24bits,
-    size_t payload_size,
-    uint32_t ssrc,
-    bool was_paced) {
+void DelayBasedBwe::IncomingPacket(int64_t arrival_time_ms,
+                                   size_t payload_size,
+                                   const RTPHeader& header,
+                                   bool was_paced,
+                                   int probe_cluster_id) {
+  RTC_DCHECK(network_thread_.CalledOnValidThread());
+  if (!header.extension.hasAbsoluteSendTime) {
+    // NOTE! The BitrateEstimatorTest relies on this EXACT log line.
+    LOG(LS_WARNING) << "RemoteBitrateEstimatorAbsSendTime: Incoming packet "
+                       "is missing absolute send time extension!";
+    return;
+  }
+  IncomingPacketInfo(arrival_time_ms, header.extension.absoluteSendTime,
+                     payload_size, header.ssrc, was_paced, probe_cluster_id);
+}
+
+void DelayBasedBwe::IncomingPacketInfo(int64_t arrival_time_ms,
+                                       uint32_t send_time_24bits,
+                                       size_t payload_size,
+                                       uint32_t ssrc,
+                                       bool was_paced,
+                                       int probe_cluster_id) {
   assert(send_time_24bits < (1ul << 24));
   // Shift up send time to use the full 32 bits that inter_arrival works with,
   // so wrapping works properly.
   uint32_t timestamp = send_time_24bits << kAbsSendTimeInterArrivalUpshift;
   int64_t send_time_ms = static_cast<int64_t>(timestamp) * kTimestampToMs;
 
   int64_t now_ms = arrival_time_ms;
   // TODO(holmer): SSRCs are only needed for REMB, should be broken out from
   // here.
-
-  // Check if incoming bitrate estimate is valid, and if it needs to be reset.
-  rtc::Optional<uint32_t> incoming_bitrate = incoming_bitrate_.Rate(now_ms);
-  if (incoming_bitrate) {
-    incoming_bitrate_initialized_ = true;
-  } else if (incoming_bitrate_initialized_) {
-    // Incoming bitrate had a previous valid value, but now not enough data
-    // point are left within the current window. Reset incoming bitrate
-    // estimator so that the window size will only contain new data points.
-    incoming_bitrate_.Reset();
-    incoming_bitrate_initialized_ = false;
-  }
   incoming_bitrate_.Update(payload_size, now_ms);
 
   if (first_packet_time_ms_ == -1)
     first_packet_time_ms_ = arrival_time_ms;
 
   uint32_t ts_delta = 0;
   int64_t t_delta = 0;
   int size_delta = 0;
-  // For now only try to detect probes while we don't have a valid estimate, and
-  // make sure the packet was paced. We currently assume that only packets
-  // larger than 200 bytes are paced by the sender.
-  was_paced = was_paced && payload_size > PacedSender::kMinProbePacketSize;
+
   bool update_estimate = false;
   uint32_t target_bitrate_bps = 0;
   std::vector<uint32_t> ssrcs;
   {
     rtc::CritScope lock(&crit_);
 
     TimeoutStreams(now_ms);
     RTC_DCHECK(inter_arrival_.get());
     RTC_DCHECK(estimator_.get());
     ssrcs_[ssrc] = now_ms;
 
-    if (was_paced &&
+    // For now only try to detect probes while we don't have a valid estimate,
+    // and make sure the packet was paced. We currently assume that only packets
+    // larger than 200 bytes are paced by the sender.
+    if (probe_cluster_id != PacketInfo::kNotAProbe &&
+        payload_size > PacedSender::kMinProbePacketSize &&
         (!remote_rate_.ValidEstimate() ||
          now_ms - first_packet_time_ms_ < kInitialProbingIntervalMs)) {
       // TODO(holmer): Use a map instead to get correct order?
       if (total_probes_received_ < kMaxProbePackets) {
         int send_delta_ms = -1;
         int recv_delta_ms = -1;
         if (!probes_.empty()) {
           send_delta_ms = send_time_ms - probes_.back().send_time_ms;
           recv_delta_ms = arrival_time_ms - probes_.back().recv_time_ms;
         }
         LOG(LS_INFO) << "Probe packet received: send time=" << send_time_ms
                      << " ms, recv time=" << arrival_time_ms
                      << " ms, send delta=" << send_delta_ms
                      << " ms, recv delta=" << recv_delta_ms << " ms.";
       }
-      probes_.push_back(Probe(send_time_ms, arrival_time_ms, payload_size));
+      probes_.push_back(
+          Probe(send_time_ms, arrival_time_ms, payload_size, probe_cluster_id));
       ++total_probes_received_;
       // Make sure that a probe which updated the bitrate immediately has an
       // effect by calling the OnReceiveBitrateChanged callback.
       if (ProcessClusters(now_ms) == ProbeResult::kBitrateUpdated)
         update_estimate = true;
     }
     if (inter_arrival_->ComputeDeltas(timestamp, arrival_time_ms, payload_size,
                                       &ts_delta, &t_delta, &size_delta)) {
       double ts_delta_ms = (1000.0 * ts_delta) / (1 << kInterArrivalShift);
       estimator_->Update(t_delta, ts_delta_ms, size_delta, detector_.State());
@@ skipping 28 matching lines @@
       update_estimate = remote_rate_.ValidEstimate();
       ssrcs = Keys(ssrcs_);
     }
   }
   if (update_estimate) {
     last_update_ms_ = now_ms;
     observer_->OnReceiveBitrateChanged(ssrcs, target_bitrate_bps);
   }
 }
 
-void RemoteBitrateEstimatorAbsSendTime::Process() {}
+void DelayBasedBwe::Process() {}
 
-int64_t RemoteBitrateEstimatorAbsSendTime::TimeUntilNextProcess() {
+int64_t DelayBasedBwe::TimeUntilNextProcess() {
   const int64_t kDisabledModuleTime = 1000;
   return kDisabledModuleTime;
 }
 
-void RemoteBitrateEstimatorAbsSendTime::TimeoutStreams(int64_t now_ms) {
+void DelayBasedBwe::TimeoutStreams(int64_t now_ms) {
   for (Ssrcs::iterator it = ssrcs_.begin(); it != ssrcs_.end();) {
     if ((now_ms - it->second) > kStreamTimeOutMs) {
       ssrcs_.erase(it++);
     } else {
       ++it;
     }
   }
   if (ssrcs_.empty()) {
     // We can't update the estimate if we don't have any active streams.
     inter_arrival_.reset(
         new InterArrival((kTimestampGroupLengthMs << kInterArrivalShift) / 1000,
                          kTimestampToMs, true));
     estimator_.reset(new OveruseEstimator(OverUseDetectorOptions()));
     // We deliberately don't reset the first_packet_time_ms_ here for now since
     // we only probe for bandwidth in the beginning of a call right now.
   }
 }
 
-void RemoteBitrateEstimatorAbsSendTime::OnRttUpdate(int64_t avg_rtt_ms,
-                                                    int64_t max_rtt_ms) {
+void DelayBasedBwe::OnRttUpdate(int64_t avg_rtt_ms, int64_t max_rtt_ms) {
   rtc::CritScope lock(&crit_);
   remote_rate_.SetRtt(avg_rtt_ms);
 }
 
-void RemoteBitrateEstimatorAbsSendTime::RemoveStream(uint32_t ssrc) {
+void DelayBasedBwe::RemoveStream(uint32_t ssrc) {
   rtc::CritScope lock(&crit_);
   ssrcs_.erase(ssrc);
 }
 
-bool RemoteBitrateEstimatorAbsSendTime::LatestEstimate(
-    std::vector<uint32_t>* ssrcs,
-    uint32_t* bitrate_bps) const {
+bool DelayBasedBwe::LatestEstimate(std::vector<uint32_t>* ssrcs,
+                                   uint32_t* bitrate_bps) const {
   // Currently accessed from both the process thread (see
   // ModuleRtpRtcpImpl::Process()) and the configuration thread (see
   // Call::GetStats()). Should in the future only be accessed from a single
   // thread.
   RTC_DCHECK(ssrcs);
   RTC_DCHECK(bitrate_bps);
   rtc::CritScope lock(&crit_);
   if (!remote_rate_.ValidEstimate()) {
     return false;
   }
   *ssrcs = Keys(ssrcs_);
   if (ssrcs_.empty()) {
     *bitrate_bps = 0;
   } else {
     *bitrate_bps = remote_rate_.LatestEstimate();
   }
   return true;
 }
 
-void RemoteBitrateEstimatorAbsSendTime::SetMinBitrate(int min_bitrate_bps) {
+void DelayBasedBwe::SetMinBitrate(int min_bitrate_bps) {
   // Called from both the configuration thread and the network thread. Shouldn't
   // be called from the network thread in the future.
   rtc::CritScope lock(&crit_);
   remote_rate_.SetMinBitrate(min_bitrate_bps);
 }
 }  // namespace webrtc