NackModule2: Rename to NackRequester.
The alternative new name proposed, NackTracker, is already in
use in audio_coding.
Fixed: webrtc:11594
Change-Id: I6a05fafc05fa7ddb18ea4f64886a135e5ef59f7e
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/226744
Reviewed-by: Ilya Nikolaevskiy <ilnik@webrtc.org>
Reviewed-by: Harald Alvestrand <hta@webrtc.org>
Commit-Queue: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#34539}
diff --git a/modules/video_coding/nack_requester.cc b/modules/video_coding/nack_requester.cc
new file mode 100644
index 0000000..6983576
--- /dev/null
+++ b/modules/video_coding/nack_requester.cc
@@ -0,0 +1,384 @@
+/*
+ * 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 "modules/video_coding/nack_requester.h"
+
+#include <algorithm>
+#include <limits>
+
+#include "api/sequence_checker.h"
+#include "api/task_queue/task_queue_base.h"
+#include "api/units/timestamp.h"
+#include "rtc_base/checks.h"
+#include "rtc_base/experiments/field_trial_parser.h"
+#include "rtc_base/logging.h"
+#include "rtc_base/task_queue.h"
+#include "system_wrappers/include/field_trial.h"
+
+namespace webrtc {
+
+namespace {
+const int kMaxPacketAge = 10000;
+const int kMaxNackPackets = 1000;
+const int kDefaultRttMs = 100;
+const int kMaxNackRetries = 10;
+const int kMaxReorderedPackets = 128;
+const int kNumReorderingBuckets = 10;
+const int kDefaultSendNackDelayMs = 0;
+
+int64_t GetSendNackDelay() {
+ int64_t delay_ms = strtol(
+ webrtc::field_trial::FindFullName("WebRTC-SendNackDelayMs").c_str(),
+ nullptr, 10);
+ if (delay_ms > 0 && delay_ms <= 20) {
+ RTC_LOG(LS_INFO) << "SendNackDelay is set to " << delay_ms;
+ return delay_ms;
+ }
+ return kDefaultSendNackDelayMs;
+}
+} // namespace
+
+constexpr TimeDelta NackPeriodicProcessor::kUpdateInterval;
+
+NackPeriodicProcessor::NackPeriodicProcessor(TimeDelta update_interval)
+ : update_interval_(update_interval) {}
+
+NackPeriodicProcessor::~NackPeriodicProcessor() {}
+
+void NackPeriodicProcessor::RegisterNackModule(NackRequesterBase* module) {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ modules_.push_back(module);
+ if (modules_.size() != 1)
+ return;
+ repeating_task_ = RepeatingTaskHandle::DelayedStart(
+ TaskQueueBase::Current(), update_interval_, [this] {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ ProcessNackModules();
+ return update_interval_;
+ });
+}
+
+void NackPeriodicProcessor::UnregisterNackModule(NackRequesterBase* module) {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ auto it = std::find(modules_.begin(), modules_.end(), module);
+ RTC_DCHECK(it != modules_.end());
+ modules_.erase(it);
+ if (modules_.empty())
+ repeating_task_.Stop();
+}
+
+// RTC_RUN_ON(sequence_)
+void NackPeriodicProcessor::ProcessNackModules() {
+ for (NackRequesterBase* module : modules_)
+ module->ProcessNacks();
+}
+
+ScopedNackPeriodicProcessorRegistration::
+ ScopedNackPeriodicProcessorRegistration(NackRequesterBase* module,
+ NackPeriodicProcessor* processor)
+ : module_(module), processor_(processor) {
+ processor_->RegisterNackModule(module_);
+}
+
+ScopedNackPeriodicProcessorRegistration::
+ ~ScopedNackPeriodicProcessorRegistration() {
+ processor_->UnregisterNackModule(module_);
+}
+
+NackRequester::NackInfo::NackInfo()
+ : seq_num(0), send_at_seq_num(0), sent_at_time(-1), retries(0) {}
+
+NackRequester::NackInfo::NackInfo(uint16_t seq_num,
+ uint16_t send_at_seq_num,
+ int64_t created_at_time)
+ : seq_num(seq_num),
+ send_at_seq_num(send_at_seq_num),
+ created_at_time(created_at_time),
+ sent_at_time(-1),
+ retries(0) {}
+
+NackRequester::BackoffSettings::BackoffSettings(TimeDelta min_retry,
+ TimeDelta max_rtt,
+ double base)
+ : min_retry_interval(min_retry), max_rtt(max_rtt), base(base) {}
+
+absl::optional<NackRequester::BackoffSettings>
+NackRequester::BackoffSettings::ParseFromFieldTrials() {
+ // Matches magic number in RTPSender::OnReceivedNack().
+ const TimeDelta kDefaultMinRetryInterval = TimeDelta::Millis(5);
+ // Upper bound on link-delay considered for exponential backoff.
+ // Selected so that cumulative delay with 1.25 base and 10 retries ends up
+ // below 3s, since above that there will be a FIR generated instead.
+ const TimeDelta kDefaultMaxRtt = TimeDelta::Millis(160);
+ // Default base for exponential backoff, adds 25% RTT delay for each retry.
+ const double kDefaultBase = 1.25;
+
+ FieldTrialParameter<bool> enabled("enabled", false);
+ FieldTrialParameter<TimeDelta> min_retry("min_retry",
+ kDefaultMinRetryInterval);
+ FieldTrialParameter<TimeDelta> max_rtt("max_rtt", kDefaultMaxRtt);
+ FieldTrialParameter<double> base("base", kDefaultBase);
+ ParseFieldTrial({&enabled, &min_retry, &max_rtt, &base},
+ field_trial::FindFullName("WebRTC-ExponentialNackBackoff"));
+
+ if (enabled) {
+ return NackRequester::BackoffSettings(min_retry.Get(), max_rtt.Get(),
+ base.Get());
+ }
+ return absl::nullopt;
+}
+
+NackRequester::NackRequester(TaskQueueBase* current_queue,
+ NackPeriodicProcessor* periodic_processor,
+ Clock* clock,
+ NackSender* nack_sender,
+ KeyFrameRequestSender* keyframe_request_sender)
+ : worker_thread_(current_queue),
+ clock_(clock),
+ nack_sender_(nack_sender),
+ keyframe_request_sender_(keyframe_request_sender),
+ reordering_histogram_(kNumReorderingBuckets, kMaxReorderedPackets),
+ initialized_(false),
+ rtt_ms_(kDefaultRttMs),
+ newest_seq_num_(0),
+ send_nack_delay_ms_(GetSendNackDelay()),
+ backoff_settings_(BackoffSettings::ParseFromFieldTrials()),
+ processor_registration_(this, periodic_processor) {
+ RTC_DCHECK(clock_);
+ RTC_DCHECK(nack_sender_);
+ RTC_DCHECK(keyframe_request_sender_);
+ RTC_DCHECK(worker_thread_);
+ RTC_DCHECK(worker_thread_->IsCurrent());
+}
+
+NackRequester::~NackRequester() {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+}
+
+void NackRequester::ProcessNacks() {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ std::vector<uint16_t> nack_batch = GetNackBatch(kTimeOnly);
+ if (!nack_batch.empty()) {
+ // This batch of NACKs is triggered externally; there is no external
+ // initiator who can batch them with other feedback messages.
+ nack_sender_->SendNack(nack_batch, /*buffering_allowed=*/false);
+ }
+}
+
+int NackRequester::OnReceivedPacket(uint16_t seq_num, bool is_keyframe) {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ return OnReceivedPacket(seq_num, is_keyframe, false);
+}
+
+int NackRequester::OnReceivedPacket(uint16_t seq_num,
+ bool is_keyframe,
+ bool is_recovered) {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ // TODO(philipel): When the packet includes information whether it is
+ // retransmitted or not, use that value instead. For
+ // now set it to true, which will cause the reordering
+ // statistics to never be updated.
+ bool is_retransmitted = true;
+
+ if (!initialized_) {
+ newest_seq_num_ = seq_num;
+ if (is_keyframe)
+ keyframe_list_.insert(seq_num);
+ initialized_ = true;
+ return 0;
+ }
+
+ // Since the |newest_seq_num_| is a packet we have actually received we know
+ // that packet has never been Nacked.
+ if (seq_num == newest_seq_num_)
+ return 0;
+
+ if (AheadOf(newest_seq_num_, seq_num)) {
+ // An out of order packet has been received.
+ auto nack_list_it = nack_list_.find(seq_num);
+ int nacks_sent_for_packet = 0;
+ if (nack_list_it != nack_list_.end()) {
+ nacks_sent_for_packet = nack_list_it->second.retries;
+ nack_list_.erase(nack_list_it);
+ }
+ if (!is_retransmitted)
+ UpdateReorderingStatistics(seq_num);
+ return nacks_sent_for_packet;
+ }
+
+ // Keep track of new keyframes.
+ if (is_keyframe)
+ keyframe_list_.insert(seq_num);
+
+ // And remove old ones so we don't accumulate keyframes.
+ auto it = keyframe_list_.lower_bound(seq_num - kMaxPacketAge);
+ if (it != keyframe_list_.begin())
+ keyframe_list_.erase(keyframe_list_.begin(), it);
+
+ if (is_recovered) {
+ recovered_list_.insert(seq_num);
+
+ // Remove old ones so we don't accumulate recovered packets.
+ auto it = recovered_list_.lower_bound(seq_num - kMaxPacketAge);
+ if (it != recovered_list_.begin())
+ recovered_list_.erase(recovered_list_.begin(), it);
+
+ // Do not send nack for packets recovered by FEC or RTX.
+ return 0;
+ }
+
+ AddPacketsToNack(newest_seq_num_ + 1, seq_num);
+ newest_seq_num_ = seq_num;
+
+ // Are there any nacks that are waiting for this seq_num.
+ std::vector<uint16_t> nack_batch = GetNackBatch(kSeqNumOnly);
+ if (!nack_batch.empty()) {
+ // This batch of NACKs is triggered externally; the initiator can
+ // batch them with other feedback messages.
+ nack_sender_->SendNack(nack_batch, /*buffering_allowed=*/true);
+ }
+
+ return 0;
+}
+
+void NackRequester::ClearUpTo(uint16_t seq_num) {
+ // Called via RtpVideoStreamReceiver2::FrameContinuous on the network thread.
+ worker_thread_->PostTask(ToQueuedTask(task_safety_, [seq_num, this]() {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ nack_list_.erase(nack_list_.begin(), nack_list_.lower_bound(seq_num));
+ keyframe_list_.erase(keyframe_list_.begin(),
+ keyframe_list_.lower_bound(seq_num));
+ recovered_list_.erase(recovered_list_.begin(),
+ recovered_list_.lower_bound(seq_num));
+ }));
+}
+
+void NackRequester::UpdateRtt(int64_t rtt_ms) {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ rtt_ms_ = rtt_ms;
+}
+
+bool NackRequester::RemovePacketsUntilKeyFrame() {
+ // Called on worker_thread_.
+ while (!keyframe_list_.empty()) {
+ auto it = nack_list_.lower_bound(*keyframe_list_.begin());
+
+ if (it != nack_list_.begin()) {
+ // We have found a keyframe that actually is newer than at least one
+ // packet in the nack list.
+ nack_list_.erase(nack_list_.begin(), it);
+ return true;
+ }
+
+ // If this keyframe is so old it does not remove any packets from the list,
+ // remove it from the list of keyframes and try the next keyframe.
+ keyframe_list_.erase(keyframe_list_.begin());
+ }
+ return false;
+}
+
+void NackRequester::AddPacketsToNack(uint16_t seq_num_start,
+ uint16_t seq_num_end) {
+ // Called on worker_thread_.
+ // Remove old packets.
+ auto it = nack_list_.lower_bound(seq_num_end - kMaxPacketAge);
+ nack_list_.erase(nack_list_.begin(), it);
+
+ // If the nack list is too large, remove packets from the nack list until
+ // the latest first packet of a keyframe. If the list is still too large,
+ // clear it and request a keyframe.
+ uint16_t num_new_nacks = ForwardDiff(seq_num_start, seq_num_end);
+ if (nack_list_.size() + num_new_nacks > kMaxNackPackets) {
+ while (RemovePacketsUntilKeyFrame() &&
+ nack_list_.size() + num_new_nacks > kMaxNackPackets) {
+ }
+
+ if (nack_list_.size() + num_new_nacks > kMaxNackPackets) {
+ nack_list_.clear();
+ RTC_LOG(LS_WARNING) << "NACK list full, clearing NACK"
+ " list and requesting keyframe.";
+ keyframe_request_sender_->RequestKeyFrame();
+ return;
+ }
+ }
+
+ for (uint16_t seq_num = seq_num_start; seq_num != seq_num_end; ++seq_num) {
+ // Do not send nack for packets that are already recovered by FEC or RTX
+ if (recovered_list_.find(seq_num) != recovered_list_.end())
+ continue;
+ NackInfo nack_info(seq_num, seq_num + WaitNumberOfPackets(0.5),
+ clock_->TimeInMilliseconds());
+ RTC_DCHECK(nack_list_.find(seq_num) == nack_list_.end());
+ nack_list_[seq_num] = nack_info;
+ }
+}
+
+std::vector<uint16_t> NackRequester::GetNackBatch(NackFilterOptions options) {
+ // Called on worker_thread_.
+
+ bool consider_seq_num = options != kTimeOnly;
+ bool consider_timestamp = options != kSeqNumOnly;
+ Timestamp now = clock_->CurrentTime();
+ std::vector<uint16_t> nack_batch;
+ auto it = nack_list_.begin();
+ while (it != nack_list_.end()) {
+ TimeDelta resend_delay = TimeDelta::Millis(rtt_ms_);
+ if (backoff_settings_) {
+ resend_delay =
+ std::max(resend_delay, backoff_settings_->min_retry_interval);
+ if (it->second.retries > 1) {
+ TimeDelta exponential_backoff =
+ std::min(TimeDelta::Millis(rtt_ms_), backoff_settings_->max_rtt) *
+ std::pow(backoff_settings_->base, it->second.retries - 1);
+ resend_delay = std::max(resend_delay, exponential_backoff);
+ }
+ }
+
+ bool delay_timed_out =
+ now.ms() - it->second.created_at_time >= send_nack_delay_ms_;
+ bool nack_on_rtt_passed =
+ now.ms() - it->second.sent_at_time >= resend_delay.ms();
+ bool nack_on_seq_num_passed =
+ it->second.sent_at_time == -1 &&
+ AheadOrAt(newest_seq_num_, it->second.send_at_seq_num);
+ if (delay_timed_out && ((consider_seq_num && nack_on_seq_num_passed) ||
+ (consider_timestamp && nack_on_rtt_passed))) {
+ nack_batch.emplace_back(it->second.seq_num);
+ ++it->second.retries;
+ it->second.sent_at_time = now.ms();
+ if (it->second.retries >= kMaxNackRetries) {
+ RTC_LOG(LS_WARNING) << "Sequence number " << it->second.seq_num
+ << " removed from NACK list due to max retries.";
+ it = nack_list_.erase(it);
+ } else {
+ ++it;
+ }
+ continue;
+ }
+ ++it;
+ }
+ return nack_batch;
+}
+
+void NackRequester::UpdateReorderingStatistics(uint16_t seq_num) {
+ // Running on worker_thread_.
+ RTC_DCHECK(AheadOf(newest_seq_num_, seq_num));
+ uint16_t diff = ReverseDiff(newest_seq_num_, seq_num);
+ reordering_histogram_.Add(diff);
+}
+
+int NackRequester::WaitNumberOfPackets(float probability) const {
+ // Called on worker_thread_;
+ if (reordering_histogram_.NumValues() == 0)
+ return 0;
+ return reordering_histogram_.InverseCdf(probability);
+}
+
+} // namespace webrtc