Let FlexfecReceiveStreamImpl send RTCP RRs.

This CL adds an RTP module to FlexfecReceiveStreamImpl, and wires it up
to send RTCP RRs. It further makes some methods take const refs instead
of values, to make it more clear where packet copies are made. This
change reduces the number of copies by one, for the case when media
packets are added to the FlexFEC receiver.

The end-to-end test is modified to check for RTCP RRs being sent.
Part of this modification involves some indentation changes, and the
diff thus looks bigger than it logically is.

BUG=webrtc:5654

Review-Url: https://codereview.webrtc.org/2625633003
Cr-Commit-Position: refs/heads/master@{#16106}

webrtc/call/call.cc

@@ -702,8 +702,9 @@ FlexfecReceiveStream* Call::CreateFlexfecReceiveStream(
   RTC_DCHECK(configuration_thread_checker_.CalledOnValidThread());
 
   RecoveredPacketReceiver* recovered_packet_receiver = this;
-  FlexfecReceiveStreamImpl* receive_stream =
+  FlexfecReceiveStreamImpl* receive_stream = new FlexfecReceiveStreamImpl(
-      new FlexfecReceiveStreamImpl(config, recovered_packet_receiver);
+      config, recovered_packet_receiver, call_stats_->rtcp_rtt_stats(),
+      module_process_thread_.get());
 
   {
     WriteLockScoped write_lock(*receive_crit_);
@@ -1165,10 +1166,9 @@ PacketReceiver::DeliveryStatus Call::DeliverRtp(MediaType media_type,
           ParseRtpPacket(packet, length, packet_time);
       if (parsed_packet) {
         NotifyBweOfReceivedPacket(*parsed_packet);
-        auto status =
+        auto status = it->second->AddAndProcessReceivedPacket(*parsed_packet)
-            it->second->AddAndProcessReceivedPacket(std::move(*parsed_packet))
+                          ? DELIVERY_OK
-                ? DELIVERY_OK
+                          : DELIVERY_PACKET_ERROR;
-                : DELIVERY_PACKET_ERROR;
         if (status == DELIVERY_OK)
           event_log_->LogRtpHeader(kIncomingPacket, media_type, packet, length);
         return status;

webrtc/call/flexfec_receive_stream_impl.cc

@@ -10,10 +10,16 @@
 
 #include "webrtc/call/flexfec_receive_stream_impl.h"
 
-#include <utility>
+#include <string>
 
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
+#include "webrtc/modules/rtp_rtcp/include/receive_statistics.h"
+#include "webrtc/modules/rtp_rtcp/include/rtp_rtcp.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"
 
 namespace webrtc {
 
@@ -97,33 +103,77 @@ std::unique_ptr<FlexfecReceiver> MaybeCreateFlexfecReceiver(
                           recovered_packet_receiver));
 }
 
+std::unique_ptr<RtpRtcp> CreateRtpRtcpModule(
+    ReceiveStatistics* receive_statistics,
+    Transport* rtcp_send_transport,
+    RtcpRttStats* rtt_stats) {
+  RtpRtcp::Configuration configuration;
+  configuration.audio = false;
+  configuration.receiver_only = true;
+  configuration.clock = Clock::GetRealTimeClock();
+  configuration.receive_statistics = receive_statistics;
+  configuration.outgoing_transport = rtcp_send_transport;
+  configuration.rtt_stats = rtt_stats;
+  std::unique_ptr<RtpRtcp> rtp_rtcp(RtpRtcp::CreateRtpRtcp(configuration));
+  return rtp_rtcp;
+}
+
 }  // namespace
 
 FlexfecReceiveStreamImpl::FlexfecReceiveStreamImpl(
     const Config& config,
-    RecoveredPacketReceiver* recovered_packet_receiver)
+    RecoveredPacketReceiver* recovered_packet_receiver,
-    : started_(false),
+    RtcpRttStats* rtt_stats,
-      config_(config),
+    ProcessThread* process_thread)
-      receiver_(
+    : config_(config),
-          MaybeCreateFlexfecReceiver(config_, recovered_packet_receiver)) {
+      started_(false),
+      receiver_(MaybeCreateFlexfecReceiver(config_, recovered_packet_receiver)),
+      rtp_receive_statistics_(
+          ReceiveStatistics::Create(Clock::GetRealTimeClock())),
+      rtp_rtcp_(CreateRtpRtcpModule(rtp_receive_statistics_.get(),
+                                    config_.rtcp_send_transport,
+                                    rtt_stats)),
+      process_thread_(process_thread) {
   LOG(LS_INFO) << "FlexfecReceiveStreamImpl: " << config_.ToString();
+
+  // RTCP reporting.
+  rtp_rtcp_->SetSendingMediaStatus(false);
+  rtp_rtcp_->SetRTCPStatus(config_.rtcp_mode);
+  rtp_rtcp_->SetSSRC(config_.local_ssrc);
+  process_thread_->RegisterModule(rtp_rtcp_.get());
 }
 
 FlexfecReceiveStreamImpl::~FlexfecReceiveStreamImpl() {
   LOG(LS_INFO) << "~FlexfecReceiveStreamImpl: " << config_.ToString();
   Stop();
+  process_thread_->DeRegisterModule(rtp_rtcp_.get());
 }
 
 bool FlexfecReceiveStreamImpl::AddAndProcessReceivedPacket(
-    RtpPacketReceived packet) {
+    const RtpPacketReceived& packet) {
   {
     rtc::CritScope cs(&crit_);
     if (!started_)
       return false;
   }
+
   if (!receiver_)
     return false;
-  return receiver_->AddAndProcessReceivedPacket(std::move(packet));
+
+  if (!receiver_->AddAndProcessReceivedPacket(packet))
+    return false;
+
+  // Do not report media packets in the RTCP RRs generated by |rtp_rtcp_|.
+  if (packet.Ssrc() == config_.remote_ssrc) {
+    RTPHeader header;
+    packet.GetHeader(&header);
+    // FlexFEC packets are never retransmitted.
+    const bool kNotRetransmitted = false;
+    rtp_receive_statistics_->IncomingPacket(header, packet.size(),
+                                            kNotRetransmitted);
+  }
+
+  return true;
 }
 
 void FlexfecReceiveStreamImpl::Start() {

webrtc/call/flexfec_receive_stream_impl.h

@@ -12,25 +12,31 @@
 #define WEBRTC_CALL_FLEXFEC_RECEIVE_STREAM_IMPL_H_
 
 #include <memory>
-#include <string>
 
-#include "webrtc/base/basictypes.h"
 #include "webrtc/base/criticalsection.h"
 #include "webrtc/call/flexfec_receive_stream.h"
-#include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
-#include "webrtc/modules/rtp_rtcp/source/rtp_packet_received.h"
 
 namespace webrtc {
 
+class FlexfecReceiver;
+class ProcessThread;
+class ReceiveStatistics;
+class RecoveredPacketReceiver;
+class RtcpRttStats;
+class RtpPacketReceived;
+class RtpRtcp;
+
 class FlexfecReceiveStreamImpl : public FlexfecReceiveStream {
  public:
   FlexfecReceiveStreamImpl(const Config& config,
-                           RecoveredPacketReceiver* recovered_packet_receiver);
+                           RecoveredPacketReceiver* recovered_packet_receiver,
+                           RtcpRttStats* rtt_stats,
+                           ProcessThread* process_thread);
   ~FlexfecReceiveStreamImpl() override;
 
   const Config& GetConfig() const { return config_; }
 
-  bool AddAndProcessReceivedPacket(RtpPacketReceived packet);
+  bool AddAndProcessReceivedPacket(const RtpPacketReceived& packet);
 
   // Implements FlexfecReceiveStream.
   void Start() override;
@@ -38,11 +44,18 @@ class FlexfecReceiveStreamImpl : public FlexfecReceiveStream {
   Stats GetStats() const override;
 
  private:
-  rtc::CriticalSection crit_;
+  // Config.
-  bool started_ GUARDED_BY(crit_);
-
   const Config config_;
+  bool started_ GUARDED_BY(crit_);
+  rtc::CriticalSection crit_;
+
+  // Erasure code interfacing.
   const std::unique_ptr<FlexfecReceiver> receiver_;
+
+  // RTCP reporting.
+  const std::unique_ptr<ReceiveStatistics> rtp_receive_statistics_;
+  const std::unique_ptr<RtpRtcp> rtp_rtcp_;
+  ProcessThread* process_thread_;
 };
 
 }  // namespace webrtc

webrtc/call/flexfec_receive_stream_unittest.cc

@@ -12,10 +12,13 @@
 
 #include "webrtc/base/array_view.h"
 #include "webrtc/call/flexfec_receive_stream_impl.h"
+#include "webrtc/modules/pacing/packet_router.h"
 #include "webrtc/modules/rtp_rtcp/include/flexfec_receiver.h"
 #include "webrtc/modules/rtp_rtcp/mocks/mock_recovered_packet_receiver.h"
+#include "webrtc/modules/rtp_rtcp/mocks/mock_rtcp_rtt_stats.h"
 #include "webrtc/modules/rtp_rtcp/source/byte_io.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_header_extensions.h"
+#include "webrtc/modules/utility/include/mock/mock_process_thread.h"
 #include "webrtc/test/gmock.h"
 #include "webrtc/test/gtest.h"
 #include "webrtc/test/mock_transport.h"
@@ -74,11 +77,16 @@ class FlexfecReceiveStreamTest : public ::testing::Test {
  protected:
   FlexfecReceiveStreamTest()
       : config_(CreateDefaultConfig(&rtcp_send_transport_)),
-        receive_stream_(config_, &recovered_packet_receiver_) {}
+        receive_stream_(config_,
+                        &recovered_packet_receiver_,
+                        &rtt_stats_,
+                        &process_thread_) {}
 
+  MockTransport rtcp_send_transport_;
   FlexfecReceiveStream::Config config_;
   MockRecoveredPacketReceiver recovered_packet_receiver_;
-  MockTransport rtcp_send_transport_;
+  MockRtcpRttStats rtt_stats_;
+  MockProcessThread process_thread_;
 
   FlexfecReceiveStreamImpl receive_stream_;
 };
@@ -126,7 +134,8 @@ TEST_F(FlexfecReceiveStreamTest, RecoversPacketWhenStarted) {
   // clang-format on
 
   testing::StrictMock<MockRecoveredPacketReceiver> recovered_packet_receiver;
-  FlexfecReceiveStreamImpl receive_stream(config_, &recovered_packet_receiver);
+  FlexfecReceiveStreamImpl receive_stream(config_, &recovered_packet_receiver,
+                                          &rtt_stats_, &process_thread_);
 
   // Do not call back before being started.
   receive_stream.AddAndProcessReceivedPacket(ParsePacket(kFlexfecPacket));

webrtc/media/engine/webrtcvideoengine2.cc

@@ -2219,6 +2219,7 @@ void WebRtcVideoChannel2::WebRtcVideoReceiveStream::SetLocalSsrc(
   }
 
   config_.rtp.local_ssrc = local_ssrc;
+  flexfec_config_.local_ssrc = local_ssrc;
   LOG(LS_INFO)
       << "RecreateWebRtcStream (recv) because of SetLocalSsrc; local_ssrc="
       << local_ssrc;
@@ -2246,6 +2247,7 @@ void WebRtcVideoChannel2::WebRtcVideoReceiveStream::SetFeedbackParameters(
   config_.rtp.nack.rtp_history_ms = nack_history_ms;
   config_.rtp.transport_cc = transport_cc_enabled;
   config_.rtp.rtcp_mode = rtcp_mode;
+  flexfec_config_.rtcp_mode = rtcp_mode;
   LOG(LS_INFO)
       << "RecreateWebRtcStream (recv) because of SetFeedbackParameters; nack="
       << nack_enabled << ", remb=" << remb_enabled

webrtc/modules/rtp_rtcp/include/flexfec_receiver.h

@@ -43,13 +43,13 @@ class FlexfecReceiver {
   // Inserts a received packet (can be either media or FlexFEC) into the
   // internal buffer, and sends the received packets to the erasure code.
   // All newly recovered packets are sent back through the callback.
-  bool AddAndProcessReceivedPacket(RtpPacketReceived packet);
+  bool AddAndProcessReceivedPacket(const RtpPacketReceived& packet);
 
   // Returns a counter describing the added and recovered packets.
   FecPacketCounter GetPacketCounter() const;
 
  private:
-  bool AddReceivedPacket(RtpPacketReceived packet);
+  bool AddReceivedPacket(const RtpPacketReceived& packet);
   bool ProcessReceivedPackets();
 
   // Config.

webrtc/modules/rtp_rtcp/source/flexfec_receiver.cc

@@ -45,7 +45,8 @@ FlexfecReceiver::FlexfecReceiver(
 
 FlexfecReceiver::~FlexfecReceiver() = default;
 
-bool FlexfecReceiver::AddAndProcessReceivedPacket(RtpPacketReceived packet) {
+bool FlexfecReceiver::AddAndProcessReceivedPacket(
+    const RtpPacketReceived& packet) {
   RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
   if (!AddReceivedPacket(std::move(packet))) {
     return false;
@@ -58,7 +59,7 @@ FecPacketCounter FlexfecReceiver::GetPacketCounter() const {
   return packet_counter_;
 }
 
-bool FlexfecReceiver::AddReceivedPacket(RtpPacketReceived packet) {
+bool FlexfecReceiver::AddReceivedPacket(const RtpPacketReceived& packet) {
   RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
 
   // RTP packets with a full base header (12 bytes), but without payload,

webrtc/test/call_test.cc

@@ -279,6 +279,7 @@ void CallTest::CreateMatchingReceiveConfigs(Transport* rtcp_send_transport) {
     config.payload_type = kFlexfecPayloadType;
     config.remote_ssrc = kFlexfecSendSsrc;
     config.protected_media_ssrcs = {kVideoSendSsrcs[0]};
+    config.local_ssrc = kReceiverLocalVideoSsrc;
     for (const RtpExtension& extension : video_send_config_.rtp.extensions)
       config.rtp_header_extensions.push_back(extension);
     flexfec_receive_configs_.push_back(config);

webrtc/test/call_test.h

@@ -56,7 +56,6 @@ class CallTest : public ::testing::Test {
   static const uint32_t kFlexfecSendSsrc;
   static const uint32_t kReceiverLocalVideoSsrc;
   static const uint32_t kReceiverLocalAudioSsrc;
-  static const uint32_t kReceiverLocalFlexfecSsrc;
   static const int kNackRtpHistoryMs;
 
  protected:

webrtc/video/end_to_end_tests.cc

@@ -705,84 +705,126 @@ TEST_P(EndToEndTest, CanReceiveUlpfec) {
   RunBaseTest(&test);
 }
 
-TEST_P(EndToEndTest, CanReceiveFlexfec) {
+class FlexfecRenderObserver : public test::EndToEndTest,
-  class FlexfecRenderObserver : public test::EndToEndTest,
+                              public rtc::VideoSinkInterface<VideoFrame> {
-                                public rtc::VideoSinkInterface<VideoFrame> {
+ public:
-   public:
+  static constexpr uint32_t kVideoLocalSsrc = 123;
-    FlexfecRenderObserver()
+  static constexpr uint32_t kFlexfecLocalSsrc = 456;
-        : EndToEndTest(kDefaultTimeoutMs), random_(0xcafef00d1) {}
 
-    size_t GetNumFlexfecStreams() const override { return 1; }
+  explicit FlexfecRenderObserver(bool expect_flexfec_rtcp)
+      : test::EndToEndTest(test::CallTest::kDefaultTimeoutMs),
+        expect_flexfec_rtcp_(expect_flexfec_rtcp),
+        received_flexfec_rtcp_(false),
+        random_(0xcafef00d1) {}
 
-   private:
+  size_t GetNumFlexfecStreams() const override { return 1; }
-    Action OnSendRtp(const uint8_t* packet, size_t length) override {
-      rtc::CritScope lock(&crit_);
-      RTPHeader header;
-      EXPECT_TRUE(parser_->Parse(packet, length, &header));
 
-      uint8_t payload_type = header.payloadType;
+ private:
-      if (payload_type != kFakeVideoSendPayloadType) {
+  Action OnSendRtp(const uint8_t* packet, size_t length) override {
-        EXPECT_EQ(kFlexfecPayloadType, payload_type);
+    rtc::CritScope lock(&crit_);
-      }
+    RTPHeader header;
+    EXPECT_TRUE(parser_->Parse(packet, length, &header));
 
-      // Is this a retransmitted media packet? From the perspective of FEC, this
+    uint8_t payload_type = header.payloadType;
-      // packet is then no longer dropped, so remove it from the list of
+    if (payload_type != test::CallTest::kFakeVideoSendPayloadType) {
-      // dropped packets.
+      EXPECT_EQ(test::CallTest::kFlexfecPayloadType, payload_type);
-      if (payload_type == kFakeVideoSendPayloadType) {
-        auto seq_num_it = dropped_sequence_numbers_.find(header.sequenceNumber);
-        if (seq_num_it != dropped_sequence_numbers_.end()) {
-          dropped_sequence_numbers_.erase(seq_num_it);
-          auto ts_it = dropped_timestamps_.find(header.timestamp);
-          EXPECT_NE(ts_it, dropped_timestamps_.end());
-          dropped_timestamps_.erase(ts_it);
-
-          return SEND_PACKET;
-        }
-      }
-
-      // Simulate 5% packet loss. Record what media packets, and corresponding
-      // timestamps, that were dropped.
-      if (random_.Rand(1, 100) <= 5) {
-        if (payload_type == kFakeVideoSendPayloadType) {
-          dropped_sequence_numbers_.insert(header.sequenceNumber);
-          dropped_timestamps_.insert(header.timestamp);
-        }
-
-        return DROP_PACKET;
-      }
-
-      return SEND_PACKET;
     }
 
-    void OnFrame(const VideoFrame& video_frame) override {
+    // Is this a retransmitted media packet? From the perspective of FEC, this
-      rtc::CritScope lock(&crit_);
+    // packet is then no longer dropped, so remove it from the list of
-      // Rendering frame with timestamp of packet that was dropped -> FEC
+    // dropped packets.
-      // protection worked.
+    if (payload_type == test::CallTest::kFakeVideoSendPayloadType) {
-      auto it = dropped_timestamps_.find(video_frame.timestamp());
+      auto seq_num_it = dropped_sequence_numbers_.find(header.sequenceNumber);
-      if (it != dropped_timestamps_.end())
+      if (seq_num_it != dropped_sequence_numbers_.end()) {
+        dropped_sequence_numbers_.erase(seq_num_it);
+        auto ts_it = dropped_timestamps_.find(header.timestamp);
+        EXPECT_NE(ts_it, dropped_timestamps_.end());
+        dropped_timestamps_.erase(ts_it);
+
+        return SEND_PACKET;
+      }
+    }
+
+    // Simulate 5% packet loss. Record what media packets, and corresponding
+    // timestamps, that were dropped.
+    if (random_.Rand(1, 100) <= 5) {
+      if (payload_type == test::CallTest::kFakeVideoSendPayloadType) {
+        dropped_sequence_numbers_.insert(header.sequenceNumber);
+        dropped_timestamps_.insert(header.timestamp);
+      }
+
+      return DROP_PACKET;
+    }
+
+    return SEND_PACKET;
+  }
+
+  Action OnReceiveRtcp(const uint8_t* data, size_t length) override {
+    test::RtcpPacketParser parser;
+
+    parser.Parse(data, length);
+    if (parser.sender_ssrc() == kFlexfecLocalSsrc) {
+      EXPECT_EQ(1, parser.receiver_report()->num_packets());
+      const std::vector<rtcp::ReportBlock>& report_blocks =
+          parser.receiver_report()->report_blocks();
+      if (!report_blocks.empty()) {
+        EXPECT_EQ(1U, report_blocks.size());
+        EXPECT_EQ(test::CallTest::kFlexfecSendSsrc,
+                  report_blocks[0].source_ssrc());
+        received_flexfec_rtcp_ = true;
+      }
+    }
+
+    return SEND_PACKET;
+  }
+
+  void OnFrame(const VideoFrame& video_frame) override {
+    rtc::CritScope lock(&crit_);
+    // Rendering frame with timestamp of packet that was dropped -> FEC
+    // protection worked.
+    auto it = dropped_timestamps_.find(video_frame.timestamp());
+    if (it != dropped_timestamps_.end()) {
+      if (!expect_flexfec_rtcp_ || received_flexfec_rtcp_) {
         observation_complete_.Set();
+      }
     }
+  }
 
-    void ModifyVideoConfigs(
+  void ModifyVideoConfigs(
-        VideoSendStream::Config* send_config,
+      VideoSendStream::Config* send_config,
-        std::vector<VideoReceiveStream::Config>* receive_configs,
+      std::vector<VideoReceiveStream::Config>* receive_configs,
-        VideoEncoderConfig* encoder_config) override {
+      VideoEncoderConfig* encoder_config) override {
-      (*receive_configs)[0].renderer = this;
+    (*receive_configs)[0].rtp.local_ssrc = kVideoLocalSsrc;
-    }
+    (*receive_configs)[0].renderer = this;
+  }
 
-    void PerformTest() override {
+  void ModifyFlexfecConfigs(
-      EXPECT_TRUE(Wait())
+      std::vector<FlexfecReceiveStream::Config>* receive_configs) override {
-          << "Timed out waiting for dropped frames to be rendered.";
+    (*receive_configs)[0].local_ssrc = kFlexfecLocalSsrc;
-    }
+  }
 
-    rtc::CriticalSection crit_;
+  void PerformTest() override {
-    std::set<uint32_t> dropped_sequence_numbers_ GUARDED_BY(crit_);
+    EXPECT_TRUE(Wait())
-    // Since several packets can have the same timestamp a multiset is used
+        << "Timed out waiting for dropped frames to be rendered.";
-    // instead of a set.
+  }
-    std::multiset<uint32_t> dropped_timestamps_ GUARDED_BY(crit_);
-    Random random_;
-  } test;
 
+  rtc::CriticalSection crit_;
+  std::set<uint32_t> dropped_sequence_numbers_ GUARDED_BY(crit_);
+  // Since several packets can have the same timestamp a multiset is used
+  // instead of a set.
+  std::multiset<uint32_t> dropped_timestamps_ GUARDED_BY(crit_);
+  bool expect_flexfec_rtcp_;
+  bool received_flexfec_rtcp_;
+  Random random_;
+};
+
+TEST_P(EndToEndTest, ReceivesFlexfec) {
+  FlexfecRenderObserver test(false);
+  RunBaseTest(&test);
+}
+
+TEST_P(EndToEndTest, ReceivesFlexfecAndSendsCorrespondingRtcp) {
+  FlexfecRenderObserver test(true);
   RunBaseTest(&test);
 }
 

webrtc/video/video_quality_test.cc

@@ -1121,6 +1121,7 @@ void VideoQualityTest::SetupVideo(Transport* send_transport,
     flexfec_receive_config.remote_ssrc = video_send_config_.rtp.flexfec.ssrc;
     flexfec_receive_config.protected_media_ssrcs =
         video_send_config_.rtp.flexfec.protected_media_ssrcs;
+    flexfec_receive_config.local_ssrc = kReceiverLocalVideoSsrc;
     flexfec_receive_config.transport_cc = params_.call.send_side_bwe;
     if (params_.call.send_side_bwe) {
       flexfec_receive_config.rtp_header_extensions.push_back(