Add support for transport wide sequence numbers

webrtc/modules/pacing/packet_router.cc

Index: webrtc/modules/pacing/packet_router.cc
diff --git a/webrtc/modules/pacing/packet_router.cc b/webrtc/modules/pacing/packet_router.cc
index 9e15a713174b493af7d89f7ad3971c6a1fb43029..cb9962c5474cbd5d9ee240cb9efae6d81cc2359d 100644
--- a/webrtc/modules/pacing/packet_router.cc
+++ b/webrtc/modules/pacing/packet_router.cc
@@ -10,52 +10,148 @@
 
 #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;
+}
+
+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;
 }
 
-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::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