Reduce locking when collecting receive statistics.

webrtc/modules/rtp_rtcp/source/receive_statistics_impl.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 11 matching lines @@
 #include "webrtc/system_wrappers/include/clock.h"
 
 namespace webrtc {
 
 const int64_t kStatisticsTimeoutMs = 8000;
 const int64_t kStatisticsProcessIntervalMs = 1000;
 
 StreamStatistician::~StreamStatistician() {}
 
 StreamStatisticianImpl::StreamStatisticianImpl(
+    uint32_t ssrc,
     Clock* clock,
     RtcpStatisticsCallback* rtcp_callback,
     StreamDataCountersCallback* rtp_callback)
-    : clock_(clock),
+    : ssrc_(ssrc),
+      clock_(clock),
       incoming_bitrate_(kStatisticsProcessIntervalMs,
                         RateStatistics::kBpsScale),
-      ssrc_(0),
       max_reordering_threshold_(kDefaultMaxReorderingThreshold),
       jitter_q4_(0),
       cumulative_loss_(0),
       jitter_q4_transmission_time_offset_(0),
       last_receive_time_ms_(0),
       last_received_timestamp_(0),
       last_received_transmission_time_offset_(0),
       received_seq_first_(0),
       received_seq_max_(0),
       received_seq_wraps_(0),
       received_packet_overhead_(12),
       last_report_inorder_packets_(0),
       last_report_old_packets_(0),
       last_report_seq_max_(0),
       rtcp_callback_(rtcp_callback),
       rtp_callback_(rtp_callback) {}
 
 void StreamStatisticianImpl::IncomingPacket(const RTPHeader& header,
                                             size_t packet_length,
                                             bool retransmitted) {
-  UpdateCounters(header, packet_length, retransmitted);
-  NotifyRtpCallback();
+  auto counters = UpdateCounters(header, packet_length, retransmitted);
+  rtp_callback_->DataCountersUpdated(counters, ssrc_);
 }
 
-void StreamStatisticianImpl::UpdateCounters(const RTPHeader& header,
-                                            size_t packet_length,
-                                            bool retransmitted) {
+StreamDataCounters StreamStatisticianImpl::UpdateCounters(
+    const RTPHeader& header,
+    size_t packet_length,
+    bool retransmitted) {
   rtc::CritScope cs(&stream_lock_);
   bool in_order = InOrderPacketInternal(header.sequenceNumber);
-  ssrc_ = header.ssrc;
+  RTC_DCHECK_EQ(ssrc_, header.ssrc);
   incoming_bitrate_.Update(packet_length, clock_->TimeInMilliseconds());
   receive_counters_.transmitted.AddPacket(packet_length, header);
   if (!in_order && retransmitted) {
     receive_counters_.retransmitted.AddPacket(packet_length, header);
   }
 
   if (receive_counters_.transmitted.packets == 1) {
     received_seq_first_ = header.sequenceNumber;
     receive_counters_.first_packet_time_ms = clock_->TimeInMilliseconds();
   }
@@ skipping 23 matching lines @@
     last_received_timestamp_ = header.timestamp;
     last_receive_time_ntp_ = receive_time;
     last_receive_time_ms_ = clock_->TimeInMilliseconds();
   }
 
   size_t packet_oh = header.headerLength + header.paddingLength;
 
   // Our measured overhead. Filter from RFC 5104 4.2.1.2:
   // avg_OH (new) = 15/16*avg_OH (old) + 1/16*pckt_OH,
   received_packet_overhead_ = (15 * received_packet_overhead_ + packet_oh) >> 4;
+  return receive_counters_;
 }
 
 void StreamStatisticianImpl::UpdateJitter(const RTPHeader& header,
                                           NtpTime receive_time) {
   uint32_t receive_time_rtp =
       NtpToRtp(receive_time, header.payload_type_frequency);
   uint32_t last_receive_time_rtp =
       NtpToRtp(last_receive_time_ntp_, header.payload_type_frequency);
   int32_t time_diff_samples = (receive_time_rtp - last_receive_time_rtp) -
       (header.timestamp - last_received_timestamp_);
@@ skipping 21 matching lines @@
   time_diff_samples_ext = std::abs(time_diff_samples_ext);
 
   if (time_diff_samples_ext < 450000) {
     int32_t jitter_diffQ4TransmissionTimeOffset =
       (time_diff_samples_ext << 4) - jitter_q4_transmission_time_offset_;
     jitter_q4_transmission_time_offset_ +=
       ((jitter_diffQ4TransmissionTimeOffset + 8) >> 4);
   }
 }
 
-void StreamStatisticianImpl::NotifyRtpCallback() {
-  StreamDataCounters data;
-  uint32_t ssrc;
-  {
-    rtc::CritScope cs(&stream_lock_);
-    data = receive_counters_;
-    ssrc = ssrc_;
-  }
-  rtp_callback_->DataCountersUpdated(data, ssrc);
-}
-
-void StreamStatisticianImpl::NotifyRtcpCallback() {
-  RtcpStatistics data;
-  uint32_t ssrc;
-  {
-    rtc::CritScope cs(&stream_lock_);
-    data = last_reported_statistics_;
-    ssrc = ssrc_;
-  }
-  rtcp_callback_->StatisticsUpdated(data, ssrc);
-}
-
 void StreamStatisticianImpl::FecPacketReceived(const RTPHeader& header,
                                                size_t packet_length) {
+  StreamDataCounters counters;
   {
     rtc::CritScope cs(&stream_lock_);
     receive_counters_.fec.AddPacket(packet_length, header);
+    counters = receive_counters_;
   }
-  NotifyRtpCallback();
+  rtp_callback_->DataCountersUpdated(counters, ssrc_);
 }
 
 void StreamStatisticianImpl::SetMaxReorderingThreshold(
     int max_reordering_threshold) {
   rtc::CritScope cs(&stream_lock_);
   max_reordering_threshold_ = max_reordering_threshold;
 }
 
 bool StreamStatisticianImpl::GetStatistics(RtcpStatistics* statistics,
                                            bool reset) {
@@ skipping 11 matching lines @@
         return false;
       }
       // Just get last report.
       *statistics = last_reported_statistics_;
       return true;
     }
 
     *statistics = CalculateRtcpStatistics();
   }
 
-  NotifyRtcpCallback();
-
+  rtcp_callback_->StatisticsUpdated(*statistics, ssrc_);
   return true;
 }
 
 RtcpStatistics StreamStatisticianImpl::CalculateRtcpStatistics() {
   RtcpStatistics stats;
 
   if (last_report_inorder_packets_ == 0) {
     // First time we send a report.
     last_report_seq_max_ = received_seq_first_ - 1;
   }
@@ skipping 170 matching lines @@
 void ReceiveStatisticsImpl::IncomingPacket(const RTPHeader& header,
                                            size_t packet_length,
                                            bool retransmitted) {
   StreamStatisticianImpl* impl;
   {
     rtc::CritScope cs(&receive_statistics_lock_);
     StatisticianImplMap::iterator it = statisticians_.find(header.ssrc);
     if (it != statisticians_.end()) {
       impl = it->second;
     } else {
-      impl = new StreamStatisticianImpl(clock_, this, this);
+      impl = new StreamStatisticianImpl(header.ssrc, clock_, this, this);
       statisticians_[header.ssrc] = impl;
     }
   }
   // StreamStatisticianImpl instance is created once and only destroyed when
   // this whole ReceiveStatisticsImpl is destroyed. StreamStatisticianImpl has
   // it's own locking so don't hold receive_statistics_lock_ (potential
   // deadlock).
   impl->IncomingPacket(header, packet_length, retransmitted);
 }
 
 void ReceiveStatisticsImpl::FecPacketReceived(const RTPHeader& header,
                                               size_t packet_length) {
-  rtc::CritScope cs(&receive_statistics_lock_);
-  StatisticianImplMap::iterator it = statisticians_.find(header.ssrc);
-  // Ignore FEC if it is the first packet.
-  if (it != statisticians_.end()) {
-    it->second->FecPacketReceived(header, packet_length);
+  StreamStatisticianImpl* impl;
+  {
+    rtc::CritScope cs(&receive_statistics_lock_);
+    StatisticianImplMap::iterator it = statisticians_.find(header.ssrc);
+    // Ignore FEC if it is the first packet.
+    if (it == statisticians_.end())
+      return;
+    impl = it->second;
   }
+  impl->FecPacketReceived(header, packet_length);
 }
 
 StatisticianMap ReceiveStatisticsImpl::GetActiveStatisticians() const {
   StatisticianMap active_statisticians;
   rtc::CritScope cs(&receive_statistics_lock_);
   for (StatisticianImplMap::const_iterator it = statisticians_.begin();
        it != statisticians_.end(); ++it) {
     uint32_t secs;
     uint32_t frac;
     it->second->LastReceiveTimeNtp(&secs, &frac);
@@ skipping 85 matching lines @@
       result.pop_back();
       continue;
     }
     block.SetExtHighestSeqNum(stats.extended_highest_sequence_number);
     block.SetJitter(stats.jitter);
   }
   return result;
 }
 
 }  // namespace webrtc