path: root/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.cc

/*
 *  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 "modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h"

#include <math.h>

#include <algorithm>
#include <memory>
#include <utility>

#include "api/transport/field_trial_based_config.h"
#include "api/units/data_rate.h"
#include "api/units/data_size.h"
#include "api/units/time_delta.h"
#include "api/units/timestamp.h"
#include "modules/remote_bitrate_estimator/include/bwe_defines.h"
#include "modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "rtc_base/thread_annotations.h"
#include "system_wrappers/include/metrics.h"

namespace webrtc {
namespace {

constexpr TimeDelta kMinClusterDelta = TimeDelta::Millis(1);
constexpr TimeDelta kInitialProbingInterval = TimeDelta::Seconds(2);
constexpr int kTimestampGroupLengthMs = 5;
constexpr int kAbsSendTimeInterArrivalUpshift = 8;
constexpr int kInterArrivalShift =
    RTPHeaderExtension::kAbsSendTimeFraction + kAbsSendTimeInterArrivalUpshift;
constexpr int kMinClusterSize = 4;
constexpr int kMaxProbePackets = 15;
constexpr int kExpectedNumberOfProbes = 3;
constexpr double kTimestampToMs =
    1000.0 / static_cast<double>(1 << kInterArrivalShift);

absl::optional<DataRate> OptionalRateFromOptionalBps(
    absl::optional<int> bitrate_bps) {
  if (bitrate_bps) {
    return DataRate::BitsPerSec(*bitrate_bps);
  } else {
    return absl::nullopt;
  }
}

template <typename K, typename V>
std::vector<K> Keys(const std::map<K, V>& map) {
  std::vector<K> keys;
  keys.reserve(map.size());
  for (const auto& kv_pair : map) {
    keys.push_back(kv_pair.first);
  }
  return keys;
}

}  // namespace

RemoteBitrateEstimatorAbsSendTime::~RemoteBitrateEstimatorAbsSendTime() =
    default;

bool RemoteBitrateEstimatorAbsSendTime::IsWithinClusterBounds(
    TimeDelta send_delta,
    const Cluster& cluster_aggregate) {
  if (cluster_aggregate.count == 0)
    return true;
  TimeDelta cluster_mean =
      cluster_aggregate.send_mean / cluster_aggregate.count;
  return (send_delta - cluster_mean).Abs() < TimeDelta::Micros(2'500);
}

void RemoteBitrateEstimatorAbsSendTime::MaybeAddCluster(
    const Cluster& cluster_aggregate,
    std::list<Cluster>& clusters) {
  if (cluster_aggregate.count < kMinClusterSize ||
      cluster_aggregate.send_mean <= TimeDelta::Zero() ||
      cluster_aggregate.recv_mean <= TimeDelta::Zero()) {
    return;
  }

  Cluster cluster;
  cluster.send_mean = cluster_aggregate.send_mean / cluster_aggregate.count;
  cluster.recv_mean = cluster_aggregate.recv_mean / cluster_aggregate.count;
  cluster.mean_size = cluster_aggregate.mean_size / cluster_aggregate.count;
  cluster.count = cluster_aggregate.count;
  cluster.num_above_min_delta = cluster_aggregate.num_above_min_delta;
  clusters.push_back(cluster);
}

RemoteBitrateEstimatorAbsSendTime::RemoteBitrateEstimatorAbsSendTime(
    RemoteBitrateObserver* observer,
    Clock* clock)
    : clock_(clock),
      observer_(observer),
      detector_(&field_trials_),
      remote_rate_(&field_trials_) {
  RTC_DCHECK(clock_);
  RTC_DCHECK(observer_);
  RTC_LOG(LS_INFO) << "RemoteBitrateEstimatorAbsSendTime: Instantiating.";
}

std::list<RemoteBitrateEstimatorAbsSendTime::Cluster>
RemoteBitrateEstimatorAbsSendTime::ComputeClusters() const {
  std::list<Cluster> clusters;
  Cluster cluster_aggregate;
  Timestamp prev_send_time = Timestamp::MinusInfinity();
  Timestamp prev_recv_time = Timestamp::MinusInfinity();
  for (const Probe& probe : probes_) {
    if (prev_send_time.IsFinite()) {
      TimeDelta send_delta = probe.send_time - prev_send_time;
      TimeDelta recv_delta = probe.recv_time - prev_recv_time;
      if (send_delta >= kMinClusterDelta && recv_delta >= kMinClusterDelta) {
        ++cluster_aggregate.num_above_min_delta;
      }
      if (!IsWithinClusterBounds(send_delta, cluster_aggregate)) {
        MaybeAddCluster(cluster_aggregate, clusters);
        cluster_aggregate = Cluster();
      }
      cluster_aggregate.send_mean += send_delta;
      cluster_aggregate.recv_mean += recv_delta;
      cluster_aggregate.mean_size += probe.payload_size;
      ++cluster_aggregate.count;
    }
    prev_send_time = probe.send_time;
    prev_recv_time = probe.recv_time;
  }
  MaybeAddCluster(cluster_aggregate, clusters);
  return clusters;
}

const RemoteBitrateEstimatorAbsSendTime::Cluster*
RemoteBitrateEstimatorAbsSendTime::FindBestProbe(
    const std::list<Cluster>& clusters) const {
  DataRate highest_probe_bitrate = DataRate::Zero();
  const Cluster* best = nullptr;
  for (const auto& cluster : clusters) {
    if (cluster.send_mean == TimeDelta::Zero() ||
        cluster.recv_mean == TimeDelta::Zero()) {
      continue;
    }
    if (cluster.num_above_min_delta > cluster.count / 2 &&
        (cluster.recv_mean - cluster.send_mean <= TimeDelta::Millis(2) &&
         cluster.send_mean - cluster.recv_mean <= TimeDelta::Millis(5))) {
      DataRate probe_bitrate =
          std::min(cluster.SendBitrate(), cluster.RecvBitrate());
      if (probe_bitrate > highest_probe_bitrate) {
        highest_probe_bitrate = probe_bitrate;
        best = &cluster;
      }
    } else {
      RTC_LOG(LS_INFO) << "Probe failed, sent at "
                       << cluster.SendBitrate().bps() << " bps, received at "
                       << cluster.RecvBitrate().bps()
                       << " bps. Mean send delta: " << cluster.send_mean.ms()
                       << " ms, mean recv delta: " << cluster.recv_mean.ms()
                       << " ms, num probes: " << cluster.count;
      break;
    }
  }
  return best;
}

RemoteBitrateEstimatorAbsSendTime::ProbeResult
RemoteBitrateEstimatorAbsSendTime::ProcessClusters(Timestamp now) {
  std::list<Cluster> clusters = ComputeClusters();
  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;
  }

  if (const Cluster* best = FindBestProbe(clusters)) {
    DataRate probe_bitrate = std::min(best->SendBitrate(), best->RecvBitrate());
    // Make sure that a probe sent on a lower bitrate than our estimate can't
    // reduce the estimate.
    if (IsBitrateImproving(probe_bitrate)) {
      RTC_LOG(LS_INFO) << "Probe successful, sent at "
                       << best->SendBitrate().bps() << " bps, received at "
                       << best->RecvBitrate().bps()
                       << " bps. Mean send delta: " << best->send_mean.ms()
                       << " ms, mean recv delta: " << best->recv_mean.ms()
                       << " ms, num probes: " << best->count;
      remote_rate_.SetEstimate(probe_bitrate, now);
      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(
    DataRate probe_bitrate) const {
  bool initial_probe =
      !remote_rate_.ValidEstimate() && probe_bitrate > DataRate::Zero();
  bool bitrate_above_estimate = remote_rate_.ValidEstimate() &&
                                probe_bitrate > remote_rate_.LatestEstimate();
  return initial_probe || bitrate_above_estimate;
}

void RemoteBitrateEstimatorAbsSendTime::IncomingPacket(
    int64_t arrival_time_ms,
    size_t payload_size,
    const RTPHeader& header) {
  RTC_DCHECK_RUNS_SERIALIZED(&network_race_);
  if (!header.extension.hasAbsoluteSendTime) {
    RTC_LOG(LS_WARNING)
        << "RemoteBitrateEstimatorAbsSendTimeImpl: Incoming packet "
           "is missing absolute send time extension!";
    return;
  }
  IncomingPacketInfo(Timestamp::Millis(arrival_time_ms),
                     header.extension.absoluteSendTime,
                     DataSize::Bytes(payload_size), header.ssrc);
}

void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo(
    Timestamp arrival_time,
    uint32_t send_time_24bits,
    DataSize 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;
  Timestamp send_time =
      Timestamp::Millis(static_cast<int64_t>(timestamp) * kTimestampToMs);

  Timestamp now = clock_->CurrentTime();
  // 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.
  absl::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.bytes(), arrival_time.ms());

  if (first_packet_time_.IsInfinite()) {
    first_packet_time_ = now;
  }

  uint32_t ts_delta = 0;
  int64_t t_delta = 0;
  int size_delta = 0;
  bool update_estimate = false;
  DataRate target_bitrate = DataRate::Zero();
  std::vector<uint32_t> ssrcs;
  {
    MutexLock lock(&mutex_);

    TimeoutStreams(now);
    RTC_DCHECK(inter_arrival_);
    RTC_DCHECK(estimator_);
    // TODO(danilchap): Replace 5 lines below with insert_or_assign when that
    // c++17 function is available.
    auto inserted = ssrcs_.insert(std::make_pair(ssrc, now));
    if (!inserted.second) {
      // Already inserted, update.
      inserted.first->second = now;
    }

    // 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.
    static constexpr DataSize kMinProbePacketSize = DataSize::Bytes(200);
    if (payload_size > kMinProbePacketSize &&
        (!remote_rate_.ValidEstimate() ||
         now - first_packet_time_ < kInitialProbingInterval)) {
      // TODO(holmer): Use a map instead to get correct order?
      if (total_probes_received_ < kMaxProbePackets) {
        TimeDelta send_delta = TimeDelta::Millis(-1);
        TimeDelta recv_delta = TimeDelta::Millis(-1);
        if (!probes_.empty()) {
          send_delta = send_time - probes_.back().send_time;
          recv_delta = arrival_time - probes_.back().recv_time;
        }
        RTC_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_.emplace_back(send_time, arrival_time, 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) == ProbeResult::kBitrateUpdated)
        update_estimate = true;
    }
    if (inter_arrival_->ComputeDeltas(timestamp, arrival_time.ms(), now.ms(),
                                      payload_size.bytes(), &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_.IsInfinite() ||
          now.ms() - last_update_.ms() >
              remote_rate_.GetFeedbackInterval().ms()) {
        update_estimate = true;
      } else if (detector_.State() == BandwidthUsage::kBwOverusing) {
        absl::optional<uint32_t> incoming_rate =
            incoming_bitrate_.Rate(arrival_time.ms());
        if (incoming_rate && remote_rate_.TimeToReduceFurther(
                                 now, DataRate::BitsPerSec(*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(), OptionalRateFromOptionalBps(
                                 incoming_bitrate_.Rate(arrival_time.ms())));
      target_bitrate = remote_rate_.Update(&input, now);
      update_estimate = remote_rate_.ValidEstimate();
      ssrcs = Keys(ssrcs_);
    }
  }
  if (update_estimate) {
    last_update_ = now;
    observer_->OnReceiveBitrateChanged(ssrcs, target_bitrate.bps<uint32_t>());
  }
}

void RemoteBitrateEstimatorAbsSendTime::Process() {}

int64_t RemoteBitrateEstimatorAbsSendTime::TimeUntilNextProcess() {
  const int64_t kDisabledModuleTime = 1000;
  return kDisabledModuleTime;
}

void RemoteBitrateEstimatorAbsSendTime::TimeoutStreams(Timestamp now) {
  for (auto it = ssrcs_.begin(); it != ssrcs_.end();) {
    if (now - it->second > TimeDelta::Millis(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_ = std::make_unique<InterArrival>(
        (kTimestampGroupLengthMs << kInterArrivalShift) / 1000, kTimestampToMs,
        true);
    estimator_ = std::make_unique<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*/) {
  MutexLock lock(&mutex_);
  remote_rate_.SetRtt(TimeDelta::Millis(avg_rtt_ms));
}

void RemoteBitrateEstimatorAbsSendTime::RemoveStream(uint32_t ssrc) {
  MutexLock lock(&mutex_);
  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);
  MutexLock lock(&mutex_);
  if (!remote_rate_.ValidEstimate()) {
    return false;
  }
  *ssrcs = Keys(ssrcs_);
  if (ssrcs_.empty()) {
    *bitrate_bps = 0;
  } else {
    *bitrate_bps = remote_rate_.LatestEstimate().bps<uint32_t>();
  }
  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.
  MutexLock lock(&mutex_);
  remote_rate_.SetMinBitrate(DataRate::BitsPerSec(min_bitrate_bps));
}
}  // namespace webrtc