Implement the NackModule as part of the new jitter buffer.

webrtc/modules/video_coding/nack_module.cc

Index: webrtc/modules/video_coding/nack_module.cc
diff --git a/webrtc/modules/video_coding/nack_module.cc b/webrtc/modules/video_coding/nack_module.cc
new file mode 100644
index 0000000000000000000000000000000000000000..ada0d012fefd6642fff44e879f115f716e83fb71
--- /dev/null
+++ b/webrtc/modules/video_coding/nack_module.cc
@@ -0,0 +1,258 @@
+/*
+ *  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 <algorithm>
+#include <limits>
+
+#include "webrtc/modules/video_coding/nack_module.h"
+
+#include "webrtc/base/checks.h"
+#include "webrtc/base/mod_ops.h"
+#include "webrtc/modules/utility/include/process_thread.h"
+
+namespace webrtc {
+
+namespace {
+  const int kMaxPacketAge = 10000;
+  const int kMaxNackPackets = 1000;
+  const int kDefaultRttMs = 100;
+  const int kMaxNackRetries = 10;
+  const int kProcessFrequency = 50;
+  const int kProcessIntervalMs = 1000 / kProcessFrequency;
+  const int kMaxReorderedPackets = 128;
+  const int kNumReorderingBuckets = 10;
+}  // namespace
+
+NackModule::NackInfo::NackInfo()
+    : seq_num(0), send_at_seq_num(0), sent_at_time(-1), retries(0) {}
+
+NackModule::NackInfo::NackInfo(uint16_t seq_num, uint16_t send_at_seq_num)
+    : seq_num(seq_num),
+      send_at_seq_num(send_at_seq_num),
+      sent_at_time(-1),
+      retries(0) {}
+
+NackModule::NackModule(Clock* clock,
+                       VCMNackSender* nack_sender,
+                       VCMKeyFrameRequestSender* keyframe_request_sender)
+    : clock_(clock),
+      nack_sender_(nack_sender),
+      keyframe_request_sender_(keyframe_request_sender),
+      reordering_distribution_(kNumReorderingBuckets, kMaxReorderedPackets),
+      running_(true),
+      initialized_(false),
+      rtt_ms_(kDefaultRttMs),
+      last_seq_num_(0),
+      next_process_time_ms_(-1) {
+  RTC_DCHECK(clock_);
+  RTC_DCHECK(nack_sender_);
+  RTC_DCHECK(keyframe_request_sender_);
+}
+
+void NackModule::OnReceivedPacket(const VCMPacket& packet) {
+  rtc::CritScope lock(&crit_);
+  if (!running_)
+    return;
+  uint16_t seq_num = packet.seqNum;
+  // 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;
+  bool is_keyframe = packet.isFirstPacket && packet.frameType == kVideoFrameKey;
+
+  if (!initialized_) {
+    last_seq_num_ = seq_num;
+    if (is_keyframe) keyframe_list_.emplace(seq_num);

[stefan-webrtc, 2016/03/01 08:52:10]

Is there a reason for doing emplace rather than insert?

[philipel, 2016/03/01 10:27:06]

Good habit since emplace does the same thing as insert only it might do it more
efficient for certain data types.

[stefan-webrtc, 2016/03/01 12:16:34]

Acknowledged.

+    initialized_ = true;
+    return;
+  }
+
+  if (seq_num == last_seq_num_)
+    return;
+
+  if (AheadOf(last_seq_num_, seq_num)) {
+    // An out of order packet has been received.
+    RemovePacketFromNack(seq_num);
+    if (!is_retransmitted)
+      UpdateReorderingStatistics(seq_num);
+    return;
+  } else {
+    AddPacketsToNack(last_seq_num_ + 1, seq_num);
+    last_seq_num_ = seq_num;
+
+    // Keep track of new keyframes.
+    if (is_keyframe)
+      keyframe_list_.emplace(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);
+
+    // Are there any nacks that are waiting for this seq_num.
+    std::vector<uint16_t> nack_batch = GetNackBatch(kSeqNumOnly);
+    if (!nack_batch.empty())
+      nack_sender_->SendNack(nack_batch.data(), nack_batch.size());
+  }
+}
+
+void NackModule::ClearUpTo(uint16_t seq_num) {
+  rtc::CritScope lock(&crit_);
+  nack_list_.erase(nack_list_.begin(),
+                   nack_list_.lower_bound(seq_num));
+  keyframe_list_.erase(keyframe_list_.begin(),
+                       keyframe_list_.lower_bound(seq_num));
+}
+
+void NackModule::UpdateRtt(int64_t rtt_ms) {
+  rtc::CritScope lock(&crit_);
+  rtt_ms_ = rtt_ms;
+}
+
+void NackModule::Stop() {
+  rtc::CritScope lock(&crit_);
+  running_ = false;
+}
+
+int64_t NackModule::TimeUntilNextProcess() {
+  rtc::CritScope lock(&crit_);
+  return std::max(next_process_time_ms_ - clock_->TimeInMilliseconds(), 0l);
+}
+
+int32_t NackModule::Process() {
+  rtc::CritScope lock(&crit_);
+  if (!running_)
+    return 0;
+
+  // Update the next_process_time_ms_ in intervals to achieve
+  // the targeted frequency over time. Also add multiple intervals
+  // in case of a skip in time as to not make uneccessary
+  // calls to Process in order to catch up.
+  int64_t now_ms = clock_->TimeInMilliseconds();
+  if (next_process_time_ms_ == -1) {
+    next_process_time_ms_ = now_ms + kProcessIntervalMs;
+  } else {
+    next_process_time_ms_ = next_process_time_ms_ +
+                            kProcessIntervalMs +
+                            (now_ms - next_process_time_ms_) /
+                            kProcessIntervalMs * kProcessIntervalMs;
+  }
+
+  std::vector<uint16_t> nack_batch = GetNackBatch(kTimeOnly);
+  if (!nack_batch.empty() && nack_sender_ != nullptr)
+    nack_sender_->SendNack(nack_batch.data(), nack_batch.size());
+
+  return 0;
+}
+
+bool NackModule::RemovePacketsUntilKeyframe() {
+  while (!keyframe_list_.empty()) {
+    auto it = nack_list_.lower_bound(*keyframe_list_.begin());
+
+    // If this keyframe is so old it does not remove any
+    // packets from the list, remove if from the list of

[stefan-webrtc, 2016/03/01 08:52:10]

"remove *it* from the list"

[philipel, 2016/03/01 10:27:06]

Done.

+    // keyframes and try the next keyframe.
+    if (it == nack_list_.begin()) {
+      keyframe_list_.erase(keyframe_list_.begin());
+      continue;
+    }
+
+    // We have found a keyframe that actually is newer than
+    // atleast one packet in the nack list.

[stefan-webrtc, 2016/03/01 08:52:10]

at least

[philipel, 2016/03/01 10:27:06]

Done.

+    nack_list_.erase(nack_list_.begin(), it);
+    return true;

[stefan-webrtc, 2016/03/01 08:52:10]

I'd prefer to change the structure of this loop a bit, so that the special case
is to return true and the default is to continue, e.g.:

while (...) {
  auto it = lower_bound(...);
  if (it != nack_list_.begin()) {
    DCHECK(it != nack_list.end());
    nack_list_.erase(...);
    return true;
  }
  key_frame_list_.erase(...);
}

[philipel, 2016/03/01 10:27:06]

Done.

+  }
+  return false;
+}
+
+void NackModule::AddPacketsToNack(uint16_t seq_num_start,
+                                  uint16_t seq_num_end) {
+  // Remove old packets.
+  auto it = nack_list_.lower_bound(seq_num_end - kMaxPacketAge);

[stefan-webrtc, 2016/03/01 08:52:10]

We've been talking about adding support for a time limit on NACK lists as
negotiated with rtx (see rtx-time here
https://tools.ietf.org/html/rfc4588#section-8.1). Is that something we should
take into account here already?

[philipel, 2016/03/01 10:27:06]

It is something we can implement later on, I don't want to add more to this CL
than necessary. When we add it I think it's easier to add it in the GetNackBatch
functions which can remove all old packets according to some timestamp we can
add in the NackInfo struct.

[stefan-webrtc, 2016/03/01 12:16:34]

Sounds good.

+  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() &&

[stefan-webrtc, 2016/03/01 08:52:10]

Search and replace Keyframe -> KeyFrame

[philipel, 2016/03/01 10:27:06]

Done.

+          nack_list_.size() + num_new_nacks > kMaxNackPackets) {}
+
+    if (nack_list_.size() + num_new_nacks > kMaxNackPackets) {
+      nack_list_.clear();
+      keyframe_request_sender_->RequestKeyFrame();
+      return;
+    }
+  }
+
+  for (uint16_t seq_num = seq_num_start; seq_num != seq_num_end; ++seq_num) {
+    NackInfo nack_info(seq_num, seq_num + WaitNumberOfPackets(0.5));
+    RTC_DCHECK(nack_list_.find(seq_num) == nack_list_.end());
+    nack_list_[seq_num] = nack_info;
+  }
+}
+
+void NackModule::RemovePacketFromNack(uint16_t seq_num) {

[stefan-webrtc, 2016/03/01 08:52:10]

Maybe remove this method now that it's trivial?

[philipel, 2016/03/01 10:27:06]

Done.

+  nack_list_.erase(seq_num);
+}
+
+std::vector<uint16_t> NackModule::GetNackBatch(NackFilterOptions options) {
+  bool consider_seq_num = options != kTimeOnly;
+  bool consider_timestamp = options != kSeqNumOnly;
+  int64_t now_ms = clock_->TimeInMilliseconds();
+  std::vector<uint16_t> nack_batch;
+  auto it = nack_list_.begin();
+  while (it != nack_list_.end()) {
+    if (consider_seq_num &&
+        it->second.sent_at_time == -1 &&
+        AheadOrAt(last_seq_num_, it->second.send_at_seq_num)) {
+      nack_batch.emplace_back(it->second.seq_num);
+      ++it->second.retries;
+      it->second.sent_at_time = now_ms;
+      if (it->second.retries >= kMaxNackRetries) {
+        it = nack_list_.erase(it);
+      } else {
+        ++it;
+      }
+      continue;
+    }
+
+    if (consider_timestamp &&
+        it->second.sent_at_time + rtt_ms_ <= now_ms) {
+      nack_batch.emplace_back(it->second.seq_num);
+      ++it->second.retries;
+      it->second.sent_at_time = now_ms;
+      if (it->second.retries >= kMaxNackRetries) {
+        it = nack_list_.erase(it);
+      } else {
+        ++it;
+      }
+      continue;
+    }
+    ++it;
+  }
+  return nack_batch;
+}
+
+void NackModule::UpdateReorderingStatistics(uint16_t seq_num) {
+  RTC_DCHECK(AheadOf(last_seq_num_, seq_num));
+  uint16_t diff = ReverseDiff(last_seq_num_, seq_num);
+  reordering_distribution_.AddValue(diff);
+}
+
+int NackModule::WaitNumberOfPackets(float probability) const {
+  if (reordering_distribution_.NumValues() == 0)
+    return 0;
+  return reordering_distribution_.InverseCDF(probability);
+}
+
+}  // namespace webrtc