Parse FlexFEC RTP headers in Call and add integration with BWE.

webrtc/call/call.cc

Index: webrtc/call/call.cc
diff --git a/webrtc/call/call.cc b/webrtc/call/call.cc
index 20daba88dfee9f71e9e57ea42801c01b3945fbd2..70b537c87e796d65a07a7c98b3c56a59906bf2f1 100644
--- a/webrtc/call/call.cc
+++ b/webrtc/call/call.cc
@@ -24,6 +24,7 @@
 #include "webrtc/base/checks.h"
 #include "webrtc/base/constructormagic.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/base/optional.h"
 #include "webrtc/base/task_queue.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/base/thread_checker.h"
@@ -39,6 +40,9 @@
 #include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_header_parser.h"
 #include "webrtc/modules/rtp_rtcp/source/byte_io.h"
+#include "webrtc/modules/rtp_rtcp/source/rtp_header_extension.h"
+#include "webrtc/modules/rtp_rtcp/source/rtp_header_extensions.h"
+#include "webrtc/modules/rtp_rtcp/source/rtp_packet_received.h"
 #include "webrtc/modules/utility/include/process_thread.h"
 #include "webrtc/system_wrappers/include/clock.h"
 #include "webrtc/system_wrappers/include/cpu_info.h"
@@ -107,6 +111,8 @@ class Call : public webrtc::Call,
   // Implements RecoveredPacketReceiver.
   bool OnRecoveredPacket(const uint8_t* packet, size_t length) override;
 
+  void NotifyBweOfReceivedPacket(const RtpPacketReceived& packet);
+
   void SetBitrateConfig(
       const webrtc::Call::Config::BitrateConfig& bitrate_config) override;
 
@@ -154,6 +160,11 @@ class Call : public webrtc::Call,
       return nullptr;
   }
 
+  rtc::Optional<RtpPacketReceived> ParseRtpPacket(const uint8_t* packet,
+                                                  size_t length,
+                                                  const PacketTime& packet_time)
+      SHARED_LOCKS_REQUIRED(receive_crit_);
+
   void UpdateSendHistograms() EXCLUSIVE_LOCKS_REQUIRED(&bitrate_crit_);
   void UpdateReceiveHistograms();
   void UpdateHistograms();
@@ -192,6 +203,14 @@ class Call : public webrtc::Call,
   std::map<std::string, AudioReceiveStream*> sync_stream_mapping_
       GUARDED_BY(receive_crit_);
 
+  // Registered RTP header extensions for each stream.
+  // Note that RTP header extensions are negotiated per track ("m= line") in the
+  // SDP, but we have no notion of tracks at the Call level. We therefore store
+  // the RTP header extensions per SSRC instead, which leads to some storage
+  // overhead.
+  std::map<uint32_t, RtpHeaderExtensionMap> received_rtp_header_extensions_
+      GUARDED_BY(receive_crit_);
+
   std::unique_ptr<RWLockWrapper> send_crit_;
   // Audio and Video send streams are owned by the client that creates them.
   std::map<uint32_t, AudioSendStream*> audio_send_ssrcs_ GUARDED_BY(send_crit_);
@@ -226,6 +245,10 @@ class Call : public webrtc::Call,
   // issues with GetRemoteBitrateEstimator (and maybe others).
   const std::unique_ptr<CongestionController> congestion_controller_;
   const std::unique_ptr<SendDelayStats> video_send_delay_stats_;
+  // Cached results from congestion_controller_->GetRemoteBitrateEstimator.
+  // Currently only used by FlexFEC in ParseRtpPacket.
+  RemoteBitrateEstimator* const remote_bitrate_estimator_;
+  RemoteBitrateEstimator* const remote_estimator_proxy_;  // Send-side BWE.
   const int64_t start_ms_;
   // TODO(perkj): |worker_queue_| is supposed to replace
   // |module_process_thread_|.
@@ -284,6 +307,10 @@ Call::Call(const Call::Config& config)
                                                       event_log_,
                                                       &packet_router_)),
       video_send_delay_stats_(new SendDelayStats(clock_)),
+      remote_bitrate_estimator_(
+          congestion_controller_->GetRemoteBitrateEstimator(false)),
+      remote_estimator_proxy_(
+          congestion_controller_->GetRemoteBitrateEstimator(true)),
       start_ms_(clock_->TimeInMilliseconds()),
       worker_queue_("call_worker_queue") {
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
@@ -345,6 +372,39 @@ Call::~Call() {
   Trace::ReturnTrace();
 }
 
+rtc::Optional<RtpPacketReceived> Call::ParseRtpPacket(
+    const uint8_t* packet,
+    size_t length,
+    const PacketTime& packet_time) {
+  uint32_t ssrc = ByteReader<uint32_t>::ReadBigEndian(&packet[8]);

[danilchap, 2016/12/14 13:39:02]

there is no need to hacky-extract ssrc before parsing,
There is IdentifyExtensions function exactly for this usecase:

RtpPacketReceived parsed_packet(nullptr);
if (!parsed_packet.Parse(packet, length)) return;
auto it = extensions_.find(parsed_packet->Ssrc());
if (it != extensions_.end()) {
  parsed_packet.IdentifyExtensions(&it->second);
}

[brandtr, 2016/12/14 14:08:31]

Great! Then I don't need all the stuff with the pointer below.

+
+  std::unique_ptr<RtpPacketReceived> parsed_packet;
+  auto it = received_rtp_header_extensions_.find(ssrc);
+  if (it != received_rtp_header_extensions_.end()) {
+    // The existence of |it->second| is guaranteed during the lifetime of

[danilchap, 2016/12/14 13:39:02]

if it is a problem, file a bug and assign it to me. I'll then adjust RtpPacket
not to hold pointer to the RtpHeaderExtensionMap.

[brandtr, 2016/12/14 14:08:31]

Shouldn't be a problem right now, due to the lock and the implementation of
rtp::Packet. It is a little brittle to rely on that for the future though, so if
it isn't too much work I think it would be nicer if IdentifyExtensions didn't
store the supplied pointer internally. (I think it's only used in the Parse and
IdentifyExtensions methods?)

+    // |parsed_packet| and its copies, due to us holding the
+    // |receive_crit_| lock in the DeliverRtp method.
+    parsed_packet =
+        std::unique_ptr<RtpPacketReceived>(new RtpPacketReceived(&it->second));
+  } else {
+    parsed_packet =
+        std::unique_ptr<RtpPacketReceived>(new RtpPacketReceived(nullptr));
+  }
+
+  if (!parsed_packet->Parse(packet, length))
+    return rtc::Optional<RtpPacketReceived>();
+
+  int64_t arrival_time_ms;
+  if (packet_time.timestamp != -1) {
+    arrival_time_ms = (packet_time.timestamp + 500) / 1000;
+  } else {
+    arrival_time_ms = clock_->TimeInMilliseconds();
+  }
+  parsed_packet->set_arrival_time_ms(arrival_time_ms);
+
+  return rtc::Optional<RtpPacketReceived>(*parsed_packet);
+}
+
 void Call::UpdateHistograms() {
   RTC_HISTOGRAM_COUNTS_100000(
       "WebRTC.Call.LifetimeInSeconds",
@@ -659,18 +719,33 @@ webrtc::FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
     const webrtc::FlexfecReceiveStream::Config& config) {
   TRACE_EVENT0("webrtc", "Call::CreateFlexfecReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
-  FlexfecReceiveStream* receive_stream = new FlexfecReceiveStream(config, this);
+
+  RecoveredPacketReceiver* recovered_packet_receiver = this;
+  FlexfecReceiveStream* receive_stream =
+      new FlexfecReceiveStream(config, recovered_packet_receiver);
 
   {
     WriteLockScoped write_lock(*receive_crit_);
+
+    RTC_DCHECK(flexfec_receive_streams_.find(receive_stream) ==
+               flexfec_receive_streams_.end());
+    flexfec_receive_streams_.insert(receive_stream);
+
     for (auto ssrc : config.protected_media_ssrcs)
       flexfec_receive_ssrcs_media_.insert(std::make_pair(ssrc, receive_stream));
+
     RTC_DCHECK(flexfec_receive_ssrcs_protection_.find(config.remote_ssrc) ==
                flexfec_receive_ssrcs_protection_.end());
     flexfec_receive_ssrcs_protection_[config.remote_ssrc] = receive_stream;
-    flexfec_receive_streams_.insert(receive_stream);
+
+    RTC_DCHECK(received_rtp_header_extensions_.find(config.remote_ssrc) ==
+               received_rtp_header_extensions_.end());
+    RtpHeaderExtensionMap rtp_header_extensions(config.rtp_header_extensions);
+    received_rtp_header_extensions_[config.remote_ssrc] = rtp_header_extensions;
   }
+
   // TODO(brandtr): Store config in RtcEventLog here.
+
   return receive_stream;
 }
 
@@ -678,21 +753,21 @@ void Call::DestroyFlexfecReceiveStream(
     webrtc::FlexfecReceiveStream* receive_stream) {
   TRACE_EVENT0("webrtc", "Call::DestroyFlexfecReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
+
   RTC_DCHECK(receive_stream != nullptr);
+
   // There exist no other derived classes of webrtc::FlexfecReceiveStream,
   // so this downcast is safe.
   FlexfecReceiveStream* receive_stream_impl =
       static_cast<FlexfecReceiveStream*>(receive_stream);
+
   {
     WriteLockScoped write_lock(*receive_crit_);
+
+    uint32_t ssrc = receive_stream_impl->GetConfig().remote_ssrc;
+    received_rtp_header_extensions_.erase(ssrc);
+
     // Remove all SSRCs pointing to the FlexfecReceiveStream to be destroyed.
-    auto media_it = flexfec_receive_ssrcs_media_.begin();
-    while (media_it != flexfec_receive_ssrcs_media_.end()) {
-      if (media_it->second == receive_stream_impl)
-        media_it = flexfec_receive_ssrcs_media_.erase(media_it);
-      else
-        ++media_it;
-    }
     auto prot_it = flexfec_receive_ssrcs_protection_.begin();
     while (prot_it != flexfec_receive_ssrcs_protection_.end()) {
       if (prot_it->second == receive_stream_impl)
@@ -700,8 +775,17 @@ void Call::DestroyFlexfecReceiveStream(
       else
         ++prot_it;
     }
+    auto media_it = flexfec_receive_ssrcs_media_.begin();
+    while (media_it != flexfec_receive_ssrcs_media_.end()) {
+      if (media_it->second == receive_stream_impl)
+        media_it = flexfec_receive_ssrcs_media_.erase(media_it);
+      else
+        ++media_it;
+    }
+
     flexfec_receive_streams_.erase(receive_stream_impl);
   }
+
   delete receive_stream_impl;
 }
 
@@ -1075,13 +1159,21 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
     if (it != video_receive_ssrcs_.end()) {
       received_bytes_per_second_counter_.Add(static_cast<int>(length));
       received_video_bytes_per_second_counter_.Add(static_cast<int>(length));
+      // TODO(brandtr): Notify the BWE of received media packets here.
       auto status = it->second->DeliverRtp(packet, length, packet_time)
                         ? DELIVERY_OK
                         : DELIVERY_PACKET_ERROR;
-      // Deliver media packets to FlexFEC subsystem.
-      auto it_bounds = flexfec_receive_ssrcs_media_.equal_range(ssrc);
-      for (auto it = it_bounds.first; it != it_bounds.second; ++it)
-        it->second->AddAndProcessReceivedPacket(packet, length);
+      // Deliver media packets to FlexFEC subsystem. RTP header extensions need
+      // not be parsed, as FlexFEC is oblivious to the semantic meaning of the
+      // packet contents beyond the 12 byte RTP base header. The BWE is fed
+      // information about these media packets from the regular media pipeline.
+      rtc::Optional<RtpPacketReceived> parsed_packet =
+          ParseRtpPacket(packet, length, packet_time);
+      if (parsed_packet) {
+        auto it_bounds = flexfec_receive_ssrcs_media_.equal_range(ssrc);
+        for (auto it = it_bounds.first; it != it_bounds.second; ++it)
+          it->second->AddAndProcessReceivedPacket(*parsed_packet);
+      }
       if (status == DELIVERY_OK)
         event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
       return status;
@@ -1090,12 +1182,17 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
   if (media_type == MediaType::ANY || media_type == MediaType::VIDEO) {
     auto it = flexfec_receive_ssrcs_protection_.find(ssrc);
     if (it != flexfec_receive_ssrcs_protection_.end()) {
-      auto status = it->second->AddAndProcessReceivedPacket(packet, length)
-                        ? DELIVERY_OK
-                        : DELIVERY_PACKET_ERROR;
-      if (status == DELIVERY_OK)
-        event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
-      return status;
+      rtc::Optional<RtpPacketReceived> parsed_packet =
+          ParseRtpPacket(packet, length, packet_time);
+      if (parsed_packet) {
+        NotifyBweOfReceivedPacket(*parsed_packet);
+        auto status = it->second->AddAndProcessReceivedPacket(*parsed_packet)
+                          ? DELIVERY_OK
+                          : DELIVERY_PACKET_ERROR;
+        if (status == DELIVERY_OK)
+          event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
+        return status;
+      }
     }
   }
   return DELIVERY_UNKNOWN_SSRC;
@@ -1127,5 +1224,31 @@ bool Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
   return it->second->OnRecoveredPacket(packet, length);
 }
 
+void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet) {
+  const bool transport_wide = packet.HasExtension<TransportSequenceNumber>();
+  const bool abs_send_time = packet.HasExtension<AbsoluteSendTime>();
+  const bool t_offset = packet.HasExtension<TransmissionOffset>();
+
+  // At least one of the header extensions is needed for the BWE.
+  if (!transport_wide && !abs_send_time && !t_offset)
+    return;
+
+  RTPHeader header;
+  packet.GetHeader(&header);
+
+  // Send-side BWE?
+  if (remote_estimator_proxy_ && transport_wide) {
+    remote_estimator_proxy_->IncomingPacket(packet.arrival_time_ms(),
+                                            packet.payload_size(), header);
+    return;
+  }
+
+  // Other BWE.

[danilchap, 2016/12/14 13:39:02]

may be 'Receive-side' instead of 'Other'

[brandtr, 2016/12/14 14:08:31]

Done.

+  if (remote_bitrate_estimator_) {
+    remote_bitrate_estimator_->IncomingPacket(packet.arrival_time_ms(),
+                                              packet.payload_size(), header);
+  }
+}
+
 }  // namespace internal
 }  // namespace webrtc