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..24ae6d94311b8fbf148226a6d5affb022d90afe9 100644
--- a/webrtc/modules/pacing/packet_router.cc
+++ b/webrtc/modules/pacing/packet_router.cc
@@ -10,52 +10,113 @@
 
 #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 {
 
-PacketRouter::PacketRouter()
-    : crit_(CriticalSectionWrapper::CreateCriticalSection()) {
+PacketRouter::PacketRouter() : dirty_map_(0), 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);
+}

[stefan-webrtc, 2015/07/29 09:04:11]

I'd say you could argue whether the code actually became better by switching to
a map from ssrc to modules instead of having the list. What do you think? Is it
worth doing?

[sprang_webrtc, 2015/07/29 10:03:25]

Especially after it turned out that we actually need to hold the lock while
calling the rtp modules, no. This is not worth it. 
Will still keep some things though, like changes to vie_channel and asserts here
that we don't double-remove. Also change to allow padding on multiple modules.
WDYT?

+
+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);
-  }
-  return 0;
+void PacketRouter::SetTransportWideSequenceNumber(uint16_t sequence_number) {
+  rtc::AtomicOps::ReleaseStore(&transport_seq_, sequence_number);
 }
+
+uint16_t PacketRouter::AllocateSequenceNumber() {
+  int prev_seq = rtc::AtomicOps::AcquireLoad(&transport_seq_);
+  int desired_prev_seq;
+  int new_seq;
+  do {
+    desired_prev_seq = prev_seq;
+    new_seq = (desired_prev_seq + 1) & 0xFFFF;
+    // Note: CompareAndSwap returns the actual value of transport_seq at the
+    // time the CAS operation was executed. Thus, if prev_seq is returned, the
+    // operation was successful - otherwise we need to retry. Saving the
+    // return value saves us a load on retry.
+    prev_seq = rtc::AtomicOps::CompareAndSwap(&transport_seq_, desired_prev_seq,
+                                              new_seq);
+  } while (prev_seq != desired_prev_seq);
+
+  return new_seq;
+}
+
 }  // namespace webrtc