Integrate FlexfecReceiveStream with Call.

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 <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/task_queue.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/base/thread_checker.h"
 #include "webrtc/base/trace_event.h"
 #include "webrtc/call.h"
 #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/utility/include/process_thread.h"
 #include "webrtc/system_wrappers/include/clock.h"
@@ skipping 18 matching lines @@
 
 class Call : public webrtc::Call,
              public PacketReceiver,
              public RecoveredPacketReceiver,
              public CongestionController::Observer,
              public BitrateAllocator::LimitObserver {
  public:
   explicit Call(const Call::Config& config);
   virtual ~Call();
 
+  // Implements webrtc::Call.
   PacketReceiver* Receiver() override;
 
   webrtc::AudioSendStream* CreateAudioSendStream(
       const webrtc::AudioSendStream::Config& config) override;
   void DestroyAudioSendStream(webrtc::AudioSendStream* send_stream) override;
 
   webrtc::AudioReceiveStream* CreateAudioReceiveStream(
       const webrtc::AudioReceiveStream::Config& config) override;
   void DestroyAudioReceiveStream(
       webrtc::AudioReceiveStream* receive_stream) override;
 
   webrtc::VideoSendStream* CreateVideoSendStream(
       webrtc::VideoSendStream::Config config,
       VideoEncoderConfig encoder_config) override;
   void DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) override;
 
   webrtc::VideoReceiveStream* CreateVideoReceiveStream(
       webrtc::VideoReceiveStream::Config configuration) override;
   void DestroyVideoReceiveStream(
       webrtc::VideoReceiveStream* receive_stream) override;
 
+  webrtc::FlexfecReceiveStream* CreateFlexfecReceiveStream(
+      webrtc::FlexfecReceiveStream::Config configuration) override;
+  void DestroyFlexfecReceiveStream(
+      webrtc::FlexfecReceiveStream* receive_stream) override;
+
   Stats GetStats() const override;
 
+  // Implements PacketReceiver.
   DeliveryStatus DeliverPacket(MediaType media_type,
                                const uint8_t* packet,
                                size_t length,
                                const PacketTime& packet_time) override;
 
   // Implements RecoveredPacketReceiver.
   bool OnRecoveredPacket(const uint8_t* packet, size_t length) override;
 
   void SetBitrateConfig(
       const webrtc::Call::Config::BitrateConfig& bitrate_config) override;
@@ skipping 51 matching lines @@
   NetworkState video_network_state_;
 
   std::unique_ptr<RWLockWrapper> receive_crit_;
   // Audio and Video receive streams are owned by the client that creates them.
   std::map<uint32_t, AudioReceiveStream*> audio_receive_ssrcs_
       GUARDED_BY(receive_crit_);
   std::map<uint32_t, VideoReceiveStream*> video_receive_ssrcs_
       GUARDED_BY(receive_crit_);
   std::set<VideoReceiveStream*> video_receive_streams_
       GUARDED_BY(receive_crit_);
+  std::multimap<uint32_t, FlexfecReceiveStream*> flexfec_receive_ssrcs_media_

[philipel, 2016/10/19 12:15:34]

Add comment about multiple flex fec streams (possibly) protecting the same media
stream.

[brandtr, 2016/10/19 13:53:25]

Done.

+      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_);
 
   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_;
@@ skipping 398 matching lines @@
       }
     }
     video_receive_streams_.erase(receive_stream_impl);
     RTC_CHECK(receive_stream_impl != nullptr);
     ConfigureSync(receive_stream_impl->config().sync_group);
   }
   UpdateAggregateNetworkState();
   delete receive_stream_impl;
 }
 
+webrtc::FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
+    webrtc::FlexfecReceiveStream::Config configuration) {
+  TRACE_EVENT0("webrtc", "Call::CreateFlexfecReceiveStream");
+  RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
+  FlexfecReceiveStream* receive_stream =
+      new FlexfecReceiveStream(std::move(configuration), this);
+
+  const webrtc::FlexfecReceiveStream::Config& config = receive_stream->config();
+  {
+    WriteLockScoped write_lock(*receive_crit_);
+    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.flexfec_ssrc) ==
+               flexfec_receive_ssrcs_protection_.end());
+    flexfec_receive_ssrcs_protection_[config.flexfec_ssrc] = receive_stream;
+    flexfec_receive_streams_.insert(receive_stream);
+  }
+  // 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);
+  FlexfecReceiveStream* receive_stream_impl = nullptr;

[philipel, 2016/10/19 12:15:34]

|receive_stream| is the |receive_stream_impl| right? I can't see why you need to
"find" the |receive_stream_impl|.

Looking at the DCHECK (631) and assignment (632) that must be true right?

Remove this variable.

[brandtr, 2016/10/19 13:53:25]

They point to the same object, except that the former points to the base class
whereas the latter points to the specific class.

Based on offline discussions, I'm keeping this variable since I need to refer to
the specific type below. The unnecessary DCHECKs have been removed however.

+  {
+    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 ==
+          static_cast<FlexfecReceiveStream*>(receive_stream)) {
+        if (receive_stream_impl != nullptr)
+          RTC_DCHECK(receive_stream_impl == media_it->second);
+        receive_stream_impl = media_it->second;
+        flexfec_receive_ssrcs_media_.erase(media_it++);

[philipel, 2016/10/19 12:15:34]

media_it = flexfec_receive_ssrcs_media_.erase(media_it);

same for loop below.

[brandtr, 2016/10/19 13:53:25]

Done.

+      } else {
+        ++media_it;
+      }
+    }
+    auto prot_it = flexfec_receive_ssrcs_protection_.begin();
+    while (prot_it != flexfec_receive_ssrcs_protection_.end()) {
+      if (prot_it->second ==
+          static_cast<FlexfecReceiveStream*>(receive_stream)) {
+        if (receive_stream_impl != nullptr)
+          RTC_DCHECK(receive_stream_impl == prot_it->second);
+        receive_stream_impl = prot_it->second;
+        flexfec_receive_ssrcs_protection_.erase(prot_it++);
+      } else {
+        ++prot_it;
+      }
+    }
+    flexfec_receive_streams_.erase(receive_stream_impl);

[philipel, 2016/10/19 12:15:34]

erase(receive_stream)

[brandtr, 2016/10/19 13:53:25]

The containers store the specific type pointers, so need to
|receive_stream_impl| here here.

+    RTC_CHECK(receive_stream_impl != nullptr);
+  }
+  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;
   congestion_controller_->GetBitrateController()->AvailableBandwidth(
       &send_bandwidth);
   std::vector<unsigned int> ssrcs;
@@ skipping 325 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;
+      auto flexfec_lower_it = flexfec_receive_ssrcs_media_.lower_bound(ssrc);

[philipel, 2016/10/19 12:15:34]

Use equal_range instead:
http://en.cppreference.com/w/cpp/container/multimap/equal_range

[brandtr, 2016/10/19 13:53:25]

Done.

+      auto flexfec_upper_it = flexfec_receive_ssrcs_media_.upper_bound(ssrc);
+      while (flexfec_lower_it != flexfec_upper_it) {
+        flexfec_lower_it->second->AddAndProcessReceivedPacket(packet, length);
+        ++flexfec_lower_it;
+      }
+      if (status == DELIVERY_OK)

[philipel, 2016/10/19 12:15:34]

if (it->second->DeliverRtp(packet, length, packet_time))) {
  event_log_...
  return DELIVERY_OK;
}

Same for line 1018.

[brandtr, 2016/10/19 13:53:25]

Kept for symmetry with audio/video parts (see above and below). I prefer not to
make changes in those parts.

[philipel, 2016/10/19 14:25:32]

Acknowledged.

+        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);

[terelius, 2016/10/19 09:48:51]

Where are the recovered packets sent by the flexfec stream? Is there any risk
that we log the "same" packet multiple times? E.g. suppose that the sender sends
packets A, B, C and a FEC packet D. If the receiver gets A, B and D, will it log
A, B, D or A, B, D, C?

[brandtr, 2016/10/19 11:29:20]

When FlexFEC recovers a media packet, it is returned to the Call layer, which
then propagates it down to the OnRecoveredPacket method of the appropriate
RtpStreamReceiver. (See https://codereview.webrtc.org/2390823009/.) As far as I
can tell, from there on the recovered packet will never enter the Call layer
again, i.e. the recovered packet will not be logged to the RtcEventLog. (Do we
log packets to the event log anywhere else?)

So the answer is, the receiver will log packets A, B, D.

       return status;
     }
   }
   return DELIVERY_UNKNOWN_SSRC;
 }
 
 PacketReceiver::DeliveryStatus Call::DeliverPacket(
     MediaType media_type,
     const uint8_t* packet,
     size_t length,
@@ skipping 14 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 it->second->OnRecoveredPacket(packet, length);
   return false;
 }
 
 }  // namespace internal
 }  // namespace webrtc