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

webrtc/call/call.cc

 /*
  *  Copyright (c) 2013 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 <string.h>
 #include <algorithm>
 #include <map>
 #include <memory>
 #include <set>
 #include <utility>
 #include <vector>
 
 #include "webrtc/audio/audio_receive_stream.h"
 #include "webrtc/audio/audio_send_stream.h"
 #include "webrtc/audio/audio_state.h"
 #include "webrtc/audio/scoped_voe_interface.h"
 #include "webrtc/base/basictypes.h"
 #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"
 #include "webrtc/base/trace_event.h"
 #include "webrtc/call/bitrate_allocator.h"
 #include "webrtc/call/call.h"
 #include "webrtc/call/flexfec_receive_stream.h"
 #include "webrtc/config.h"
 #include "webrtc/logging/rtc_event_log/rtc_event_log.h"
 #include "webrtc/modules/bitrate_controller/include/bitrate_controller.h"
 #include "webrtc/modules/congestion_controller/include/congestion_controller.h"
 #include "webrtc/modules/pacing/paced_sender.h"
 #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_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"
 #include "webrtc/system_wrappers/include/critical_section_wrapper.h"
 #include "webrtc/system_wrappers/include/metrics.h"
 #include "webrtc/system_wrappers/include/rw_lock_wrapper.h"
 #include "webrtc/system_wrappers/include/trace.h"
 #include "webrtc/video/call_stats.h"
 #include "webrtc/video/send_delay_stats.h"
 #include "webrtc/video/stats_counter.h"
@@ skipping 95 matching lines @@
 
   VoiceEngine* voice_engine() {
     internal::AudioState* audio_state =
         static_cast<internal::AudioState*>(config_.audio_state.get());
     if (audio_state)
       return audio_state->voice_engine();
     else
       return nullptr;
   }
 
+  rtc::Optional<RtpPacketReceived> ParseRtpPacket(
+      const uint8_t* packet,
+      size_t length,
+      const PacketTime& packet_time,
+      const RtpHeaderExtensionMap* rtp_header_extensions);
+
   void UpdateSendHistograms() EXCLUSIVE_LOCKS_REQUIRED(&bitrate_crit_);
   void UpdateReceiveHistograms();
   void UpdateHistograms();
   void UpdateAggregateNetworkState();
 
   Clock* const clock_;
 
   const int num_cpu_cores_;
   const std::unique_ptr<ProcessThread> module_process_thread_;
   const std::unique_ptr<ProcessThread> pacer_thread_;
@@ skipping 18 matching lines @@
   // streams.
   std::multimap<uint32_t, FlexfecReceiveStream*> flexfec_receive_ssrcs_media_
       GUARDED_BY(receive_crit_);
   std::map<uint32_t, FlexfecReceiveStream*> flexfec_receive_ssrcs_protection_
       GUARDED_BY(receive_crit_);
   std::set<FlexfecReceiveStream*> flexfec_receive_streams_
       GUARDED_BY(receive_crit_);
   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_);
   std::map<uint32_t, VideoSendStream*> video_send_ssrcs_ GUARDED_BY(send_crit_);
   std::set<VideoSendStream*> video_send_streams_ GUARDED_BY(send_crit_);
 
   VideoSendStream::RtpStateMap suspended_video_send_ssrcs_;
   webrtc::RtcEventLog* event_log_;
 
   // The following members are only accessed (exclusively) from one thread and
@@ skipping 133 matching lines @@
   {
     rtc::CritScope lock(&bitrate_crit_);
     UpdateSendHistograms();
   }
   UpdateReceiveHistograms();
   UpdateHistograms();
 
   Trace::ReturnTrace();
 }
 
+rtc::Optional<RtpPacketReceived> Call::ParseRtpPacket(
+    const uint8_t* packet,
+    size_t length,
+    const PacketTime& packet_time,
+    const RtpHeaderExtensionMap* rtp_header_extensions) {
+  RtpPacketReceived parsed_packet(rtp_header_extensions);
+  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",
       (clock_->TimeInMilliseconds() - start_ms_) / 1000);
 }
 
 void Call::UpdateSendHistograms() {
   if (first_packet_sent_ms_ == -1)
     return;
   int64_t elapsed_sec =
@@ skipping 116 matching lines @@
   UpdateAggregateNetworkState();
   delete audio_send_stream;
 }
 
 webrtc::AudioReceiveStream* Call::CreateAudioReceiveStream(
     const webrtc::AudioReceiveStream::Config& config) {
   TRACE_EVENT0("webrtc", "Call::CreateAudioReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   event_log_->LogAudioReceiveStreamConfig(config);
   AudioReceiveStream* receive_stream = new AudioReceiveStream(
-      &packet_router_,
-      // TODO(nisse): Used only when UseSendSideBwe(config) is true.
-      congestion_controller_->GetRemoteBitrateEstimator(true), config,
-      config_.audio_state, event_log_);
+      &packet_router_, congestion_controller_->GetRemoteBitrateEstimator(
+                           CongestionController::UseSendSideBwe(

[brandtr, 2016/12/12 13:51:08]

Nisse: please check.

[nisse-webrtc, 2016/12/13 10:32:32]

Maybe the TODO item was badly worded. The thing is that in case
UseSendSideBwe(config) is false, this argument is ignored by the callee (the
AudioReceiveStream constructor), and the call to GetRemoteBitrateEstimator is
useless. But it's no big deal because the call is cheap.

unless you have made some related changes, I think it's still fine to always
pass true to the GetRemoteBitrateEstimator method. And
CongestionController::UseSendSideBwe is an internal function which shouldn't be
used here, and if I remember correctly, it's deleted in one of my cl:s.

[brandtr, 2016/12/13 11:05:49]

Got it, thanks. I've reverted this part of the patch.

CongestionController::UseSendSideBwe is added by this CL :) See
https://codereview.webrtc.org/2553863003/diff/80001/webrtc/modules/congestion....

+                               config.rtp.transport_cc,
+                               RtpHeaderExtensionMap(config.rtp.extensions))),
+      config, config_.audio_state, event_log_);
   {
     WriteLockScoped write_lock(*receive_crit_);
     RTC_DCHECK(audio_receive_ssrcs_.find(config.rtp.remote_ssrc) ==
                audio_receive_ssrcs_.end());
     audio_receive_ssrcs_[config.rtp.remote_ssrc] = receive_stream;
     ConfigureSync(config.sync_group);
   }
   {
     ReadLockScoped read_lock(*send_crit_);
     auto it = audio_send_ssrcs_.find(config.rtp.local_ssrc);
@@ skipping 154 matching lines @@
     ConfigureSync(receive_stream_impl->config().sync_group);
   }
   UpdateAggregateNetworkState();
   delete receive_stream_impl;
 }
 
 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);
+
+  RtpHeaderExtensionMap rtp_header_extensions(config.rtp_header_extensions);
+  RecoveredPacketReceiver* recovered_packet_receiver = this;
+  RemoteBitrateEstimator* remote_bitrate_estimator =
+      congestion_controller_->GetRemoteBitrateEstimator(
+          CongestionController::UseSendSideBwe(config.transport_cc,
+                                               rtp_header_extensions));
+  FlexfecReceiveStream* receive_stream = new FlexfecReceiveStream(
+      config, recovered_packet_receiver, remote_bitrate_estimator);
 
   {
     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());
+    received_rtp_header_extensions_[config.remote_ssrc] = rtp_header_extensions;
   }
+
   // TODO(brandtr): Store config in RtcEventLog here.
+
   return receive_stream;
 }
 
 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);
+  uint32_t ssrc = receive_stream_impl->GetConfig().remote_ssrc;
+
   {
     WriteLockScoped write_lock(*receive_crit_);
+
+    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)
         prot_it = flexfec_receive_ssrcs_protection_.erase(prot_it);
       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;
 }
 
 Call::Stats Call::GetStats() const {
   // TODO(solenberg): Some test cases in EndToEndTest use this from a different
   // thread. Re-enable once that is fixed.
   // RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   Stats stats;
   // Fetch available send/receive bitrates.
   uint32_t send_bandwidth = 0;
@@ skipping 356 matching lines @@
     }
   }
   if (media_type == MediaType::ANY || media_type == MediaType::VIDEO) {
     auto it = video_receive_ssrcs_.find(ssrc);
     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));
       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);
-      if (status == DELIVERY_OK)
-        event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
-      return status;
+      // 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, nullptr);
+      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;
+      }
     }
   }
   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;
+      const RtpHeaderExtensionMap& extensions =
+          received_rtp_header_extensions_[ssrc];
+      rtc::Optional<RtpPacketReceived> parsed_packet =
+          ParseRtpPacket(packet, length, packet_time, &extensions);
+      if (parsed_packet) {
+        auto status =
+            it->second->AddAndProcessReceivedPacket(std::move(*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;
 }
 
 PacketReceiver::DeliveryStatus Call::DeliverPacket(
     MediaType media_type,
     const uint8_t* packet,
     size_t length,
     const PacketTime& packet_time) {
@@ skipping 13 matching lines @@
   uint32_t ssrc = ByteReader<uint32_t>::ReadBigEndian(&packet[8]);
   ReadLockScoped read_lock(*receive_crit_);
   auto it = video_receive_ssrcs_.find(ssrc);
   if (it == video_receive_ssrcs_.end())
     return false;
   return it->second->OnRecoveredPacket(packet, length);
 }
 
 }  // namespace internal
 }  // namespace webrtc