Implement the NackModule as part of the new jitter buffer.

webrtc/modules/video_coding/nack_module.cc

+/*
+ *  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/base/checks.h"
+#include "webrtc/base/mod_ops.h"
+#include "webrtc/modules/video_coding/nack_module.h"
+#include "webrtc/modules/utility/include/process_thread.h"
+
+namespace webrtc {
+
+NackModule::NackInfo::NackInfo()
+    : seq_num(0), send_at_seq_num(0), send_at_time(0), 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),
+      send_at_time(0),
+      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),
+      running_(true),
+      initialized_(false),
+      rtt_ms_(kDefaultRttMs),
+      last_seq_num_(0),
+      last_process_time_ms_(-1),
+      reordering_count_(0) {
+  memset(reordering_occurences_, -1, sizeof(reordering_occurences_));
+  memset(reordering_buckets_, 0, sizeof(reordering_buckets_));
+  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) seq_num_keyframes_.emplace(unwrapper_.Unwrap(seq_num));

[stefan-webrtc, 2016/02/26 09:26:55]

Maybe we should be unwrapping all sequence numbers and timestamps instead of
mixing unwrapping with modular comparisons?

Either use unwrapping everywhere, or use a map with a specialized comparison
operator? I think I would prefer the latter.

[philipel, 2016/02/26 15:12:07]

Implemented custom comparator.

+    initialized_ = true;
+    return;
+  }
+
+  if (seq_num == last_seq_num_)
+    return;
+
+  if (AheadOf(seq_num, last_seq_num_)) {
+    AddPacketsToNack(last_seq_num_ + 1, seq_num);
+    last_seq_num_ = seq_num;
+
+    int64_t seq_num_unwrapped = unwrapper_.Unwrap(seq_num);
+
+    // Keep track of new keyframes.
+    if (is_keyframe)
+      seq_num_keyframes_.emplace(seq_num_unwrapped);
+
+    // And remove old ones so we don't accumulate keyframes.
+    auto it = seq_num_keyframes_.lower_bound(seq_num_unwrapped -
+                                             kMaxNackPackets);
+    if (it != seq_num_keyframes_.begin())
+      seq_num_keyframes_.erase(seq_num_keyframes_.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());

[stefan-webrtc, 2016/02/26 09:26:55]

Is there any risk that we end up deadlocking here? Packets are being received
and parsed in the rtp module on the same thread which calls SendNack which also
goes to the rtp module. I suspect we may end up taking the same locks in rtp
module, so that we at least get recursive locking?

[philipel, 2016/02/26 15:12:07]

I have not experienced any deadlocks when testing it and by looking at the
RTCPSender::SendCompoundRTCP function I can see that the lock
critical_section_rtcp_sender_ is obtained when sending nacks. I guess we don't
use the RTCPSender while parsing/inserting packets, so I think we should be
fine.

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

Acknowledged.

+  } else {
+    // An out of order packet has been received.
+    RemovePacketFromNack(seq_num);
+    if (!is_retransmitted)
+      UpdateReorderingStatistics(seq_num);
+  }

[stefan-webrtc, 2016/02/26 09:26:55]

move the else to line 70 and make an early return:

if (AheadOf(last_seq_num_, seq_num)) {
  RemovePacketFromNack(seq_num);
  if (!is_retransmitted)
    UpdateReorderingStatistics(seq_num);
  return;
}

AddPacketsToNack(...);
...

[philipel, 2016/02/26 15:12:06]

Done.

+}
+
+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(
+            last_process_time_ms_ +
+              kProcessIntervalMs -
+              clock_->TimeInMilliseconds(),
+              0l);

[stefan-webrtc, 2016/02/26 09:26:55]

0ll is what you want, I think? Or just do 0 as I think that works equally well.

You could also do
std::max<int64_t>(...);

[philipel, 2016/02/26 15:12:07]

Compiler complains if I use 0ll :)

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

Acknowledged.

+}
+
+int32_t NackModule::Process() {
+  rtc::CritScope lock(&crit_);
+  if (!running_)
+    return 0;
+
+  // Update the last_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 (last_process_time_ms_ == 0) {
+    last_process_time_ms_ = now_ms + kProcessIntervalMs;
+  } else {
+    last_process_time_ms_ = last_process_time_ms_ +
+                            kProcessIntervalMs +
+                            (now_ms - 1 - last_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 (!seq_num_keyframes_.empty()) {
+    int64_t kf_unwrapped_seq_num = *seq_num_keyframes_.begin();
+    auto it = nack_list_.lower_bound(kf_unwrapped_seq_num);
+
+    // If this keyframe is so old it does not remove any
+    // packets from the list, remove if from the list of
+    // keyframes and try the next keyframe.
+    if (it == nack_list_.begin()) {
+      seq_num_keyframes_.erase(seq_num_keyframes_.begin());
+      continue;
+    }
+
+    // We have found a keyframe that actually is newer than
+    // atleast one packet in the nack list.
+    nack_list_.erase(nack_list_.begin(), it);
+    return true;
+  }
+  return false;
+}
+
+void NackModule::AddPacketsToNack(uint16_t seq_num_start,
+                                  uint16_t seq_num_end) {
+  // 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();
+      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));
+    int64_t seq_num_unwrapped = unwrapper_.Unwrap(seq_num);
+    nack_list_[seq_num_unwrapped] = nack_info;
+  }
+}
+
+void NackModule::RemovePacketFromNack(uint16_t seq_num) {
+  int64_t seq_num_unwrapped = unwrapper_.Unwrap(seq_num);
+  nack_list_.erase(seq_num_unwrapped);
+}
+
+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.send_at_time == 0 &&
+        AheadOrAt(last_seq_num_, it->second.send_at_seq_num)) {
+      nack_batch.emplace_back(it->second.seq_num);
+      ++it->second.retries;
+      it->second.send_at_time = now_ms + rtt_ms_;
+      if (it->second.retries >= kMaxNackRetries) {
+        it = nack_list_.erase(it);
+      } else {
+        ++it;
+      }
+      continue;
+    }
+
+    if (consider_timestamp && it->second.send_at_time > 0 &&
+        it->second.send_at_time <= now_ms) {
+      nack_batch.emplace_back(it->second.seq_num);
+      ++it->second.retries;
+      it->second.send_at_time = now_ms + rtt_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));
+
+  int remove_from_bucket = reordering_occurences_[reordering_index_];
+  if (remove_from_bucket != -1)
+    reordering_buckets_[remove_from_bucket] -= 1;
+
+  uint16_t diff = ReverseDiff(last_seq_num_, seq_num);
+  uint16_t add_to_bucket =
+      std::min(static_cast<uint16_t>(kNumReorderingBuckets - 1),
+               static_cast<uint16_t>(diff - 1));
+  reordering_buckets_[add_to_bucket] += 1;
+  if (reordering_occurences_[reordering_index_] == -1)
+    ++reordering_count_;
+  reordering_occurences_[reordering_index_] = add_to_bucket;
+  reordering_index_ = Add<kMaxReorderedPackets>(reordering_index_, 1);
+}
+
+int NackModule::WaitNumberOfPackets(float probability) const {
+  RTC_DCHECK_GE(probability, 0.f);
+  RTC_DCHECK_LE(probability, 1.f);
+
+  // Until there are enough values we default to 0. Also
+  // protect against divide by zero later in the code.
+  if (reordering_count_ < 20)
+    return 0;
+
+  int bucket = 0;
+  float accumulated_probability = 0;
+  while (accumulated_probability < probability &&
+         bucket < kNumReorderingBuckets) {
+    accumulated_probability +=
+        static_cast<float>(reordering_buckets_[bucket]) / reordering_count_;
+    ++bucket;
+  }
+  return bucket + 1;
+}
+
+}  // namespace webrtc