Introduce RtpStreamReceiver and RtpStreamReceiverControllerInterface.

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 16 matching lines @@
 #include "webrtc/base/logging.h"
 #include "webrtc/base/optional.h"
 #include "webrtc/base/ptr_util.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_impl.h"
-#include "webrtc/call/rtp_demuxer.h"
+#include "webrtc/call/rtp_transport_controller_receive.h"
 #include "webrtc/call/rtp_transport_controller_send.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/receive_side_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"
@@ skipping 158 matching lines @@
   // Audio, Video, and FlexFEC receive streams are owned by the client that
   // creates them.
   std::set<AudioReceiveStream*> audio_receive_streams_
       GUARDED_BY(receive_crit_);
   std::set<VideoReceiveStream*> video_receive_streams_
       GUARDED_BY(receive_crit_);
 
   std::map<std::string, AudioReceiveStream*> sync_stream_mapping_
       GUARDED_BY(receive_crit_);
 
-  // TODO(nisse): Should eventually be part of injected
-  // RtpTransportControllerReceive, with a single demuxer in the bundled case.
-  RtpDemuxer audio_rtp_demuxer_ GUARDED_BY(receive_crit_);
-  RtpDemuxer video_rtp_demuxer_ GUARDED_BY(receive_crit_);
+  // TODO(nisse): Should eventually be injected at creation,
+  // with a single object in the bundled case.
+  RtpTransportControllerReceive audio_transport_receive_
+      GUARDED_BY(receive_crit_);
+  RtpTransportControllerReceive video_transport_receive_
+      GUARDED_BY(receive_crit_);
 
   // This extra map is used for receive processing which is
   // independent of media type.
 
   // TODO(nisse): In the RTP transport refactoring, we should have a
   // single mapping from ssrc to a more abstract receive stream, with
   // accessor methods for all configuration we need at this level.
   struct ReceiveRtpConfig {
     ReceiveRtpConfig() = default;  // Needed by std::map
     ReceiveRtpConfig(const std::vector<RtpExtension>& extensions,
@@ skipping 322 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(transport_send_->packet_router(), config,
-                             config_.audio_state, event_log_);
+  AudioReceiveStream* receive_stream;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    audio_rtp_demuxer_.AddSink(config.rtp.remote_ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+
+    // TODO(nisse): Make RtpTransportControllerReceive thread safe. But it's
+    // unfortunate if we need additional locks for each packet processed.
+    receive_stream = new AudioReceiveStream(
+        &audio_transport_receive_, transport_send_->packet_router(), config,
+        config_.audio_state, event_log_);
     receive_rtp_config_[config.rtp.remote_ssrc] =
         ReceiveRtpConfig(config.rtp.extensions, UseSendSideBwe(config));
     audio_receive_streams_.insert(receive_stream);
 
     ConfigureSync(config.sync_group);
   }
   {
     ReadLockScoped read_lock(*send_crit_);
     auto it = audio_send_ssrcs_.find(config.rtp.local_ssrc);
     if (it != audio_send_ssrcs_.end()) {
@@ skipping 11 matching lines @@
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   RTC_DCHECK(receive_stream != nullptr);
   webrtc::internal::AudioReceiveStream* audio_receive_stream =
       static_cast<webrtc::internal::AudioReceiveStream*>(receive_stream);
   {
     WriteLockScoped write_lock(*receive_crit_);
     const AudioReceiveStream::Config& config = audio_receive_stream->config();
     uint32_t ssrc = config.rtp.remote_ssrc;
     receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
         ->RemoveStream(ssrc);
-    size_t num_deleted = audio_rtp_demuxer_.RemoveSink(audio_receive_stream);
-    RTC_DCHECK(num_deleted == 1);
     audio_receive_streams_.erase(audio_receive_stream);
     const std::string& sync_group = audio_receive_stream->config().sync_group;
     const auto it = sync_stream_mapping_.find(sync_group);
     if (it != sync_stream_mapping_.end() &&
         it->second == audio_receive_stream) {
       sync_stream_mapping_.erase(it);
       ConfigureSync(sync_group);
     }
     receive_rtp_config_.erase(ssrc);
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete audio_receive_stream;
   }
   UpdateAggregateNetworkState();
-  delete audio_receive_stream;
 }
 
 webrtc::VideoSendStream* Call::CreateVideoSendStream(
     webrtc::VideoSendStream::Config config,
     VideoEncoderConfig encoder_config) {
   TRACE_EVENT0("webrtc", "Call::CreateVideoSendStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   video_send_delay_stats_->AddSsrcs(config);
   event_log_->LogVideoSendStreamConfig(config);
@@ skipping 55 matching lines @@
 
   UpdateAggregateNetworkState();
   delete send_stream_impl;
 }
 
 webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
     webrtc::VideoReceiveStream::Config configuration) {
   TRACE_EVENT0("webrtc", "Call::CreateVideoReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
-  VideoReceiveStream* receive_stream =
-      new VideoReceiveStream(num_cpu_cores_, transport_send_->packet_router(),
-                             std::move(configuration),
-                             module_process_thread_.get(), call_stats_.get());
-
-  const webrtc::VideoReceiveStream::Config& config = receive_stream->config();
-  ReceiveRtpConfig receive_config(config.rtp.extensions,
-                                  UseSendSideBwe(config));
+  VideoReceiveStream* receive_stream;
+  const webrtc::VideoReceiveStream::Config* config;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    video_rtp_demuxer_.AddSink(config.rtp.remote_ssrc, receive_stream);
-    if (config.rtp.rtx_ssrc) {
-      video_rtp_demuxer_.AddSink(config.rtp.rtx_ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    receive_stream = new VideoReceiveStream(
+        &video_transport_receive_, num_cpu_cores_,
+        transport_send_->packet_router(), std::move(configuration),
+        module_process_thread_.get(), call_stats_.get());
+
+    config = &receive_stream->config();
+    ReceiveRtpConfig receive_config(config->rtp.extensions,
+                                    UseSendSideBwe(*config));
+    if (config->rtp.rtx_ssrc) {
       // We record identical config for the rtx stream as for the main
       // stream. Since the transport_send_cc negotiation is per payload
       // type, we may get an incorrect value for the rtx stream, but
       // that is unlikely to matter in practice.
-      receive_rtp_config_[config.rtp.rtx_ssrc] = receive_config;
+      receive_rtp_config_[config->rtp.rtx_ssrc] = receive_config;
     }
-    receive_rtp_config_[config.rtp.remote_ssrc] = receive_config;
+    receive_rtp_config_[config->rtp.remote_ssrc] = receive_config;
     video_receive_streams_.insert(receive_stream);
-    ConfigureSync(config.sync_group);
+    ConfigureSync(config->sync_group);
   }
   receive_stream->SignalNetworkState(video_network_state_);
   UpdateAggregateNetworkState();
-  event_log_->LogVideoReceiveStreamConfig(config);
+  event_log_->LogVideoReceiveStreamConfig(*config);
   return receive_stream;
 }
 
 void Call::DestroyVideoReceiveStream(
     webrtc::VideoReceiveStream* receive_stream) {
   TRACE_EVENT0("webrtc", "Call::DestroyVideoReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
   RTC_DCHECK(receive_stream != nullptr);
   VideoReceiveStream* receive_stream_impl =
       static_cast<VideoReceiveStream*>(receive_stream);
   const VideoReceiveStream::Config& config = receive_stream_impl->config();
   {
     WriteLockScoped write_lock(*receive_crit_);
     // Remove all ssrcs pointing to a receive stream. As RTX retransmits on a
     // separate SSRC there can be either one or two.
-    size_t num_deleted = video_rtp_demuxer_.RemoveSink(receive_stream_impl);
-    RTC_DCHECK_GE(num_deleted, 1);
     receive_rtp_config_.erase(config.rtp.remote_ssrc);
     if (config.rtp.rtx_ssrc) {
       receive_rtp_config_.erase(config.rtp.rtx_ssrc);
     }
     video_receive_streams_.erase(receive_stream_impl);
     ConfigureSync(config.sync_group);
+
+    receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
+        ->RemoveStream(config.rtp.remote_ssrc);
+
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete receive_stream_impl;
   }
-
-  receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
-      ->RemoveStream(config.rtp.remote_ssrc);
-
   UpdateAggregateNetworkState();
-  delete receive_stream_impl;
 }
 
 FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
     const FlexfecReceiveStream::Config& config) {
   TRACE_EVENT0("webrtc", "Call::CreateFlexfecReceiveStream");
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   RecoveredPacketReceiver* recovered_packet_receiver = this;
-  FlexfecReceiveStreamImpl* receive_stream = new FlexfecReceiveStreamImpl(
-      config, recovered_packet_receiver, call_stats_->rtcp_rtt_stats(),
-      module_process_thread_.get());
 
+  FlexfecReceiveStreamImpl* receive_stream;
   {
     WriteLockScoped write_lock(*receive_crit_);
-    video_rtp_demuxer_.AddSink(config.remote_ssrc, receive_stream);
-
-    for (auto ssrc : config.protected_media_ssrcs)
-      video_rtp_demuxer_.AddSink(ssrc, receive_stream);
+    // Constructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    receive_stream = new FlexfecReceiveStreamImpl(
+        &video_transport_receive_, config, recovered_packet_receiver,
+        call_stats_->rtcp_rtt_stats(), module_process_thread_.get());
 
     RTC_DCHECK(receive_rtp_config_.find(config.remote_ssrc) ==
                receive_rtp_config_.end());
     receive_rtp_config_[config.remote_ssrc] =
         ReceiveRtpConfig(config.rtp_header_extensions, UseSendSideBwe(config));
   }
 
   // TODO(brandtr): Store config in RtcEventLog here.
 
   return receive_stream;
@@ skipping 11 matching lines @@
   {
     WriteLockScoped write_lock(*receive_crit_);
 
     const FlexfecReceiveStream::Config& config =
         receive_stream_impl->GetConfig();
     uint32_t ssrc = config.remote_ssrc;
     receive_rtp_config_.erase(ssrc);
 
     // Remove all SSRCs pointing to the FlexfecReceiveStreamImpl to be
     // destroyed.
-    video_rtp_demuxer_.RemoveSink(receive_stream_impl);
     receive_side_cc_.GetRemoteBitrateEstimator(UseSendSideBwe(config))
         ->RemoveStream(ssrc);
+
+    // Destructor is expected to register with audio_transport_receive_,
+    // and hence needs to be called while holding the lock.
+    delete 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;
   transport_send_->send_side_cc()->GetBitrateController()->AvailableBandwidth(
@@ skipping 346 matching lines @@
   // on parsed_packet to the receive streams.
   rtc::Optional<RtpPacketReceived> parsed_packet =
       ParseRtpPacket(packet, length, &packet_time);
 
   if (!parsed_packet)
     return DELIVERY_PACKET_ERROR;
 
   NotifyBweOfReceivedPacket(*parsed_packet, media_type);
 
   if (media_type == MediaType::AUDIO) {
-    if (audio_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
+    if (audio_transport_receive_.OnRtpPacket(*parsed_packet)) {
       received_bytes_per_second_counter_.Add(static_cast<int>(length));
       received_audio_bytes_per_second_counter_.Add(static_cast<int>(length));
       event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
       return DELIVERY_OK;
     }
   } else if (media_type == MediaType::VIDEO) {
-    if (video_rtp_demuxer_.OnRtpPacket(*parsed_packet)) {
+    if (video_transport_receive_.OnRtpPacket(*parsed_packet)) {
       received_bytes_per_second_counter_.Add(static_cast<int>(length));
       received_video_bytes_per_second_counter_.Add(static_cast<int>(length));
       event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
       return DELIVERY_OK;
     }
   }
   return DELIVERY_UNKNOWN_SSRC;
 }
 
 PacketReceiver::DeliveryStatus Call::DeliverPacket(
@@ skipping 15 matching lines @@
 // audio packets with FlexFEC.
 void Call::OnRecoveredPacket(const uint8_t* packet, size_t length) {
   ReadLockScoped read_lock(*receive_crit_);
   rtc::Optional<RtpPacketReceived> parsed_packet =
       ParseRtpPacket(packet, length, nullptr);
   if (!parsed_packet)
     return;
 
   parsed_packet->set_recovered(true);
 
-  video_rtp_demuxer_.OnRtpPacket(*parsed_packet);
+  video_transport_receive_.OnRtpPacket(*parsed_packet);

[pthatcher1, 2017/05/17 23:24:53]

What is the advantage of working with the wrapper around the demuxer rather than
just working with the demuxer?

[nisse-webrtc, 2017/05/18 08:45:43]

At the moment, little. Responsibility for parsing should be moved to this
wrapper, but before that we need to think about responsibility for the
extensions mapping.

It would also make sense to let the wrapper object do the needed locking (while
the demuxer object itself should b a simple class unaware of multithreading
issues). Do you agree? 

[danilchap, 2017/05/19 15:37:43]

using ReceiverController feels preliminary: it should be able handle recovering
packets itself.

[nisse-webrtc, 2017/05/22 07:09:37]

Agreed. I think we should aim to move all per-packet processing out of Call, and
connect RtpTransportController directly to the packet transport. But there's
many steps to get there.

 }
 
 void Call::NotifyBweOfReceivedPacket(const RtpPacketReceived& packet,
                                      MediaType media_type) {
   auto it = receive_rtp_config_.find(packet.Ssrc());
   bool use_send_side_bwe =
       (it != receive_rtp_config_.end()) && it->second.use_send_side_bwe;
 
   RTPHeader header;
   packet.GetHeader(&header);
@@ skipping 13 matching lines @@
       (use_send_side_bwe && header.extension.hasTransportSequenceNumber)) {
     receive_side_cc_.OnReceivedPacket(
         packet.arrival_time_ms(), packet.payload_size() + packet.padding_size(),
         header);
   }
 }
 
 }  // namespace internal
 
 }  // namespace webrtc