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..7fb87f77dd12317c64d3e2509161c818b950321c
--- /dev/null
+++ b/webrtc/modules/video_coding/nack_module.cc
@@ -0,0 +1,254 @@
+/*
+ *  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"

[stefan-webrtc, 2016/02/23 12:29:00]

I tend to prefer short names. Should we call this nack.h/.cc instead?

[philipel, 2016/02/25 15:09:31]

Acknowledged.

+#include "webrtc/modules/utility/include/process_thread.h"
+
+namespace webrtc {
+
+// ===========================================================
+// ||                                                       ||
+// ||                       NackModule                      ||
+// ||                                                       ||
+// ===========================================================

[stefan-webrtc, 2016/02/23 12:29:00]

I prefer not having comments like this.

[philipel, 2016/02/25 15:09:31]

Done.

+
+// Needs to be defined in the .cc file since these values has
+// to be referenced. Actual values can be found in .h file.
+const uint16_t NackModule::kMaxReorderedPackets;
+const uint16_t NackModule::kNumReorderingBuckets;
+const uint16_t NackModule::kMaxNackPackets;
+
+NackModule::NackModule(CriticalSectionWrapper* critical_section,
+                       Clock* clock,
+                       VCMNackSender* nack_sender,
+                       VCMKeyFrameRequestSender* keyframe_request_sender)
+    : critical_section_(critical_section),
+      clock_(clock),
+      nack_sender_(nack_sender),
+      keyframe_request_sender_(keyframe_request_sender),
+      running_(true),
+      last_seq_num_(0),
+      initialized_(false),
+      rtt_ms_(kDefaultRttMs),
+      last_process_timestamp_ms_(0),

[stefan-webrtc, 2016/02/23 12:29:01]

Initialize to -1 instead. 0 is a bad initial value for time.

[philipel, 2016/02/25 15:09:32]

Done.

+      last_keyframe_seq_num_(0),
+      reordering_count_(0) {
+  memset(reordering_occurences_, -1, sizeof(reordering_occurences_));
+  memset(reordering_buckets_, 0, sizeof(reordering_buckets_));
+  RTC_DCHECK(critical_section_ != nullptr);
+  RTC_DCHECK(clock_ != nullptr);
+  RTC_DCHECK(nack_sender_ != nullptr);
+  RTC_DCHECK(keyframe_request_sender_ != nullptr);

[stefan-webrtc, 2016/02/23 12:29:01]

You don't need != nullptr.

Just do RTC_DCHECK(keyframe_request_sender_);

[philipel, 2016/02/25 15:09:31]

Done.

+}
+
+NackModule::~NackModule() {
+  CriticalSectionScoped cs(critical_section_);

[stefan-webrtc, 2016/02/23 12:29:01]

Seems odd that we would have to take the critical section in the destructor. Why
is that?

[philipel, 2016/02/25 15:09:31]

If some thread is inside another function then the object wont be destroyed
until that thread is done. But if the nack module is destroyed when some other
thread is using it then something is very wrong.

Removed the destructor.

+  running_ = false;
+}
+
+void NackModule::Notify(const VCMPacket& packet) {
+  CriticalSectionScoped cs(critical_section_);

[stefan-webrtc, 2016/02/23 12:29:01]

Name this lock instead of cs, and use rtc::CritScoped

[philipel, 2016/02/25 15:09:31]

Done.

+  if (!running_)
+    return;
+  uint16_t seq_num(packet.seqNum);

[stefan-webrtc, 2016/02/23 12:29:00]

uint16_t seq_num = packet.seqNum;

or use packet.seqNum directly?

[philipel, 2016/02/25 15:09:31]

If we change packet class then we only have to change here.

+  // 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;
+    last_keyframe_seq_num_ = seq_num;
+    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;
+    if (is_keyframe)
+      last_keyframe_seq_num_ = seq_num;
+    std::vector<uint16_t> nack_batch = GetNackBatch(true, false);

[stefan-webrtc, 2016/02/23 12:29:01]

I think this call is a bit hard to read since I need to know what true and false
means. Could we use enums or helper methods with a more descriptive name?

[philipel, 2016/02/25 15:09:31]

Done.

+    if (!nack_batch.empty() && nack_sender_ != nullptr)

[stefan-webrtc, 2016/02/23 12:29:01]

Can we make sure nack_sender_ is always set so that we don't have to check this?

[philipel, 2016/02/25 15:09:31]

Done.

+      nack_sender_->SendNack(nack_batch.data(), nack_batch.size());
+  } else {
+    // An out of order packet has been received.
+    RemovePacketFromNack(seq_num);
+    if (!is_retransmitted)
+      UpdateReorderingStatistics(seq_num);
+  }
+}
+
+void NackModule::UpdateRtt(int64_t rtt_ms) {
+  CriticalSectionScoped cs(critical_section_);
+  rtt_ms_ = rtt_ms;
+}
+
+void NackModule::Stop() {
+  CriticalSectionScoped cs(critical_section_);
+  running_ = false;
+}
+
+int64_t NackModule::TimeUntilNextProcess() {
+  CriticalSectionScoped cs(critical_section_);
+  return last_process_timestamp_ms_ + kProcessIntervalMs -
+         clock_->TimeInMilliseconds();

[stefan-webrtc, 2016/02/23 12:29:01]

Maybe use max(0, expr) to ensure it doesn't get negative?

[philipel, 2016/02/25 15:09:31]

Done.

+}
+
+int32_t NackModule::Process() {
+  CriticalSectionScoped cs(critical_section_);
+  if (!running_)
+    return 0;
+
+  // Update the last_process_timestamp_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_timestamp_ms_ == 0)

[stefan-webrtc, 2016/02/23 12:29:00]

s/timestamp/time

Use brackets {} for multi-line if/else-statements.

[philipel, 2016/02/25 15:09:31]

Done.

+    last_process_timestamp_ms_ = now_ms + kProcessIntervalMs;
+  else
+    last_process_timestamp_ms_ = last_process_timestamp_ms_ +
+                                 kProcessIntervalMs +
+                                 (now_ms - 1 - last_process_timestamp_ms_) /
+                                     kProcessIntervalMs * kProcessIntervalMs;

[stefan-webrtc, 2016/02/23 12:29:00]

All of this looks really strange to me. Why not simply set
last_process_timestamp_ms_ = now_ms?

[philipel, 2016/02/25 15:09:32]

The idea was that to make sure we send at a certain rate over time. If process
is called on average lets say 5 ms late then the interval would become 25 ms
instead of 20, or 40 times per second instead of 50.

In short, it is a way to compensate for the worker thread to always call process
a bit late (usually somewhere between 2-8 ms).

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

Ok, but isn't it weird to not set last_process_time_ms_ = now_ms on the first
call to Process(). Why now_ms + kProcessIntervalMs? Sounds more like
next_process_time_ms_ to me.

I'd like to have unittests which verifies this is correct too.

How much we want to compensate for this? We don't want to burst out 5 nacks in
10 ms just because the worker thread was busy for some time, right? Does this
guarantee this won't happen? Maybe add a unittest for that case too.

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

The name last_process_timestamp_ms_ is not perfect since it doesn't actually
tells us when it was last processed, it is the adjusted timestamp to achieve a
certain frequency.

The last two lines of the calculation is used to adjust the time so we don't
burst if the worker thread has been busy. Maybe I should change my comment if it
wasn't clear :)

I will absolutely implement a unittest for this.

+
+  std::vector<uint16_t> nack_batch = GetNackBatch(false, true);
+  if (!nack_batch.empty() && nack_sender_ != nullptr)
+    nack_sender_->SendNack(nack_batch.data(), nack_batch.size());
+
+  return 0;
+}
+
+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 = PositiveDiff(seq_num_start, seq_num_end);

[stefan-webrtc, 2016/02/23 12:29:00]

Are these sequence numbers guaranteed not to already be in the nack list?

[philipel, 2016/02/25 15:09:30]

This function is called from Notify (now OnReceivePacket) and the logic for
detecting new packets to nack is in the Notify function, so yes, they SHOULD not
already be in the nack list.

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

Would it make sense to dcheck on that somewhere in this method?

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

It would, will add it for the next patch.

+  if (nack_list_.size() + num_new_nacks >= kMaxNackPackets) {
+    NackList::iterator nack_it = nack_list_.lower_bound(last_keyframe_seq_num_);

[stefan-webrtc, 2016/02/23 12:29:01]

What if last_keyframe_seq_num_ is 0xffff and the last incoming packet had
sequence number 0? Don't we end up erasing 0 in that case?

[stefan-webrtc, 2016/02/23 12:29:01]

auto instead of NackList::iterator, and just call it "it".

How do we notify the jitter buffer that it should skip ahead to the last key
frame when this happens? And is the last key frame correct, or should we pick
the oldest key frame?

[philipel, 2016/02/25 15:09:31]

Changed the implementation completely, now we clear until the next keyframe and
continue to do so until there is enough space.

+    if (nack_it != nack_list_.end())
+      nack_list_.erase(nack_list_.begin(), nack_it);
+
+    if (nack_list_.size() + num_new_nacks >= kMaxNackPackets) {

[stefan-webrtc, 2016/02/23 12:29:00]

Are the new sequence numbers guaranteed to be later than last_keyframe_seq_num_,
or is there a chance they should have been removed too?

+      nack_list_.clear();
+      keyframe_request_sender_->RequestKeyFrame();

[stefan-webrtc, 2016/02/23 12:29:01]

What do you think of instead returning kRequestKeyFrame if the nack module needs
a key frame to avoid another callback?

[philipel, 2016/02/25 15:09:31]

Don't we need to do the callback anyway, only in the jitter buffer instead of in
the nack module? I would like to keep the nack module as contained as possible.

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

Ok! At some point we will have to call back, yes.

+      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));
+    nack_list_[seq_num] = nack_info;
+  }
+}
+
+void NackModule::RemovePacketFromNack(uint16_t seq_num) {
+  nack_list_.erase(seq_num);
+}
+
+std::vector<uint16_t> NackModule::GetNackBatch(bool consider_seq_num,
+                                               bool consider_timestamp) {
+  int64_t now_ms = clock_->TimeInMilliseconds();
+  std::vector<uint16_t> nack_batch;
+  NackList::iterator nack_it = nack_list_.begin();
+  while (nack_it != nack_list_.end()) {
+    if (consider_seq_num && nack_it->second.send_at_timestamp == 0 &&
+        AheadOrAt(last_seq_num_, nack_it->second.send_at_seq_num)) {
+      nack_batch.emplace_back(nack_it->second.seq_num);
+      nack_it->second.retries++;

[stefan-webrtc, 2016/02/23 12:29:01]

++...retries;

[philipel, 2016/02/25 15:09:31]

Done.

+      nack_it->second.send_at_timestamp = now_ms + rtt_ms_;
+      if (nack_it->second.retries >= kMaxNackRetries)

[stefan-webrtc, 2016/02/23 12:29:01]

Brackets

+        nack_it = nack_list_.erase(nack_it);

[stefan-webrtc, 2016/02/23 12:29:00]

If we decide to stop trying, should we then make sure there is a key frame to
skip ahead to? Or should we allow nacking indefinitely and rely on the jitter
buffer to let us know when to stop retrying? Maybe by having an API which it can
call to notify the nack module that it's no longer interested in packets with
timestamp < X (or seq num < Y)?

[philipel, 2016/02/25 15:09:31]

I think a ClearUpToSequenceNumber(uint16_t seq_num) function would be very
useful to have. But I still think we should avoid signaling to the jitter buffer
that we have stopped nacking a packet. The general idea is that the nack module
will try hard to get every packet, and if that is not possible then the jitter
buffer has to recover on its own. Of course requesting a keyframe from the nack
module should be very rare.

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

So in that case we will be waiting for the jitter buffer to time out and request
a key frame on its own? The question is how much extra latency we will get for
having to wait for the jitter buffer timeout. The trade off is that the nack
module may be requesting a key frame for a packet it has stopped nacking which
wasn't really needed to continue decoding.

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

The idea with this nack module is that it should never stop nacking a packet
until we have tried really really hard to get that packet and it seems extremely
unlikely to ever get those packets. If this condition is met then the jitter
buffer should have tried to recover before that.

+      else
+        ++nack_it;
+      continue;
+    }
+
+    if (consider_timestamp && nack_it->second.send_at_timestamp > 0 &&
+        nack_it->second.send_at_timestamp <= now_ms) {
+      nack_batch.emplace_back(nack_it->second.seq_num);
+      nack_it->second.retries++;
+      nack_it->second.send_at_timestamp = now_ms + rtt_ms_;
+      if (nack_it->second.retries >= kMaxNackRetries)

[stefan-webrtc, 2016/02/23 12:29:01]

See above

[philipel, 2016/02/25 15:09:31]

Done.

+        nack_it = nack_list_.erase(nack_it);
+      else
+        ++nack_it;
+      continue;
+    }
+    ++nack_it;
+  }
+  return nack_batch;
+}
+
+void NackModule::UpdateReorderingStatistics(uint16_t seq_num) {

[stefan-webrtc, 2016/02/23 12:29:01]

I think I would recommend breaking out the histogram handling and probability
computations into a separate histogram class that can be tested in isolation.

[philipel, 2016/02/25 15:09:31]

I agree, will do it for the next upload.

+  RTC_DCHECK(AheadOf(last_seq_num_, seq_num));
+
+  int8_t remove_from_bucket = reordering_occurences_[reordering_index_];

[stefan-webrtc, 2016/02/23 12:29:01]

int

[philipel, 2016/02/25 15:09:31]

Done.

+  if (remove_from_bucket != -1)
+    reordering_buckets_[remove_from_bucket] -= 1;
+

[stefan-webrtc, 2016/02/23 12:29:01]

DCHECK_GE(remove_from_bucket, 0);
and maybe also DCHECK_LT(remove_from_bucket, reordering_buckets_.size())?

+  uint16_t diff = NegativeDiff(last_seq_num_, seq_num);

[stefan-webrtc, 2016/02/23 12:29:00]

Is this the same as PositiveDiff(seq_num, last_seq_num)? I think it's a bit hard
to interpret what a negative diff is.

[philipel, 2016/02/25 15:09:32]

NegativeDiff(a,b) == PositiveDiff(b,a)
Added comments in mod_ops.h which describes what it does.

+  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);
+}
+
+uint8_t NackModule::WaitNumberOfPackets(float probability) const {

[stefan-webrtc, 2016/02/23 12:29:01]

return int

[philipel, 2016/02/25 15:09:31]

Done.

+  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;
+
+  uint8_t bucket = 0;
+  float accumelated_probability = 0;

[stefan-webrtc, 2016/02/23 12:29:01]

accumulated

[philipel, 2016/02/25 15:09:31]

Done.

+  while (accumelated_probability < probability &&
+         bucket < kNumReorderingBuckets) {
+    accumelated_probability +=
+        static_cast<float>(reordering_buckets_[bucket]) / reordering_count_;
+    ++bucket;
+  }
+  return bucket + 1;
+}
+
+// ===========================================================
+// ||                                                       ||
+// ||                NackModule::NackInfo                   ||
+// ||                                                       ||
+// ===========================================================

[stefan-webrtc, 2016/02/23 12:29:01]

Remove

[philipel, 2016/02/25 15:09:31]

Done.

+NackModule::NackInfo::NackInfo()
+    : seq_num(0), send_at_seq_num(0), send_at_timestamp(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_timestamp(0),
+      retries(0) {}

[stefan-webrtc, 2016/02/23 12:29:00]

Maybe put these methods at the top instead.

[philipel, 2016/02/25 15:09:30]

Done.

+
+}  // namespace webrtc