text: /*
   *  Copyright (c) 2013 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 <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/system_wrappers/include/metrics.h"
  #include "webrtc/typedefs.h"
  
  namespace webrtc {
  
  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);
  
  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) {
      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;
  }
  
  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;
    }
  
    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,
        Clock* clock)
        : clock_(clock),
          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),
          uma_recorded_(false) {
      RTC_DCHECK(observer_);
      LOG(LS_INFO) << "RemoteBitrateEstimatorAbsSendTime: Instantiating.";
      network_thread_.DetachFromThread();
  }
  
  void RemoteBitrateEstimatorAbsSendTime::ComputeClusters(
      std::list<Cluster>* clusters) const {
    Cluster current;
    int64_t prev_send_time = -1;
    int64_t prev_recv_time = -1;
    for (std::list<Probe>::const_iterator it = probes_.begin();
         it != probes_.end();
         ++it) {
      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 (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;
      }
      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(
      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) {
      if (it->send_mean_ms == 0 || it->recv_mean_ms == 0)
        continue;
      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 {
        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;
        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) {
    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;
    }
  
    std::list<Cluster>::const_iterator best_it = FindBestProbe(clusters);
    if (best_it != clusters.end()) {
      int probe_bitrate_bps =
          std::min(best_it->GetSendBitrateBps(), best_it->GetRecvBitrateBps());
      // Make sure that a probe sent on a lower bitrate than our estimate can't
      // reduce the estimate.
      if (IsBitrateImproving(probe_bitrate_bps)) {
        LOG(LS_INFO) << "Probe successful, sent at "
                     << best_it->GetSendBitrateBps() << " bps, received at "
                     << best_it->GetRecvBitrateBps()
                     << " bps. Mean send delta: " << best_it->send_mean_ms
                     << " ms, mean recv delta: " << best_it->recv_mean_ms
                     << " ms, num probes: " << best_it->count;
        remote_rate_.SetEstimate(probe_bitrate_bps, now_ms);
        return ProbeResult::kBitrateUpdated;
      }
    }
  
    // 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 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(
      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);
    }
  }
  
  void RemoteBitrateEstimatorAbsSendTime::IncomingPacket(
      int64_t arrival_time_ms,
      size_t payload_size,
      const RTPHeader& header) {
    RTC_DCHECK(network_thread_.CalledOnValidThread());
    if (!header.extension.hasAbsoluteSendTime) {
      LOG(LS_WARNING) << "RemoteBitrateEstimatorAbsSendTimeImpl: Incoming packet "
                         "is missing absolute send time extension!";
      return;
    }
    IncomingPacketInfo(arrival_time_ms, header.extension.absoluteSendTime,
                       payload_size, header.ssrc);
  }
  
  void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo(
      int64_t arrival_time_ms,
      uint32_t send_time_24bits,
      size_t payload_size,
      uint32_t ssrc) {
    RTC_CHECK(send_time_24bits < (1ul << 24));
    if (!uma_recorded_) {
      RTC_HISTOGRAM_ENUMERATION(kBweTypeHistogram, BweNames::kReceiverAbsSendTime,
                                BweNames::kBweNamesMax);
      uma_recorded_ = true;
    }
    // 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 = clock_->TimeInMilliseconds();
    // 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(arrival_time_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, arrival_time_ms);
  
    if (first_packet_time_ms_ == -1)
      first_packet_time_ms_ = now_ms;
  
    uint32_t ts_delta = 0;
    int64_t t_delta = 0;
    int size_delta = 0;
    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;
  
      // For now only try to detect probes while we don't have a valid estimate.
      // We currently assume that only packets larger than 200 bytes are paced by
      // the sender.
      if (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));
        ++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, now_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(),
                           arrival_time_ms);
        detector_.Detect(estimator_->offset(), ts_delta_ms,
                         estimator_->num_of_deltas(), arrival_time_ms);
      }
  
      if (!update_estimate) {
        // Check if it's time for a periodic update or if we should update because
        // of an over-use.
        if (last_update_ms_ == -1 ||
            now_ms - last_update_ms_ > remote_rate_.GetFeedbackInterval()) {
          update_estimate = true;
        } else if (detector_.State() == kBwOverusing) {
          rtc::Optional<uint32_t> incoming_rate =
              incoming_bitrate_.Rate(arrival_time_ms);
          if (incoming_rate &&
              remote_rate_.TimeToReduceFurther(now_ms, *incoming_rate)) {
            update_estimate = true;
          }
        }
      }
  
      if (update_estimate) {
        // The first overuse should immediately trigger a new estimate.
        // We also have to update the estimate immediately if we are overusing
        // and the target bitrate is too high compared to what we are receiving.
        const RateControlInput input(detector_.State(),
                                     incoming_bitrate_.Rate(arrival_time_ms),
                                     estimator_->var_noise());
        remote_rate_.Update(&input, now_ms);
        target_bitrate_bps = remote_rate_.UpdateBandwidthEstimate(now_ms);
        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() {}
  
  int64_t RemoteBitrateEstimatorAbsSendTime::TimeUntilNextProcess() {
    const int64_t kDisabledModuleTime = 1000;
    return kDisabledModuleTime;
  }
  
  void RemoteBitrateEstimatorAbsSendTime::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) {
    rtc::CritScope lock(&crit_);
    remote_rate_.SetRtt(avg_rtt_ms);
  }
  
  void RemoteBitrateEstimatorAbsSendTime::RemoveStream(uint32_t ssrc) {
    rtc::CritScope lock(&crit_);
    ssrcs_.erase(ssrc);
  }
  
  bool RemoteBitrateEstimatorAbsSendTime::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) {
    // 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
  