Refactor composing report blocks for rtcp Sender/Receiver reports

webrtc/modules/rtp_rtcp/source/rtcp_sender.cc

 /*
  *  Copyright (c) 2012 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 437 matching lines @@
   uint32_t rtp_timestamp =
       timestamp_offset_ + last_rtp_timestamp_ +
       (clock_->TimeInMilliseconds() - last_frame_capture_time_ms_) * rtp_rate;
 
   rtcp::SenderReport* report = new rtcp::SenderReport();
   report->SetSenderSsrc(ssrc_);
   report->SetNtp(ctx.now_);
   report->SetRtpTimestamp(rtp_timestamp);
   report->SetPacketCount(ctx.feedback_state_.packets_sent);
   report->SetOctetCount(ctx.feedback_state_.media_bytes_sent);
-
-  for (auto it : report_blocks_)
-    report->AddReportBlock(it.second);
-
-  report_blocks_.clear();
+  report->SetReportBlocks(CreateReportBlocks(ctx.feedback_state_));
 
   return std::unique_ptr<rtcp::RtcpPacket>(report);
 }
 
 std::unique_ptr<rtcp::RtcpPacket> RTCPSender::BuildSDES(
     const RtcpContext& ctx) {
   size_t length_cname = cname_.length();
   RTC_CHECK_LT(length_cname, RTCP_CNAME_SIZE);
 
   rtcp::Sdes* sdes = new rtcp::Sdes();
   sdes->AddCName(ssrc_, cname_);
 
   for (const auto& it : csrc_cnames_)
     RTC_CHECK(sdes->AddCName(it.first, it.second));
 
   return std::unique_ptr<rtcp::RtcpPacket>(sdes);
 }
 
 std::unique_ptr<rtcp::RtcpPacket> RTCPSender::BuildRR(const RtcpContext& ctx) {
   rtcp::ReceiverReport* report = new rtcp::ReceiverReport();
   report->SetSenderSsrc(ssrc_);
-  for (auto it : report_blocks_)
-    report->AddReportBlock(it.second);
+  report->SetReportBlocks(CreateReportBlocks(ctx.feedback_state_));
 
-  report_blocks_.clear();
   return std::unique_ptr<rtcp::RtcpPacket>(report);
 }
 
 std::unique_ptr<rtcp::RtcpPacket> RTCPSender::BuildPLI(const RtcpContext& ctx) {
   rtcp::Pli* pli = new rtcp::Pli();
   pli->SetSenderSsrc(ssrc_);
   pli->SetMediaSsrc(remote_ssrc_);
 
   TRACE_EVENT_INSTANT0(TRACE_DISABLED_BY_DEFAULT("webrtc_rtp"),
                        "RTCPSender::PLI");
@@ skipping 325 matching lines @@
       }
       if (minIntervalMs > RTCP_INTERVAL_VIDEO_MS)
         minIntervalMs = RTCP_INTERVAL_VIDEO_MS;
     }
     // The interval between RTCP packets is varied randomly over the
     // range [1/2,3/2] times the calculated interval.
     uint32_t timeToNext =
         random_.Rand(minIntervalMs * 1 / 2, minIntervalMs * 3 / 2);
     next_time_to_send_rtcp_ = clock_->TimeInMilliseconds() + timeToNext;
 
-    if (receive_statistics_) {
-      StatisticianMap statisticians =
-          receive_statistics_->GetActiveStatisticians();
-      RTC_DCHECK(report_blocks_.empty());
-      for (auto& it : statisticians) {
-        AddReportBlock(feedback_state, it.first, it.second);
-      }
-    }
+    // RtcpSender expected to be used for sending either just sender reports
+    // or just receiver reports.
+    RTC_DCHECK(!(IsFlagPresent(kRtcpSr) && IsFlagPresent(kRtcpRr)));
   }
 }
 
-bool RTCPSender::AddReportBlock(const FeedbackState& feedback_state,
-                                uint32_t ssrc,
-                                StreamStatistician* statistician) {
-  // Do we have receive statistics to send?
-  RtcpStatistics stats;
-  if (!statistician->GetStatistics(&stats, true))
-    return false;
+std::vector<rtcp::ReportBlock> RTCPSender::CreateReportBlocks(
+    const FeedbackState& feedback_state) {
+  std::vector<rtcp::ReportBlock> result;
+  if (!receive_statistics_)
+    return result;
 
-  if (report_blocks_.size() >= RTCP_MAX_REPORT_BLOCKS) {
-    LOG(LS_WARNING) << "Too many report blocks.";
-    return false;
+  for (auto& statistician : receive_statistics_->GetActiveStatisticians()) {

[eladalon, 2017/07/21 13:34:45]

Maybe:
  auto name_me = receive_statistics_->GetActiveStatisticians();
  result.reserve(name_me.size());
  for (auto& statistician : name_me) {
?

[danilchap, 2017/07/21 17:11:14]

Done, that is an optimization and closer to the original code.

+    // Do we have receive statistics to send?
+    RtcpStatistics stats;
+    if (!statistician.second->GetStatistics(&stats, true))

[eladalon, 2017/07/21 13:34:45]

nits:

1.
Inside of ReceiveStatisticsImpl::GetActiveStatisticians(), I see the following
lines:
  rtc::CritScope cs(&receive_statistics_lock_);
  StatisticianMap active_statisticians;
We can probably reverse their order.

2.
While doing my due dilligence and making sure that statistician.second is
guaranteed non-null, I saw this:
  if (/* some conditional*/) {
    impl = unrelated;
  } else {
    impl = new StreamStatisticianImpl(clock_, this, this);
    statisticians_[header.ssrc] = impl;
  }
  ...
  impl->IncomingPacket(header, packet_length, retransmitted);
So if |new| fails and we have a nullptr there, we're guatanteed to crash very
soon after. However, we won't know for sure where it's from the |new| or from
|unrelated|. Perhaps an RTC_CHECK is in order after the |new|, but not here?
Wdyt?

[danilchap, 2017/07/21 17:11:14]

This CL is about refactoring rtcp_sender, would prefer to review
ReceiveStatistics in later CL.
(this version does as little checks as previous version).

On 2017/07/21 13:34:45, eladalon wrote:
yep, it won't hurt.
I assume new can't return nullptr (I expect out of memory scenario will
terminate the program)

[eladalon, 2017/07/25 12:19:20]

1. Yes, I didn't intend for this to be done in this CL; just something I came
across because I was reviewing this CL, so I mentioned it here.
2. Let me push a CL for that, then.
3. Do we always crash when OOM? Do all builds of WebRTC do that?

[danilchap, 2017/07/25 15:20:08]

That is my assumption since I never seen code in webrtc that tries to recover
from oom.

+      continue;
+    // TODO(danilchap): Support sending more than |RTCP_MAX_REPORT_BLOCKS| per
+    // compound rtcp packet when single rtcp module is used for multiple media
+    // streams.
+    if (result.size() >= RTCP_MAX_REPORT_BLOCKS) {
+      LOG(LS_WARNING) << "Too many report blocks.";
+      continue;

[eladalon, 2017/07/21 13:34:45]

When I read this, I wonder if we should |continue| or |break|. The reason to not
break is probably that you still want to reset the other statisticians? If so,
perhaps we should comment about it?

(Another thing I ask myself is if we want to produce multiple warning logs when
we continue. If we don't change to |break|, we can still probably have a boolean
local to the function that makes sure we only log once? Or would that be
overkill?)

[danilchap, 2017/07/21 17:11:14]

I initially put more reasonable break, but then switch to continue just to keep
old behaviour.

for future I put a TODO hinting that the way statistics used now, it shouldn't
be possible to go over the limit, so for now I care little what will happen with
too many report blocks.
e.g. Do not mind to switch back to break if you think it would be nicer.

[eladalon, 2017/07/25 12:19:20]

Due to my lack of context, I'm okay with either breaking/continuing, but I would
suggest a comment (in addition to the TODO), attached to that specific line (not
the if-statement), that would show the reader that the behavior (resetting the
other statisticians or avoiding doing so) was intentional, and preferably also
explain the rationale (I would personally like to know which one is the right
behavior).

I wonder if, should this happen, we'd find ourselves (almost) always failing to
send report blocks for the statisticians that appear later in the list, by the
way? But maybe this is all theoretical, because it's not ever supposed to happen
anyway?

[danilchap, 2017/07/25 15:20:08]

Next CL (early draft is https://codereview.webrtc.org/2981163003/) I plan to
move half of this function inside ReceiveStatistics itself.
Then more expensive 'continue' will be simpler to implement than cheaper
'break'. In practice it shouldn't matter for now.

yes, if this should happen, we'll fail to send report blocks to unlucky
statisticians with higher ssrc. Thus todo to propertly support it (spec do
describe several ways how to handle more than 31 report blocks)

+    }
+    rtcp::ReportBlock block;
+    block.SetMediaSsrc(statistician.first);
+    block.SetFractionLost(stats.fraction_lost);
+    if (!block.SetCumulativeLost(stats.cumulative_lost)) {
+      LOG(LS_WARNING) << "Cumulative lost is oversized.";
+      continue;
+    }
+    block.SetExtHighestSeqNum(stats.extended_max_sequence_number);
+    block.SetJitter(stats.jitter);
+
+    result.push_back(block);

[eladalon, 2017/07/21 13:34:45]

Brainstorming - what do you think about constructing the new ReportBlock
immediately in the vector, then having |block| be a reference to that new
object, so that we wouldn't have to perform copying here, but would rather be
performing all of the previous calls to Set*() on the relevant object from the
get-go?

[danilchap, 2017/07/21 17:11:14]

Done.

   }
-  RTC_DCHECK(report_blocks_.find(ssrc) == report_blocks_.end());
-  rtcp::ReportBlock* block = &report_blocks_[ssrc];
-  block->SetMediaSsrc(ssrc);
-  block->SetFractionLost(stats.fraction_lost);
-  if (!block->SetCumulativeLost(stats.cumulative_lost)) {
-    report_blocks_.erase(ssrc);
-    LOG(LS_WARNING) << "Cumulative lost is oversized.";
-    return false;
+
+  if (!result.empty() && ((feedback_state.last_rr_ntp_secs != 0) ||
+                          (feedback_state.last_rr_ntp_frac != 0))) {
+    // Get our NTP as late as possible to avoid a race.
+    uint32_t now = CompactNtp(clock_->CurrentNtpTime());
+
+    uint32_t receive_time = feedback_state.last_rr_ntp_secs & 0x0000FFFF;
+    receive_time <<= 16;
+    receive_time += (feedback_state.last_rr_ntp_frac & 0xffff0000) >> 16;
+
+    uint32_t delay_since_last_sr = now - receive_time;
+    // TODO(danilchap): Instead of setting same value on all report blocks,
+    // set only when media_ssrc match sender ssrc of the sender report
+    // remote times were taken from.
+    for (auto& report_block : result) {
+      report_block.SetLastSr(feedback_state.remote_sr);
+      report_block.SetDelayLastSr(delay_since_last_sr);
+    }
   }
-  block->SetExtHighestSeqNum(stats.extended_max_sequence_number);
-  block->SetJitter(stats.jitter);
-  block->SetLastSr(feedback_state.remote_sr);
-
-  // TODO(sprang): Do we really need separate time stamps for each report?
-  // Get our NTP as late as possible to avoid a race.
-  NtpTime ntp = clock_->CurrentNtpTime();
-
-  // Delay since last received report.
-  if ((feedback_state.last_rr_ntp_secs != 0) ||
-      (feedback_state.last_rr_ntp_frac != 0)) {
-    // Get the 16 lowest bits of seconds and the 16 highest bits of fractions.
-    uint32_t now = CompactNtp(ntp);
-
-    uint32_t receiveTime = feedback_state.last_rr_ntp_secs & 0x0000FFFF;
-    receiveTime <<= 16;
-    receiveTime += (feedback_state.last_rr_ntp_frac & 0xffff0000) >> 16;
-
-    block->SetDelayLastSr(now - receiveTime);
-  }
-  return true;
+  return result;
 }
 
 void RTCPSender::SetCsrcs(const std::vector<uint32_t>& csrcs) {
   RTC_DCHECK_LE(csrcs.size(), kRtpCsrcSize);
   rtc::CritScope lock(&critical_section_rtcp_sender_);
   csrcs_ = csrcs;
 }
 
 int32_t RTCPSender::SetApplicationSpecificData(uint8_t subType,
                                                uint32_t name,
@@ skipping 112 matching lines @@
     max_packet_size = max_packet_size_;
   }
 
   RTC_DCHECK_LE(max_packet_size, IP_PACKET_SIZE);
   uint8_t buffer[IP_PACKET_SIZE];
   return packet.BuildExternalBuffer(buffer, max_packet_size, &sender) &&
          !sender.send_failure_;
 }
 
 }  // namespace webrtc