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.
  */
 
@@ skipping 20 matching lines @@
 #include "webrtc/call/bitrate_allocator.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_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"
 #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;
   }
 
+  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 may lead to some
+  // 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 41 matching lines @@
   ss << "pacer_delay_ms: " << pacer_delay_ms << ", ";
   ss << "rtt_ms: " << rtt_ms;
   ss << '}';
   return ss.str();
 }
 
 Call* Call::Create(const Call::Config& config) {
   return new internal::Call(config);
 }
 
+namespace {
+
+RtpHeaderExtensionMap CreateRtpHeaderExtensionMap(

[danilchap, 2016/12/06 15:04:30]

this function implemented as a constructor of the RtpHeaderExtensionMap.
i.e. instead of
CreateRtpHeaderExtensionMap(extensions)
it should be possible to use
RtpHeaderExtensionMap(extensions)

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

Great!

+    const std::vector<RtpExtension>& rtp_header_extensions) {
+  RtpHeaderExtensionMap map;
+  for (const auto& extension : rtp_header_extensions) {
+    if (extension.uri == AbsoluteSendTime::kUri) {
+      map.Register<AbsoluteSendTime>(extension.id);
+    } else if (extension.uri == AudioLevel::kUri) {
+      map.Register<AudioLevel>(extension.id);
+    } else if (extension.uri == TransmissionOffset::kUri) {
+      map.Register<TransmissionOffset>(extension.id);
+    } else if (extension.uri == TransportSequenceNumber::kUri) {
+      map.Register<TransportSequenceNumber>(extension.id);
+    } else if (extension.uri == VideoOrientation::kUri) {
+      map.Register<VideoOrientation>(extension.id);
+    } else if (extension.uri == PlayoutDelayLimits::kUri) {
+      map.Register<PlayoutDelayLimits>(extension.id);
+    }
+  }
+  return map;
+}
+
+}  // namespace
+
 namespace internal {
 
 Call::Call(const Call::Config& config)
     : clock_(Clock::GetRealTimeClock()),
       num_cpu_cores_(CpuInfo::DetectNumberOfCores()),
       module_process_thread_(ProcessThread::Create("ModuleProcessThread")),
       pacer_thread_(ProcessThread::Create("PacerThread")),
       call_stats_(new CallStats(clock_)),
       bitrate_allocator_(new BitrateAllocator(this)),
       config_(config),
@@ skipping 72 matching lines @@
   {
     rtc::CritScope lock(&bitrate_crit_);
     UpdateSendHistograms();
   }
   UpdateReceiveHistograms();
   UpdateHistograms();
 
   Trace::ReturnTrace();
 }
 
+RtpPacketReceived Call::ParseRtpPacket(
+    const uint8_t* packet,
+    size_t length,
+    const PacketTime& packet_time,
+    const RtpHeaderExtensionMap* rtp_header_extensions) {
+  RtpPacketReceived parsed_packet(nullptr);
+  parsed_packet.Parse(packet, length);

[danilchap, 2016/12/06 15:04:30]

what if Parse fails?

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

Fixed by using optional.

+  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 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 117 matching lines @@
   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_);
+      congestion_controller_->GetRemoteBitrateEstimator(
+          CongestionController::UseSendSideBwe(
+              config.rtp.transport_cc,
+              CreateRtpHeaderExtensionMap(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 =
+      CreateRtpHeaderExtensionMap(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] =
+        std::move(rtp_header_extensions);

[danilchap, 2016/12/06 15:04:30]

do not move:
RtpHeaderExtensionMap doesn't have a move constructor because copy is as
expensive/cheap as move.

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

Done.

   }
+
   // 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->config().remote_ssrc;
+
   {
     WriteLockScoped write_lock(*receive_crit_);
-    // 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;
-    }
+
+    received_rtp_header_extensions_.erase(ssrc);
+
     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.
+      // Deliver media packets to FlexFEC subsystem. RTP header extensions need
+      // not be parsed, as FlexFEC is oblivious to the semantic meaning of the

[danilchap, 2016/12/06 15:04:30]

what about BWE header extensions?

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

Extended comment to why this is not relevant for media packets.

+      // payload.
+      RtpPacketReceived parsed_packet =
+          ParseRtpPacket(packet, length, packet_time, nullptr);
       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);
+        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;
+      const RtpHeaderExtensionMap& extensions =
+          received_rtp_header_extensions_[ssrc];
+      RtpPacketReceived parsed_packet =
+          ParseRtpPacket(packet, length, packet_time, &extensions);
+      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,
@@ skipping 16 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