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 on the worker thread, remove this.
     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(pc_->session()->GetChannelNamePairs().release());

[Taylor Brandstetter, 2016/12/13 19:47:07]

I don't think a new method (GetChannelNamePairs) is needed for this. You could
just call video_channel()->content_name() etc. right from here.

[pthatcher1, 2016/12/13 23:50:18]

I was thinking the exact same thing.

[hbos, 2016/12/14 14:56:31]

I replaced GetTransportStats with GetSessionStats taking ChannelNamePair as
argument, so GetChannelNamePairs is now invoked in several places.

+    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.

[Taylor Brandstetter, 2016/12/13 19:47:07]

There are still stats that are ultimately gathered on the worker thread, though,
and use multiple thread hops.

[hbos, 2016/12/14 14:56:31]

Acknowledge. Out of curiosity: this meaning (blocking) rtc::Thread::Invoke?

[Taylor Brandstetter, 2016/12/15 01:12:53]

Yes. Here, to be specific:
https://cs.chromium.org/chromium/src/third_party/webrtc/pc/channel.cc?q=chann...

[hbos, 2016/12/15 16:12:45]

Oh, good to know, so we still do a network -> worker thread hop.

[Taylor Brandstetter, 2016/12/15 17:35:19]

Yep, and it's actually two hops, one each for audio/video.

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

That should be reducible to 1 hop, but I don't think we can get around the
blocking.

 
   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_n(*channel_name_pairs_);

[Taylor Brandstetter, 2016/12/13 19:47:07]

Instead of adding GetSessionStats_n, how about just exposing TransportController
from WebRtcSession and calling GetStats on it directly? I know that removes a
level of abstraction, but for stats collecting I think it makes sense.

Ideally it would only return a const pointer, but that's not possible right now
(see comment in transportcontroller.h).

[pthatcher1, 2016/12/13 23:50:19]

I agree, but better yet, we could just expose the necessary GetTransport and
GetTransportChannel methods that return you a Transport or TransportChannel
(hopefully to be one and the same soon) and then call the necessary methods on
those directly (on the correct thread).

[Taylor Brandstetter, 2016/12/14 00:10:25]

Good point. But that may be introducing more inter-dependencies than necessary.
Since TransportController::GetStats already exists, there may be less churn if
we just go through that for now. We'll be making a bunch of changes to
TransortChannel/etc. soon that might break RTCStatsCollector if it used
TransportChannel directly.

[hbos, 2016/12/14 14:56:31]

With the latest changes there are no new functions. GetTransportStats was
replaced by GetSessionStats and the local/remote certificate getters were
modified to allow being invoked on either thread.

If I were to rely on TransportController directly I would need to move mocking
there for rtcstats_unittest.cc to work (so you would have to control which
TransportController the WebRtcSession uses from the unittest and add
MockTransportController). Sounds like a lot of refactoring. With that, plus
planned changes mentioned and current differences between
JsepTransport/TransportChannel for local/remote cert, I think that should be
done as separate work.

[Taylor Brandstetter, 2016/12/15 01:12:53]

Makes sense; I didn't consider that.

+  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(
+    if (pc_->session()->GetLocalCertificate_n(
         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(
+        pc_->session()->GetRemoteSSLCertificate_n(
             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