RTCStatsCollector: Utilize network thread to minimize thread hops.

webrtc/api/rtcstatscollector.cc

 /*
  *  Copyright 2016 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 392 matching lines @@
     // "Now" using a system clock, relative to the UNIX epoch (Jan 1, 1970,
     // UTC), in microseconds. The system clock could be modified and is not
     // necessarily monotonically increasing.
     int64_t timestamp_us = rtc::TimeUTCMicros();
 
     num_pending_partial_reports_ = 3;
     partial_report_timestamp_us_ = cache_now_us;
     invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_,
         rtc::Bind(&RTCStatsCollector::ProducePartialResultsOnSignalingThread,
             rtc::scoped_refptr<RTCStatsCollector>(this), timestamp_us));
+
+    // TODO(hbos): No stats are gathered by
+    // |ProducePartialResultsOnWorkerThread|, remove it.
     invoker_.AsyncInvoke<void>(RTC_FROM_HERE, worker_thread_,
         rtc::Bind(&RTCStatsCollector::ProducePartialResultsOnWorkerThread,
             rtc::scoped_refptr<RTCStatsCollector>(this), timestamp_us));
+
+    // Prepare |channel_names_| and |media_info_| for use in
+    // |ProducePartialResultsOnNetworkThread|.
+    channel_name_pairs_.reset(new ChannelNamePairs());
+    if (pc_->session()->voice_channel()) {
+      channel_name_pairs_->voice = rtc::Optional<ChannelNamePair>(
+          ChannelNamePair(pc_->session()->voice_channel()->content_name(),
+                          pc_->session()->voice_channel()->transport_name()));
+    }
+    if (pc_->session()->video_channel()) {
+      channel_name_pairs_->video = rtc::Optional<ChannelNamePair>(
+          ChannelNamePair(pc_->session()->video_channel()->content_name(),
+                          pc_->session()->video_channel()->transport_name()));
+    }
+    if (pc_->session()->data_channel()) {
+      channel_name_pairs_->data = rtc::Optional<ChannelNamePair>(
+          ChannelNamePair(pc_->session()->data_channel()->content_name(),
+                          pc_->session()->data_channel()->transport_name()));
+    }
+    media_info_.reset(PrepareMediaInfo_s().release());
     invoker_.AsyncInvoke<void>(RTC_FROM_HERE, network_thread_,
         rtc::Bind(&RTCStatsCollector::ProducePartialResultsOnNetworkThread,
             rtc::scoped_refptr<RTCStatsCollector>(this), timestamp_us));
   }
 }
 
 void RTCStatsCollector::ClearCachedStatsReport() {
   RTC_DCHECK(signaling_thread_->IsCurrent());
   cached_report_ = nullptr;
 }
 
 void RTCStatsCollector::ProducePartialResultsOnSignalingThread(
     int64_t timestamp_us) {
   RTC_DCHECK(signaling_thread_->IsCurrent());
   rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create(
       timestamp_us);
 
-  SessionStats session_stats;
-  if (pc_->session()->GetTransportStats(&session_stats)) {
-    std::map<std::string, CertificateStatsPair> transport_cert_stats =
-        PrepareTransportCertificateStats(session_stats);
-    MediaInfo media_info = PrepareMediaInfo(session_stats);
-
-    ProduceCertificateStats_s(
-        timestamp_us, transport_cert_stats, report.get());
-    ProduceCodecStats_s(
-        timestamp_us, media_info, report.get());
-    ProduceIceCandidateAndPairStats_s(
-        timestamp_us, session_stats, report.get());
-    ProduceRTPStreamStats_s(
-        timestamp_us, session_stats, media_info, report.get());
-    ProduceTransportStats_s(
-        timestamp_us, session_stats, transport_cert_stats, report.get());
-  }
   ProduceDataChannelStats_s(timestamp_us, report.get());
   ProduceMediaStreamAndTrackStats_s(timestamp_us, report.get());
   ProducePeerConnectionStats_s(timestamp_us, report.get());
 
   AddPartialResults(report);
 }
 
 void RTCStatsCollector::ProducePartialResultsOnWorkerThread(
     int64_t timestamp_us) {
   RTC_DCHECK(worker_thread_->IsCurrent());
   rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create(
       timestamp_us);
 
-  // TODO(hbos): Gather stats on worker thread.
-  // pc_->session()'s channels are owned by the signaling thread but there are
-  // some stats that are gathered on the worker thread. Instead of a synchronous
-  // invoke on "s->w" we could to the "w" work here asynchronously if it wasn't
-  // for the ownership issue. Synchronous invokes in other places makes it
-  // difficult to introduce locks without introducing deadlocks and the channels
-  // are not reference counted.
+  // TODO(hbos): There are no stats to be gathered on this thread, remove this
+  // method.
 
   AddPartialResults(report);
 }
 
 void RTCStatsCollector::ProducePartialResultsOnNetworkThread(
     int64_t timestamp_us) {
   RTC_DCHECK(network_thread_->IsCurrent());
   rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create(
       timestamp_us);
 
-  // TODO(hbos): Gather stats on network thread.
-  // pc_->session()'s channels are owned by the signaling thread but there are
-  // some stats that are gathered on the network thread. Instead of a
-  // synchronous invoke on "s->n" we could to the "n" work here asynchronously
-  // if it wasn't for the ownership issue. Synchronous invokes in other places
-  // makes it difficult to introduce locks without introducing deadlocks and the
-  // channels are not reference counted.
+  std::unique_ptr<SessionStats> session_stats =
+      pc_->session()->GetSessionStats(*channel_name_pairs_);

[pthatcher1, 2016/12/15 23:18:59]

I don't understand why you can't just call GetSessionStats() here and get all
the channel name pairs.  Why did you have to pass that in?

[hbos, 2016/12/16 10:38:02]

WebRtcSession owns the pointers. If we were to read their names here, channels
could be created or destroyed on the signaling thread (other threads also
involved) while we're trying to use them here on the network thread (since we
got here by a non-blocking thread hop). I would otherwise have to introduce
locks, but I think that would be a big change with risk of deadlocks in many
places. Copying the transport names should be safe. If the transports have been
destroyed during the hop this will just return null.

+  if (session_stats) {
+    std::map<std::string, CertificateStatsPair> transport_cert_stats =
+        PrepareTransportCertificateStats_n(*session_stats);
+
+    ProduceCertificateStats_n(
+        timestamp_us, transport_cert_stats, report.get());
+    ProduceCodecStats_n(
+        timestamp_us, *media_info_, report.get());
+    ProduceIceCandidateAndPairStats_n(
+        timestamp_us, *session_stats, report.get());
+    ProduceRTPStreamStats_n(
+        timestamp_us, *session_stats, *media_info_, report.get());
+    ProduceTransportStats_n(
+        timestamp_us, *session_stats, transport_cert_stats, report.get());
+  }
 
   AddPartialResults(report);
 }
 
 void RTCStatsCollector::AddPartialResults(
     const rtc::scoped_refptr<RTCStatsReport>& partial_report) {
   if (!signaling_thread_->IsCurrent()) {
     invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_,
         rtc::Bind(&RTCStatsCollector::AddPartialResults_s,
                   rtc::scoped_refptr<RTCStatsCollector>(this),
@@ skipping 24 matching lines @@
   RTC_DCHECK(signaling_thread_->IsCurrent());
   RTC_DCHECK(!callbacks_.empty());
   RTC_DCHECK(cached_report_);
   for (const rtc::scoped_refptr<RTCStatsCollectorCallback>& callback :
        callbacks_) {
     callback->OnStatsDelivered(cached_report_);
   }
   callbacks_.clear();
 }
 
-void RTCStatsCollector::ProduceCertificateStats_s(
+void RTCStatsCollector::ProduceCertificateStats_n(
     int64_t timestamp_us,
     const std::map<std::string, CertificateStatsPair>& transport_cert_stats,
     RTCStatsReport* report) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   for (const auto& transport_cert_stats_pair : transport_cert_stats) {
     if (transport_cert_stats_pair.second.local) {
       ProduceCertificateStatsFromSSLCertificateStats(
           timestamp_us, *transport_cert_stats_pair.second.local.get(), report);
     }
     if (transport_cert_stats_pair.second.remote) {
       ProduceCertificateStatsFromSSLCertificateStats(
           timestamp_us, *transport_cert_stats_pair.second.remote.get(), report);
     }
   }
 }
 
-void RTCStatsCollector::ProduceCodecStats_s(
+void RTCStatsCollector::ProduceCodecStats_n(
     int64_t timestamp_us, const MediaInfo& media_info,
     RTCStatsReport* report) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   // Audio
   if (media_info.voice) {
     // Inbound
     for (const auto& pair : media_info.voice->receive_codecs) {
       report->AddStats(CodecStatsFromRtpCodecParameters(
           timestamp_us, true, true, pair.second));
     }
     // Outbound
     for (const auto& pair : media_info.voice->send_codecs) {
       report->AddStats(CodecStatsFromRtpCodecParameters(
@@ skipping 30 matching lines @@
     data_channel_stats->state =
         DataStateToRTCDataChannelState(data_channel->state());
     data_channel_stats->messages_sent = data_channel->messages_sent();
     data_channel_stats->bytes_sent = data_channel->bytes_sent();
     data_channel_stats->messages_received = data_channel->messages_received();
     data_channel_stats->bytes_received = data_channel->bytes_received();
     report->AddStats(std::move(data_channel_stats));
   }
 }
 
-void RTCStatsCollector::ProduceIceCandidateAndPairStats_s(
+void RTCStatsCollector::ProduceIceCandidateAndPairStats_n(
       int64_t timestamp_us, const SessionStats& session_stats,
       RTCStatsReport* report) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   for (const auto& transport_stats : session_stats.transport_stats) {
     for (const auto& channel_stats : transport_stats.second.channel_stats) {
       std::string transport_id = RTCTransportStatsIDFromTransportChannel(
           transport_stats.second.transport_name, channel_stats.component);
       for (const cricket::ConnectionInfo& info :
            channel_stats.connection_infos) {
         std::unique_ptr<RTCIceCandidatePairStats> candidate_pair_stats(
             new RTCIceCandidatePairStats(
                 RTCIceCandidatePairStatsIDFromConnectionInfo(info),
                 timestamp_us));
@@ skipping 52 matching lines @@
 void RTCStatsCollector::ProducePeerConnectionStats_s(
     int64_t timestamp_us, RTCStatsReport* report) const {
   RTC_DCHECK(signaling_thread_->IsCurrent());
   std::unique_ptr<RTCPeerConnectionStats> stats(
     new RTCPeerConnectionStats("RTCPeerConnection", timestamp_us));
   stats->data_channels_opened = internal_record_.data_channels_opened;
   stats->data_channels_closed = internal_record_.data_channels_closed;
   report->AddStats(std::move(stats));
 }
 
-void RTCStatsCollector::ProduceRTPStreamStats_s(
+void RTCStatsCollector::ProduceRTPStreamStats_n(
     int64_t timestamp_us, const SessionStats& session_stats,
     const MediaInfo& media_info, RTCStatsReport* report) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   // Audio
   if (media_info.voice) {
     std::string transport_id = RTCTransportStatsIDFromBaseChannel(
         session_stats.proxy_to_transport, *pc_->session()->voice_channel());
     RTC_DCHECK(!transport_id.empty());
     // Inbound
     for (const cricket::VoiceReceiverInfo& voice_receiver_info :
          media_info.voice->receivers) {
       // TODO(nisse): SSRC == 0 currently means none. Delete check when that
@@ skipping 83 matching lines @@
       if (video_sender_info.codec_payload_type) {
         outbound_video->codec_id =
             RTCCodecStatsIDFromDirectionMediaAndPayload(
                 false, false, *video_sender_info.codec_payload_type);
       }
       report->AddStats(std::move(outbound_video));
     }
   }
 }
 
-void RTCStatsCollector::ProduceTransportStats_s(
+void RTCStatsCollector::ProduceTransportStats_n(
     int64_t timestamp_us, const SessionStats& session_stats,
     const std::map<std::string, CertificateStatsPair>& transport_cert_stats,
     RTCStatsReport* report) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   for (const auto& transport : session_stats.transport_stats) {
     // Get reference to RTCP channel, if it exists.
     std::string rtcp_transport_stats_id;
     for (const auto& channel_stats : transport.second.channel_stats) {
       if (channel_stats.component ==
           cricket::ICE_CANDIDATE_COMPONENT_RTCP) {
         rtcp_transport_stats_id = RTCTransportStatsIDFromTransportChannel(
             transport.second.transport_name, channel_stats.component);
         break;
       }
@@ skipping 42 matching lines @@
       if (!local_certificate_id.empty())
         transport_stats->local_certificate_id = local_certificate_id;
       if (!remote_certificate_id.empty())
         transport_stats->remote_certificate_id = remote_certificate_id;
       report->AddStats(std::move(transport_stats));
     }
   }
 }
 
 std::map<std::string, RTCStatsCollector::CertificateStatsPair>
-RTCStatsCollector::PrepareTransportCertificateStats(
+RTCStatsCollector::PrepareTransportCertificateStats_n(
     const SessionStats& session_stats) const {
-  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   std::map<std::string, CertificateStatsPair> transport_cert_stats;
   for (const auto& transport_stats : session_stats.transport_stats) {
     CertificateStatsPair certificate_stats_pair;
     rtc::scoped_refptr<rtc::RTCCertificate> local_certificate;
     if (pc_->session()->GetLocalCertificate(
         transport_stats.second.transport_name, &local_certificate)) {
       certificate_stats_pair.local =
           local_certificate->ssl_certificate().GetStats();
     }
     std::unique_ptr<rtc::SSLCertificate> remote_certificate =
         pc_->session()->GetRemoteSSLCertificate(
             transport_stats.second.transport_name);
     if (remote_certificate) {
       certificate_stats_pair.remote = remote_certificate->GetStats();
     }
     transport_cert_stats.insert(
         std::make_pair(transport_stats.second.transport_name,
                        std::move(certificate_stats_pair)));
   }
   return transport_cert_stats;
 }
 
-RTCStatsCollector::MediaInfo RTCStatsCollector::PrepareMediaInfo(
-    const SessionStats& session_stats) const {
-  MediaInfo media_info;
+std::unique_ptr<RTCStatsCollector::MediaInfo>
+RTCStatsCollector::PrepareMediaInfo_s() const {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  std::unique_ptr<MediaInfo> media_info(new MediaInfo());
   if (pc_->session()->voice_channel()) {
     cricket::VoiceMediaInfo voice_media_info;
     if (pc_->session()->voice_channel()->GetStats(&voice_media_info)) {
-      media_info.voice = rtc::Optional<cricket::VoiceMediaInfo>(
+      media_info->voice = rtc::Optional<cricket::VoiceMediaInfo>(
           std::move(voice_media_info));
     }
   }
   if (pc_->session()->video_channel()) {
     cricket::VideoMediaInfo video_media_info;
     if (pc_->session()->video_channel()->GetStats(&video_media_info)) {
-      media_info.video = rtc::Optional<cricket::VideoMediaInfo>(
+      media_info->video = rtc::Optional<cricket::VideoMediaInfo>(
           std::move(video_media_info));
     }
   }
   return media_info;
 }
 
 void RTCStatsCollector::OnDataChannelCreated(DataChannel* channel) {
   channel->SignalOpened.connect(this, &RTCStatsCollector::OnDataChannelOpened);
   channel->SignalClosed.connect(this, &RTCStatsCollector::OnDataChannelClosed);
 }
@@ skipping 21 matching lines @@
     const std::string& type) {
   return CandidateTypeToRTCIceCandidateType(type);
 }
 
 const char* DataStateToRTCDataChannelStateForTesting(
     DataChannelInterface::DataState state) {
   return DataStateToRTCDataChannelState(state);
 }
 
 }  // namespace webrtc