<commit_before>/*
 *  Copyright (c) 2012 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/congestion_controller/rtp/include/send_side_congestion_controller.h"

#include <algorithm>
#include <functional>
#include <memory>
#include <vector>
#include "api/transport/network_types.h"
#include "modules/congestion_controller/goog_cc/include/goog_cc_factory.h"
#include "modules/remote_bitrate_estimator/include/bwe_defines.h"
#include "rtc_base/bind.h"
#include "rtc_base/checks.h"
#include "rtc_base/format_macros.h"
#include "rtc_base/logging.h"
#include "rtc_base/numerics/safe_conversions.h"
#include "rtc_base/numerics/safe_minmax.h"
#include "rtc_base/ptr_util.h"
#include "rtc_base/rate_limiter.h"
#include "rtc_base/sequenced_task_checker.h"
#include "rtc_base/socket.h"
#include "rtc_base/timeutils.h"
#include "system_wrappers/include/field_trial.h"
#include "system_wrappers/include/runtime_enabled_features.h"

using rtc::MakeUnique;

namespace webrtc {
namespace webrtc_cc {
namespace {
using send_side_cc_internal::PeriodicTask;

// When CongestionWindowPushback is enabled, the pacer is oblivious to
// the congestion window. The relation between outstanding data and
// the congestion window affects encoder allocations directly.
const char kCongestionPushbackExperiment[] = "WebRTC-CongestionWindowPushback";

// When PacerPushbackExperiment is enabled, build-up in the pacer due to
// the congestion window and/or data spikes reduces encoder allocations.
const char kPacerPushbackExperiment[] = "WebRTC-PacerPushbackExperiment";
const int64_t PacerQueueUpdateIntervalMs = 25;
const uint32_t MinPushbackTargetBitrateBps = 30000;

bool IsPacerPushbackExperimentEnabled() {
  return webrtc::field_trial::IsEnabled(kPacerPushbackExperiment) ||
         (!webrtc::field_trial::IsDisabled(kPacerPushbackExperiment) &&
          webrtc::runtime_enabled_features::IsFeatureEnabled(
              webrtc::runtime_enabled_features::kDualStreamModeFeatureName));
}

bool IsCongestionWindowExperimentEnabled() {
  return webrtc::field_trial::IsEnabled(kCongestionPushbackExperiment);
}

void SortPacketFeedbackVector(std::vector<webrtc::PacketFeedback>* input) {
  std::sort(input->begin(), input->end(), PacketFeedbackComparator());
}

PacketResult NetworkPacketFeedbackFromRtpPacketFeedback(
    const webrtc::PacketFeedback& pf) {
  PacketResult feedback;
  if (pf.arrival_time_ms == webrtc::PacketFeedback::kNotReceived)
    feedback.receive_time = Timestamp::Infinity();
  else
    feedback.receive_time = Timestamp::ms(pf.arrival_time_ms);
  if (pf.send_time_ms != webrtc::PacketFeedback::kNoSendTime) {
    feedback.sent_packet = SentPacket();
    feedback.sent_packet->sequence_number = pf.long_sequence_number;
    feedback.sent_packet->send_time = Timestamp::ms(pf.send_time_ms);
    feedback.sent_packet->size = DataSize::bytes(pf.payload_size);
    feedback.sent_packet->pacing_info = pf.pacing_info;
  }
  return feedback;
}

std::vector<PacketResult> PacketResultsFromRtpFeedbackVector(
    const std::vector<PacketFeedback>& feedback_vector) {
  RTC_DCHECK(std::is_sorted(feedback_vector.begin(), feedback_vector.end(),
                            PacketFeedbackComparator()));

  std::vector<PacketResult> packet_feedbacks;
  packet_feedbacks.reserve(feedback_vector.size());
  for (const PacketFeedback& rtp_feedback : feedback_vector) {
    auto feedback = NetworkPacketFeedbackFromRtpPacketFeedback(rtp_feedback);
    packet_feedbacks.push_back(feedback);
  }
  return packet_feedbacks;
}

TargetRateConstraints ConvertConstraints(int min_bitrate_bps,
                                         int max_bitrate_bps,
                                         const Clock* clock) {
  TargetRateConstraints msg;
  msg.at_time = Timestamp::ms(clock->TimeInMilliseconds());
  msg.min_data_rate =
      min_bitrate_bps >= 0 ? DataRate::bps(min_bitrate_bps) : DataRate::Zero();
  msg.max_data_rate = max_bitrate_bps > 0 ? DataRate::bps(max_bitrate_bps)
                                          : DataRate::Infinity();
  return msg;
}

// The template closure pattern is based on rtc::ClosureTask.
template <class Closure>
class PeriodicTaskImpl final : public PeriodicTask {
 public:
  PeriodicTaskImpl(rtc::TaskQueue* task_queue,
                   int64_t period_ms,
                   Closure&& closure)
      : task_queue_(task_queue),
        period_ms_(period_ms),
        closure_(std::forward<Closure>(closure)) {}
  bool Run() override {
    if (!running_)
      return true;
    closure_();
    // WrapUnique lets us repost this task on the TaskQueue.
    task_queue_->PostDelayedTask(rtc::WrapUnique(this), period_ms_);
    // Return false to tell TaskQueue to not destruct this object, since we have
    // taken ownership with WrapUnique.
    return false;
  }
  void Stop() override {
    if (task_queue_->IsCurrent()) {
      RTC_DCHECK(running_);
      running_ = false;
    } else {
      task_queue_->PostTask([this] { Stop(); });
    }
  }

 private:
  rtc::TaskQueue* const task_queue_;
  const int64_t period_ms_;
  typename std::remove_const<
      typename std::remove_reference<Closure>::type>::type closure_;
  bool running_ = true;
};

template <class Closure>
static PeriodicTask* StartPeriodicTask(rtc::TaskQueue* task_queue,
                                       int64_t period_ms,
                                       Closure&& closure) {
  auto periodic_task = rtc::MakeUnique<PeriodicTaskImpl<Closure>>(
      task_queue, period_ms, std::forward<Closure>(closure));
  PeriodicTask* periodic_task_ptr = periodic_task.get();
  task_queue->PostDelayedTask(std::move(periodic_task), period_ms);
  return periodic_task_ptr;
}

}  // namespace

namespace send_side_cc_internal {
class ControlHandler {
 public:
  ControlHandler(NetworkChangedObserver* observer,
                 PacerController* pacer_controller,
                 const Clock* clock);

  void PostUpdates(NetworkControlUpdate update);

  void OnNetworkAvailability(NetworkAvailability msg);
  void OnOutstandingData(DataSize in_flight_data);
  void OnPacerQueueUpdate(TimeDelta expected_queue_time);

  rtc::Optional<TargetTransferRate> last_transfer_rate();

 private:
  void OnNetworkInvalidation();
  bool GetNetworkParameters(int32_t* estimated_bitrate_bps,
                            uint8_t* fraction_loss,
                            int64_t* rtt_ms);
  bool IsSendQueueFull() const;
  bool HasNetworkParametersToReportChanged(int64_t bitrate_bps,
                                           uint8_t fraction_loss,
                                           int64_t rtt);
  NetworkChangedObserver* observer_ = nullptr;
  PacerController* pacer_controller_;

  rtc::Optional<TargetTransferRate> current_target_rate_msg_;
  rtc::Optional<DataSize> congestion_window_;
  DataSize outstanding_data_ = DataSize::Zero();
  bool network_available_ = true;
  int64_t last_reported_target_bitrate_bps_ = 0;
  uint8_t last_reported_fraction_loss_ = 0;
  int64_t last_reported_rtt_ms_ = 0;
  const bool pacer_pushback_experiment_ = false;
  const bool congestion_window_pushback_experiment_ = false;
  int64_t pacer_expected_queue_ms_ = 0;
  double encoding_rate_ratio_ = 1.0;

  rtc::SequencedTaskChecker sequenced_checker_;
  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(ControlHandler);
};

ControlHandler::ControlHandler(NetworkChangedObserver* observer,
                               PacerController* pacer_controller,
                               const Clock* clock)
    : observer_(observer),
      pacer_controller_(pacer_controller),
      pacer_pushback_experiment_(IsPacerPushbackExperimentEnabled()),
      congestion_window_pushback_experiment_(
          IsCongestionWindowExperimentEnabled()) {
  sequenced_checker_.Detach();
}

void ControlHandler::PostUpdates(NetworkControlUpdate update) {
  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequenced_checker_);
  if (congestion_window_pushback_experiment_) {
    if (update.congestion_window) {
      congestion_window_ = update.congestion_window;
      pacer_controller_->OnCongestionWindow(*update.congestion_window);
    }
  }
  if (update.pacer_config) {
    pacer_controller_->OnPacerConfig(*update.pacer_config);
  }
  for (const auto& probe : update.probe_cluster_configs) {<commit_msg>Always apply congestion window in pacer.<commit_after>/*
 *  Copyright (c) 2012 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/congestion_controller/rtp/include/send_side_congestion_controller.h"

#include <algorithm>
#include <functional>
#include <memory>
#include <vector>
#include "api/transport/network_types.h"
#include "modules/congestion_controller/goog_cc/include/goog_cc_factory.h"
#include "modules/remote_bitrate_estimator/include/bwe_defines.h"
#include "rtc_base/bind.h"
#include "rtc_base/checks.h"
#include "rtc_base/format_macros.h"
#include "rtc_base/logging.h"
#include "rtc_base/numerics/safe_conversions.h"
#include "rtc_base/numerics/safe_minmax.h"
#include "rtc_base/ptr_util.h"
#include "rtc_base/rate_limiter.h"
#include "rtc_base/sequenced_task_checker.h"
#include "rtc_base/socket.h"
#include "rtc_base/timeutils.h"
#include "system_wrappers/include/field_trial.h"
#include "system_wrappers/include/runtime_enabled_features.h"

using rtc::MakeUnique;

namespace webrtc {
namespace webrtc_cc {
namespace {
using send_side_cc_internal::PeriodicTask;

// When CongestionWindowPushback is enabled, the pacer is oblivious to
// the congestion window. The relation between outstanding data and
// the congestion window affects encoder allocations directly.
const char kCongestionPushbackExperiment[] = "WebRTC-CongestionWindowPushback";

// When PacerPushbackExperiment is enabled, build-up in the pacer due to
// the congestion window and/or data spikes reduces encoder allocations.
const char kPacerPushbackExperiment[] = "WebRTC-PacerPushbackExperiment";
const int64_t PacerQueueUpdateIntervalMs = 25;
const uint32_t MinPushbackTargetBitrateBps = 30000;

bool IsPacerPushbackExperimentEnabled() {
  return webrtc::field_trial::IsEnabled(kPacerPushbackExperiment) ||
         (!webrtc::field_trial::IsDisabled(kPacerPushbackExperiment) &&
          webrtc::runtime_enabled_features::IsFeatureEnabled(
              webrtc::runtime_enabled_features::kDualStreamModeFeatureName));
}

bool IsCongestionWindowExperimentEnabled() {
  return webrtc::field_trial::IsEnabled(kCongestionPushbackExperiment);
}

void SortPacketFeedbackVector(std::vector<webrtc::PacketFeedback>* input) {
  std::sort(input->begin(), input->end(), PacketFeedbackComparator());
}

PacketResult NetworkPacketFeedbackFromRtpPacketFeedback(
    const webrtc::PacketFeedback& pf) {
  PacketResult feedback;
  if (pf.arrival_time_ms == webrtc::PacketFeedback::kNotReceived)
    feedback.receive_time = Timestamp::Infinity();
  else
    feedback.receive_time = Timestamp::ms(pf.arrival_time_ms);
  if (pf.send_time_ms != webrtc::PacketFeedback::kNoSendTime) {
    feedback.sent_packet = SentPacket();
    feedback.sent_packet->sequence_number = pf.long_sequence_number;
    feedback.sent_packet->send_time = Timestamp::ms(pf.send_time_ms);
    feedback.sent_packet->size = DataSize::bytes(pf.payload_size);
    feedback.sent_packet->pacing_info = pf.pacing_info;
  }
  return feedback;
}

std::vector<PacketResult> PacketResultsFromRtpFeedbackVector(
    const std::vector<PacketFeedback>& feedback_vector) {
  RTC_DCHECK(std::is_sorted(feedback_vector.begin(), feedback_vector.end(),
                            PacketFeedbackComparator()));

  std::vector<PacketResult> packet_feedbacks;
  packet_feedbacks.reserve(feedback_vector.size());
  for (const PacketFeedback& rtp_feedback : feedback_vector) {
    auto feedback = NetworkPacketFeedbackFromRtpPacketFeedback(rtp_feedback);
    packet_feedbacks.push_back(feedback);
  }
  return packet_feedbacks;
}

TargetRateConstraints ConvertConstraints(int min_bitrate_bps,
                                         int max_bitrate_bps,
                                         const Clock* clock) {
  TargetRateConstraints msg;
  msg.at_time = Timestamp::ms(clock->TimeInMilliseconds());
  msg.min_data_rate =
      min_bitrate_bps >= 0 ? DataRate::bps(min_bitrate_bps) : DataRate::Zero();
  msg.max_data_rate = max_bitrate_bps > 0 ? DataRate::bps(max_bitrate_bps)
                                          : DataRate::Infinity();
  return msg;
}

// The template closure pattern is based on rtc::ClosureTask.
template <class Closure>
class PeriodicTaskImpl final : public PeriodicTask {
 public:
  PeriodicTaskImpl(rtc::TaskQueue* task_queue,
                   int64_t period_ms,
                   Closure&& closure)
      : task_queue_(task_queue),
        period_ms_(period_ms),
        closure_(std::forward<Closure>(closure)) {}
  bool Run() override {
    if (!running_)
      return true;
    closure_();
    // WrapUnique lets us repost this task on the TaskQueue.
    task_queue_->PostDelayedTask(rtc::WrapUnique(this), period_ms_);
    // Return false to tell TaskQueue to not destruct this object, since we have
    // taken ownership with WrapUnique.
    return false;
  }
  void Stop() override {
    if (task_queue_->IsCurrent()) {
      RTC_DCHECK(running_);
      running_ = false;
    } else {
      task_queue_->PostTask([this] { Stop(); });
    }
  }

 private:
  rtc::TaskQueue* const task_queue_;
  const int64_t period_ms_;
  typename std::remove_const<
      typename std::remove_reference<Closure>::type>::type closure_;
  bool running_ = true;
};

template <class Closure>
static PeriodicTask* StartPeriodicTask(rtc::TaskQueue* task_queue,
                                       int64_t period_ms,
                                       Closure&& closure) {
  auto periodic_task = rtc::MakeUnique<PeriodicTaskImpl<Closure>>(
      task_queue, period_ms, std::forward<Closure>(closure));
  PeriodicTask* periodic_task_ptr = periodic_task.get();
  task_queue->PostDelayedTask(std::move(periodic_task), period_ms);
  return periodic_task_ptr;
}

}  // namespace

namespace send_side_cc_internal {
class ControlHandler {
 public:
  ControlHandler(NetworkChangedObserver* observer,
                 PacerController* pacer_controller,
                 const Clock* clock);

  void PostUpdates(NetworkControlUpdate update);

  void OnNetworkAvailability(NetworkAvailability msg);
  void OnOutstandingData(DataSize in_flight_data);
  void OnPacerQueueUpdate(TimeDelta expected_queue_time);

  rtc::Optional<TargetTransferRate> last_transfer_rate();

 private:
  void OnNetworkInvalidation();
  bool GetNetworkParameters(int32_t* estimated_bitrate_bps,
                            uint8_t* fraction_loss,
                            int64_t* rtt_ms);
  bool IsSendQueueFull() const;
  bool HasNetworkParametersToReportChanged(int64_t bitrate_bps,
                                           uint8_t fraction_loss,
                                           int64_t rtt);
  NetworkChangedObserver* observer_ = nullptr;
  PacerController* pacer_controller_;

  rtc::Optional<TargetTransferRate> current_target_rate_msg_;
  rtc::Optional<DataSize> congestion_window_;
  DataSize outstanding_data_ = DataSize::Zero();
  bool network_available_ = true;
  int64_t last_reported_target_bitrate_bps_ = 0;
  uint8_t last_reported_fraction_loss_ = 0;
  int64_t last_reported_rtt_ms_ = 0;
  const bool pacer_pushback_experiment_ = false;
  const bool congestion_window_pushback_experiment_ = false;
  int64_t pacer_expected_queue_ms_ = 0;
  double encoding_rate_ratio_ = 1.0;

  rtc::SequencedTaskChecker sequenced_checker_;
  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(ControlHandler);
};

ControlHandler::ControlHandler(NetworkChangedObserver* observer,
                               PacerController* pacer_controller,
                               const Clock* clock)
    : observer_(observer),
      pacer_controller_(pacer_controller),
      pacer_pushback_experiment_(IsPacerPushbackExperimentEnabled()),
      congestion_window_pushback_experiment_(
          IsCongestionWindowExperimentEnabled()) {
  sequenced_checker_.Detach();
}

void ControlHandler::PostUpdates(NetworkControlUpdate update) {
  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequenced_checker_);
  if (update.congestion_window) {
    congestion_window_ = update.congestion_window;
    pacer_controller_->OnCongestionWindow(*update.congestion_window);
  }
  if (update.pacer_config) {
    pacer_controller_->OnPacerConfig(*update.pacer_config);
  }
  for (const auto& probe : update.probe_cluster_configs) {
{'fraction_non_alphanumeric': 0.09095146536153353, 'fraction_numerical': 0.005360368233991726, 'mean_word_length': 4.171437179873456, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1089185', 'n_tokens_mistral': 5368, 'n_tokens_neox': 5079, 'n_words': 1015}