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..86bee67e4e5e2544614eb41b1a28cbd9c9b1432d 100644
--- a/webrtc/modules/pacing/packet_router.cc
+++ b/webrtc/modules/pacing/packet_router.cc
@@ -10,52 +10,151 @@
 
 #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(&ssrc_lookup_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(&ssrc_lookup_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::Load(&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);
+  RtpRtcp* rtp_module = nullptr;
+  {
+    rtc::CritScope cs(&ssrc_lookup_lock_);
+    UpdateModuleMap();
+    auto it = rtp_modules_.find(ssrc);
+    if (it == rtp_modules_.end())
+      return true;
+    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;
+  std::vector<RtpRtcp*> modules;
+  {
+    rtc::CritScope cs(&ssrc_lookup_lock_);
+    for (auto it : rtp_modules_)
+      modules.push_back(it.second);
+  }
+  for (auto it : modules) {
+    if (it->SendingMedia()) {
+      size_t bytes_sent =
+          it->TimeToSendPadding(bytes_to_send - total_bytes_sent);
+      if (bytes_sent > 0)
+        ++transport_seq_;

[stefan-webrtc, 2015/07/22 10:49:00]

This isn't being used anywhere? Seems like we should pass it into the rtp module
somewhere?

Not knowing the details about transport_seq_, why don't you have to use
AtomicOps to modify it here?

[sprang_webrtc, 2015/07/22 15:11:32]

That's a bug, that my test case evidently didn't catch. Will make sure it does.

All numbers should be assigned from the outside by calling
AllocateSequenceNumber().

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

Acknowledged.

+      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::Store(&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;
+
+  return static_cast<uint16_t>(rtc::AtomicOps::Increment(&transport_seq_) &
+                               0xFFFF);

[stefan-webrtc, 2015/07/22 10:49:00]

Isn't this implicit from the cast?

[sprang_webrtc, 2015/07/22 15:11:32]

The static_cast, yes. This was a remnant from when I did a DCHECK here.
I've also updated it to not begin counting backwards after 2^31 packets...

+}
+
+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