RTCStatsCollector collecting stats on multiple threads.

webrtc/stats/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.
  */
 
 #include "webrtc/stats/rtcstatscollector.h"
 
 #include <memory>
 #include <utility>
 #include <vector>
 
 #include "webrtc/api/peerconnection.h"
 #include "webrtc/base/checks.h"
 
 namespace webrtc {
 
+rtc::scoped_refptr<RTCStatsCollector> RTCStatsCollector::Create(
+    PeerConnection* pc,
+    double cache_lifetime,
+    std::unique_ptr<rtc::Timing> timing) {
+  return rtc::scoped_refptr<RTCStatsCollector>(
+      new rtc::RefCountedObject<RTCStatsCollector>(
+          pc, cache_lifetime, timing.release()));
+}
+
 RTCStatsCollector::RTCStatsCollector(
     PeerConnection* pc,
     double cache_lifetime,
-    std::unique_ptr<rtc::Timing> timing)
+    rtc::Timing* timing)
     : pc_(pc),
-      timing_(std::move(timing)),
+      signaling_thread_(pc->session()->signaling_thread()),
+      worker_thread_(pc->session()->worker_thread()),
+      network_thread_(pc->session()->network_thread()),
+      num_pending_partial_reports_(0),
+      timing_(timing),
       cache_timestamp_(0.0),
       cache_lifetime_(cache_lifetime) {
   RTC_DCHECK(pc_);
+  RTC_DCHECK(signaling_thread_);
+  RTC_DCHECK(worker_thread_);
+  RTC_DCHECK(network_thread_);
   RTC_DCHECK(timing_);
-  RTC_DCHECK(IsOnSignalingThread());
   RTC_DCHECK_GE(cache_lifetime_, 0.0);
 }
 
-rtc::scoped_refptr<const RTCStatsReport> RTCStatsCollector::GetStatsReport() {
-  RTC_DCHECK(IsOnSignalingThread());
-  double now = timing_->TimerNow();
-  if (cached_report_ && now - cache_timestamp_ <= cache_lifetime_)
-    return cached_report_;
-  cache_timestamp_ = now;
-
-  rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create();
-  report->AddStats(ProducePeerConnectionStats());
-
-  cached_report_ = report;
-  return cached_report_;
+void RTCStatsCollector::GetStatsReport(
+    rtc::scoped_refptr<RTCStatsCollectorCallback> callback) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(callback);
+  rtc::scoped_refptr<const RTCStatsReport> cached_report;
+  std::vector<rtc::scoped_refptr<RTCStatsCollectorCallback>> callbacks;
+  {
+    rtc::CritScope cs(&lock_);
+    callbacks_.push_back(callback);
+    double now = timing_->TimerNow();
+    if (cached_report_ && now - cache_timestamp_ <= cache_lifetime_) {
+      // We have a fresh cached report to immediately deliver.
+      cached_report = cached_report_;
+      callbacks = callbacks_;

[hta-webrtc, 2016/08/31 09:57:36]

So you use "callbacks" to pass the list of callbacks to be done out of the lock
section. I find this confusing.

I find this confusing. Can you write a helper function called "ReturnData" or
something that actually does the "for each callback, invoke it" part, and then
call that from both this function and AddPartialResults_s?

callbacks_ is protected by lock_, but has a different lock-required pattern than
cached_report_; you may consider using two locks.

[hbos, 2016/09/01 14:16:16]

Ah, only |num_pending_partial_reports_| and |partial_report_| are touched on
multiple threads and need to be guarded. This simplifies the code a lot.

Wrote helper function DeliverCachedReport().

+      callbacks_.clear();
+    } else if (!num_pending_partial_reports_) {
+      // We don't have a fresh cached report and we are not already in the
+      // process of gathering stats.
+      num_pending_partial_reports_ = 3;
+      invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_,
+          rtc::Bind(
+              &RTCStatsCollector::ProducePartialResultsOnSignalingThread_s,
+              rtc::scoped_refptr<RTCStatsCollector>(this), now));
+      invoker_.AsyncInvoke<void>(RTC_FROM_HERE, worker_thread_,
+          rtc::Bind(&RTCStatsCollector::ProducePartialResultsOnWorkerThread_w,
+              rtc::scoped_refptr<RTCStatsCollector>(this), now));
+      invoker_.AsyncInvoke<void>(RTC_FROM_HERE, network_thread_,
+          rtc::Bind(&RTCStatsCollector::ProducePartialResultsOnNetworkThread_n,
+              rtc::scoped_refptr<RTCStatsCollector>(this), now));
+    }

[hta-webrtc, 2016/08/31 09:57:36]

Where's the "else" branch (the report isn't fresh and
num_pending_partial_reports is > 0)?

My reading is that this will fall through without doing anything because
"callbacks" is empty, but this isn't obvious from the code structure.

[hbos, 2016/09/01 14:16:16]

callback_ has been added to callbacks_ and will be informed when the pending
stats gathering completes and does DeliverCachedReport(). Added a comment.

+  }
+  // Do we have a cached report that is still fresh to deliver?
+  // Then do so without holding the lock.
+  if (cached_report) {
+    for (const rtc::scoped_refptr<RTCStatsCollectorCallback>& callback :
+         callbacks) {
+      callback->OnStatsDelivered(cached_report);
+    }
+    return;
+  }
 }
 
 void RTCStatsCollector::ClearCachedStatsReport() {
-  RTC_DCHECK(IsOnSignalingThread());
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  rtc::CritScope cs(&lock_);
   cached_report_ = nullptr;
 }
 
-bool RTCStatsCollector::IsOnSignalingThread() const {
-  return pc_->session()->signaling_thread()->IsCurrent();
+void RTCStatsCollector::ProducePartialResultsOnSignalingThread_s(
+    double timestamp) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create();
+
+  report->AddStats(ProducePeerConnectionStats_s(timestamp));
+
+  AddPartialResults(report, timestamp);
+}
+
+void RTCStatsCollector::ProducePartialResultsOnWorkerThread_w(
+    double timestamp) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create();
+
+  // TODO(hbos): Gather stats on worker thread.
+
+  AddPartialResults(report, timestamp);
+}
+
+void RTCStatsCollector::ProducePartialResultsOnNetworkThread_n(
+    double timestamp) {
+  RTC_DCHECK(network_thread_->IsCurrent());
+  rtc::scoped_refptr<RTCStatsReport> report = RTCStatsReport::Create();
+
+  // TODO(hbos): Gather stats on network thread.
+
+  AddPartialResults(report, timestamp);
+}
+
+void RTCStatsCollector::AddPartialResults(
+    const rtc::scoped_refptr<RTCStatsReport>& partial_report,
+    double timestamp) {
+  if (!signaling_thread_->IsCurrent()) {
+    invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_,
+        rtc::Bind(&RTCStatsCollector::AddPartialResults_s,
+                  rtc::scoped_refptr<RTCStatsCollector>(this),
+                  partial_report, timestamp));
+    return;
+  }
+  AddPartialResults_s(partial_report, timestamp);
+}
+
+void RTCStatsCollector::AddPartialResults_s(
+    rtc::scoped_refptr<RTCStatsReport> partial_report, double timestamp) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  rtc::scoped_refptr<const RTCStatsReport> result;
+  std::vector<rtc::scoped_refptr<RTCStatsCollectorCallback>> callbacks;
+  {
+    rtc::CritScope cs(&lock_);
+    RTC_DCHECK_GT(num_pending_partial_reports_, 0);
+    if (!partial_report_)
+      partial_report_ = partial_report;
+    else
+      partial_report_->TakeMembersFrom(partial_report);
+    --num_pending_partial_reports_;
+
+    // Was this the last pending partial report, resulting in a complete and
+    // deliverable |result|?
+    if (!num_pending_partial_reports_) {
+      result = cached_report_ = partial_report_;
+      cache_timestamp_ = timestamp;
+      partial_report_ = nullptr;
+      callbacks = callbacks_;
+      callbacks_.clear();
+    }
+  }
+  // Do we have a report to deliver? Then do so without holding the lock.
+  if (result) {
+    for (const rtc::scoped_refptr<RTCStatsCollectorCallback>& callback :
+         callbacks) {
+      callback->OnStatsDelivered(result);
+    }
+  }
 }
 
 std::unique_ptr<RTCPeerConnectionStats>
-RTCStatsCollector::ProducePeerConnectionStats() const {
+RTCStatsCollector::ProducePeerConnectionStats_s(double timestamp) const {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
   // TODO(hbos): If data channels are removed from the peer connection this will
   // yield incorrect counts. Address before closing crbug.com/636818. See
   // https://w3c.github.io/webrtc-stats/webrtc-stats.html#pcstats-dict*.
   uint32_t data_channels_opened = 0;
   const std::vector<rtc::scoped_refptr<DataChannel>>& data_channels =
       pc_->sctp_data_channels();
   for (const rtc::scoped_refptr<DataChannel>& data_channel : data_channels) {
     if (data_channel->state() == DataChannelInterface::kOpen)
       ++data_channels_opened;
   }
   // There is always just one |RTCPeerConnectionStats| so its |id| can be a
   // constant.
   std::unique_ptr<RTCPeerConnectionStats> stats(
-    new RTCPeerConnectionStats("RTCPeerConnection", cache_timestamp_));
+    new RTCPeerConnectionStats("RTCPeerConnection", timestamp));
   stats->data_channels_opened = data_channels_opened;
   stats->data_channels_closed = static_cast<uint32_t>(data_channels.size()) -
                                 data_channels_opened;
   return stats;
 }
 
 }  // namespace webrtc