Add support for transport wide sequence numbers

webrtc/modules/pacing/packet_router.cc

 /*
  *  Copyright (c) 2015 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 "webrtc/modules/pacing/include/packet_router.h"
 
+#include "webrtc/base/atomicops.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/modules/rtp_rtcp/interface/rtp_rtcp.h"
 #include "webrtc/modules/rtp_rtcp/interface/rtp_rtcp_defines.h"
-#include "webrtc/system_wrappers/interface/critical_section_wrapper.h"
 
 namespace webrtc {
 
+static const int64_t kSendTimeHistoryWindowMs = 2000;
+
 PacketRouter::PacketRouter()
-    : crit_(CriticalSectionWrapper::CreateCriticalSection()) {
+    : dirty_map_(0), transport_wide_seq_enabled_(false), transport_seq_(0) {
 }
 
 PacketRouter::~PacketRouter() {
+  DCHECK(rtp_modules_.empty());
 }
 
 void PacketRouter::AddRtpModule(RtpRtcp* rtp_module) {
-  CriticalSectionScoped cs(crit_.get());
-  DCHECK(std::find(rtp_modules_.begin(), rtp_modules_.end(), rtp_module) ==
-         rtp_modules_.end());
-  rtp_modules_.push_back(rtp_module);
+  rtc::CritScope cs(&modules_lock_);
+  UpdateModuleMap();
+  uint32_t ssrc = rtp_module->SSRC();
+  DCHECK(rtp_modules_.find(ssrc) == rtp_modules_.end());
+  rtp_modules_[ssrc] = rtp_module;
 }
 
 void PacketRouter::RemoveRtpModule(RtpRtcp* rtp_module) {
-  CriticalSectionScoped cs(crit_.get());
-  rtp_modules_.remove(rtp_module);
+  rtc::CritScope cs(&modules_lock_);
+  UpdateModuleMap();
+  auto it = rtp_modules_.find(rtp_module->SSRC());
+  DCHECK(it != rtp_modules_.end());
+  rtp_modules_.erase(it);
+}
+
+void PacketRouter::OnSsrcChanged() {
+  // Just flag module map as dirty, to avoid taking the ssrc_lookup_lock and
+  // cause potential lock order inversions.
+  rtc::AtomicOps::Increment(&dirty_map_);
+}
+
+void PacketRouter::UpdateModuleMap() {
+  int dirty;
+  do {
+    // Load atomic flag and return immediately if not dirty.
+    dirty = rtc::AtomicOps::AcquireLoad(&dirty_map_);
+    if (dirty <= 0)
+      return;
+
+    // Map was dirty, re-map all modules.
+    std::map<uint32_t, RtpRtcp*> updated_map;
+    for (auto it : rtp_modules_)
+      updated_map[it.second->SSRC()] = it.second;
+    rtp_modules_ = updated_map;
+
+    // If dirty-flag was concurrently set again, we need to make another loop.
+  } while (!rtc::AtomicOps::CompareAndSwap(&dirty_map_, dirty, 0));
 }
 
 bool PacketRouter::TimeToSendPacket(uint32_t ssrc,
                                     uint16_t sequence_number,
                                     int64_t capture_timestamp,
                                     bool retransmission) {
-  CriticalSectionScoped cs(crit_.get());
-  for (auto* rtp_module : rtp_modules_) {
-    if (rtp_module->SendingMedia() && ssrc == rtp_module->SSRC()) {
-      return rtp_module->TimeToSendPacket(ssrc, sequence_number,
-                                          capture_timestamp, retransmission);
+  rtc::CritScope cs(&modules_lock_);
+  UpdateModuleMap();
+  auto it = rtp_modules_.find(ssrc);
+  if (it == rtp_modules_.end())
+    return true;
+  RtpRtcp* rtp_module = it->second;
+
+  if (!rtp_module || !rtp_module->SendingMedia())
+    return true;
+
+  return rtp_module->TimeToSendPacket(ssrc, sequence_number, capture_timestamp,
+                                      retransmission);
+}
+
+size_t PacketRouter::TimeToSendPadding(size_t bytes_to_send) {
+  size_t total_bytes_sent = 0;
+  rtc::CritScope cs(&modules_lock_);
+  for (auto it : rtp_modules_) {
+    if (it.second->SendingMedia()) {
+      size_t bytes_sent =
+          it.second->TimeToSendPadding(bytes_to_send - total_bytes_sent);
+      total_bytes_sent += bytes_sent;
+      if (total_bytes_sent >= bytes_to_send)
+        break;
     }
   }
-  return true;
+  return total_bytes_sent;
 }
 
-size_t PacketRouter::TimeToSendPadding(size_t bytes) {
-  CriticalSectionScoped cs(crit_.get());
-  for (auto* rtp_module : rtp_modules_) {
-    if (rtp_module->SendingMedia())
-      return rtp_module->TimeToSendPadding(bytes);
+void PacketRouter::EnableTransportWideFeedback() {
+  if (transport_wide_seq_enabled_)
+    return;
+
+  rtc::CritScope cs(&history_lock_);
+  send_time_history_.reset(new SendTimeHistory(kSendTimeHistoryWindowMs));
+  transport_wide_seq_enabled_ = true;
+}
+
+void PacketRouter::SetTransportWideSequenceNumber(uint16_t sequence_number) {
+  if (!transport_wide_seq_enabled_)
+    return;
+
+  rtc::AtomicOps::ReleaseStore(&transport_seq_, sequence_number);
+}
+
+size_t PacketRouter::PopulateSendTimes(std::vector<PacketInfo>* packet_info) {
+  if (!transport_wide_seq_enabled_)
+    return 0;
+
+  size_t successful_lookups = 0;
+  rtc::CritScope cs(&history_lock_);
+  for (PacketInfo& info : *packet_info) {
+    if (send_time_history_->GetSendTime(info.sequence_number,
+                                        &info.send_time_ms, true)) {
+      ++successful_lookups;
+    }
   }
-  return 0;
+  return successful_lookups;
 }
+
+uint16_t PacketRouter::AllocateSequenceNumber() {
+  if (!transport_wide_seq_enabled_)
+    return 0;
+
+  int prev_seq;
+  int new_seq;
+  do {

[stefan-webrtc, 2015/07/27 12:13:49]

I think we should comment on this code. I have read it a couple of times and I'm
still not sure why we compare the result of CompareAndSwap() to prev_seq?

[sprang_webrtc, 2015/07/28 13:29:27]

Added comment and also reformulated code a bit.

+    prev_seq = rtc::AtomicOps::AcquireLoad(&transport_seq_);
+    new_seq = (prev_seq + 1) & 0xFFFF;
+  } while (rtc::AtomicOps::CompareAndSwap(&transport_seq_, prev_seq, new_seq) !=
+           prev_seq);
+
+  return new_seq;
+}
+
+void PacketRouter::OnPacketSent(uint16_t sequence_number, int64_t send_time) {
+  if (!transport_wide_seq_enabled_)
+    return;
+
+  rtc::CritScope cs(&history_lock_);
+  send_time_history_->AddAndRemoveOldSendTimes(sequence_number, send_time);
+}
+
 }  // namespace webrtc