PeerConnectionDependencyFactory: expose thread performance stats.

This change adds thread performance stats for the Network, Signaling,
and Worker threads. New histogram UMAs added:

WebRTC.PeerConnection.Latency.Worker
WebRTC.PeerConnection.Latency.Signaling
WebRTC.PeerConnection.Latency.Network
WebRTC.PeerConnection.Duration.Worker
WebRTC.PeerConnection.Duration.Signaling
WebRTC.PeerConnection.Duration.Network

Bug: chromium:1184611
Change-Id: I7ac3ff52dd677744aea1189517a9bc8862a051ea
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2889614
Commit-Queue: Sergey Ulanov <sergeyu@chromium.org>
Auto-Submit: Markus Handell <handellm@google.com>
Reviewed-by: Sergey Ulanov <sergeyu@chromium.org>
Reviewed-by: Henrik Boström <hbos@chromium.org>
Cr-Commit-Position: refs/heads/master@{#885613}

jingle/glue/thread_wrapper.cc

@@ -11,16 +11,81 @@
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/lazy_instance.h"
+#include "base/memory/weak_ptr.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/sequence_checker.h"
 #include "base/stl_util.h"
+#include "base/strings/strcat.h"
+#include "base/strings/string_piece_forward.h"
+#include "base/thread_annotations.h"
+#include "base/threading/platform_thread.h"
+#include "base/threading/thread.h"
 #include "base/threading/thread_local.h"
 #include "base/threading/thread_task_runner_handle.h"
+#include "base/time/time.h"
 #include "base/trace_event/trace_event.h"
 #include "third_party/webrtc/rtc_base/physical_socket_server.h"
 
 namespace jingle_glue {
+namespace {
+static constexpr base::TimeDelta kTaskLatencySampleDuration =
+    base::TimeDelta::FromSeconds(3);
+}
+
+// Class intended to conditionally live for the duration of JingleThreadWrapper
+// that periodically captures task latencies (definition in docs for
+// SetLatencyAndTaskDurationCallbacks).
+class JingleThreadWrapper::PostTaskLatencySampler {
+ public:
+  PostTaskLatencySampler(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+      SampledDurationCallback task_latency_callback)
+      : task_runner_(task_runner),
+        task_latency_callback_(std::move(task_latency_callback)) {
+    ScheduleDelayedSample();
+  }
+
+  bool ShouldSampleNextTaskDuration() {
+    DCHECK_CALLED_ON_VALID_SEQUENCE(current_);
+    bool time_to_sample = should_sample_next_task_duration_;
+    should_sample_next_task_duration_ = false;
+    return time_to_sample;
+  }
+
+ private:
+  void ScheduleDelayedSample() {
+    DCHECK_CALLED_ON_VALID_SEQUENCE(current_);
+    task_runner_->PostDelayedTask(
+        FROM_HERE,
+        base::BindOnce(&PostTaskLatencySampler::TakeSample,
+                       base::Unretained(this)),
+        kTaskLatencySampleDuration);
+  }
+
+  void TakeSample() {
+    DCHECK_CALLED_ON_VALID_SEQUENCE(current_);
+    task_runner_->PostTask(
+        FROM_HERE,
+        base::BindOnce(&PostTaskLatencySampler::FinishSample,
+                       base::Unretained(this), base::TimeTicks::Now()));
+  }
+
+  void FinishSample(base::TimeTicks post_timestamp) {
+    DCHECK_CALLED_ON_VALID_SEQUENCE(current_);
+    task_latency_callback_.Run(base::TimeTicks::Now() - post_timestamp);
+    ScheduleDelayedSample();
+    should_sample_next_task_duration_ = true;
+  }
+
+  SEQUENCE_CHECKER(current_);
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  base::RepeatingCallback<void(base::TimeDelta)> task_latency_callback_
+      GUARDED_BY_CONTEXT(current_);
+  bool should_sample_next_task_duration_ GUARDED_BY_CONTEXT(current_) = false;
+};
 
 struct JingleThreadWrapper::PendingSend {
-  PendingSend(const rtc::Message& message_value)
+  explicit PendingSend(const rtc::Message& message_value)
       : sending_thread(JingleThreadWrapper::current()),
         message(message_value),
         done_event(base::WaitableEvent::ResetPolicy::MANUAL,
@@ -64,6 +129,13 @@ JingleThreadWrapper* JingleThreadWrapper::current() {
   return g_jingle_thread_wrapper.Get().Get();
 }
 
+void JingleThreadWrapper::SetLatencyAndTaskDurationCallbacks(
+    SampledDurationCallback task_latency_callback,
+    SampledDurationCallback task_duration_callback) {
+  task_latency_callback_ = std::move(task_latency_callback);
+  task_duration_callback_ = std::move(task_duration_callback);
+}
+
 JingleThreadWrapper::JingleThreadWrapper(
     scoped_refptr<base::SingleThreadTaskRunner> task_runner)
     : Thread(std::make_unique<rtc::PhysicalSocketServer>()),
@@ -260,6 +332,23 @@ void JingleThreadWrapper::PostTaskInternal(const rtc::Location& posted_from,
 }
 
 void JingleThreadWrapper::RunTask(int task_id) {
+  if (!latency_sampler_ && !task_latency_callback_.is_null()) {
+    latency_sampler_ = std::make_unique<PostTaskLatencySampler>(
+        task_runner_, std::move(task_latency_callback_));
+  }
+  absl::optional<base::TimeTicks> task_start_timestamp_;
+  if (!task_duration_callback_.is_null() && latency_sampler_ &&
+      latency_sampler_->ShouldSampleNextTaskDuration()) {
+    task_start_timestamp_ = base::TimeTicks::Now();
+  }
+  RunTaskInternal(task_id);
+  if (task_start_timestamp_.has_value()) {
+    task_duration_callback_.Run(base::TimeTicks::Now() -
+                                *task_start_timestamp_);
+  }
+}
+
+void JingleThreadWrapper::RunTaskInternal(int task_id) {
   bool have_message = false;
   rtc::Message message;
   {

jingle/glue/thread_wrapper.h

@@ -11,6 +11,7 @@
 #include <map>
 #include <memory>
 
+#include "base/callback_forward.h"
 #include "base/compiler_specific.h"
 #include "base/macros.h"
 #include "base/memory/weak_ptr.h"
@@ -18,6 +19,7 @@
 #include "base/synchronization/lock.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/task/current_thread.h"
+#include "base/time/time.h"
 #include "third_party/webrtc/rtc_base/thread.h"
 
 namespace jingle_glue {
@@ -36,6 +38,11 @@ namespace jingle_glue {
 class JingleThreadWrapper : public base::CurrentThread::DestructionObserver,
                             public rtc::Thread {
  public:
+  // A repeating callback whose TimeDelta argument indicates a duration sample.
+  // What the duration represents is contextual.
+  using SampledDurationCallback =
+      base::RepeatingCallback<void(base::TimeDelta)>;
+
   // Create JingleThreadWrapper for the current thread if it hasn't been created
   // yet. The thread wrapper is destroyed automatically when the current
   // MessageLoop is destroyed.
@@ -50,6 +57,21 @@ class JingleThreadWrapper : public base::CurrentThread::DestructionObserver,
   // exist.
   static JingleThreadWrapper* current();
 
+  // Sets task latency & duration sample callbacks intended to gather UMA
+  // statistics. Samples are acquired periodically every several seconds by
+  // JingleThreadWrapper. In this context,
+  // * task latency is defined as the duration between the moment a task is
+  //   scheduled from JingleThreadWrapper's task runner, and the moment
+  //   it begins running.
+  // * task duration is defined as the duration between the moment the
+  //   JingleThreadWrapper begins running a task and the moment it ends
+  //   executing it. It only measures durations of tasks posted to rtc::Thread.
+  // The passed callbacks are called in the JingleThreadWrapper's task runner
+  // context.
+  void SetLatencyAndTaskDurationCallbacks(
+      SampledDurationCallback task_latency_callback,
+      SampledDurationCallback task_duration_callback);
+
   ~JingleThreadWrapper() override;
 
   // Sets whether the thread can be used to send messages
@@ -105,6 +127,7 @@ class JingleThreadWrapper : public base::CurrentThread::DestructionObserver,
  private:
   typedef std::map<int, rtc::Message> MessagesQueue;
   struct PendingSend;
+  class PostTaskLatencySampler;
 
   explicit JingleThreadWrapper(
      scoped_refptr<base::SingleThreadTaskRunner> task_runner);
@@ -115,6 +138,7 @@ class JingleThreadWrapper : public base::CurrentThread::DestructionObserver,
                         uint32_t message_id,
                         rtc::MessageData* data);
   void RunTask(int task_id);
+  void RunTaskInternal(int task_id);
   void ProcessPendingSends();
 
   // Task runner used to execute messages posted on this thread.
@@ -128,6 +152,9 @@ class JingleThreadWrapper : public base::CurrentThread::DestructionObserver,
   MessagesQueue messages_;
   std::list<PendingSend*> pending_send_messages_;
   base::WaitableEvent pending_send_event_;
+  std::unique_ptr<PostTaskLatencySampler> latency_sampler_;
+  SampledDurationCallback task_latency_callback_;
+  SampledDurationCallback task_duration_callback_;
 
   base::WeakPtr<JingleThreadWrapper> weak_ptr_;
   base::WeakPtrFactory<JingleThreadWrapper> weak_ptr_factory_{this};

third_party/blink/renderer/modules/peerconnection/peer_connection_dependency_factory.cc

@@ -11,11 +11,15 @@
 #include <utility>
 #include <vector>
 
+#include "base/callback.h"
+#include "base/callback_forward.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/macros.h"
+#include "base/metrics/histogram_macros.h"
 #include "base/synchronization/waitable_event.h"
+#include "base/time/time.h"
 #include "build/build_config.h"
 #include "crypto/openssl_util.h"
 #include "jingle/glue/thread_wrapper.h"
@@ -53,6 +57,7 @@
 #include "third_party/blink/renderer/platform/scheduler/public/post_cross_thread_task.h"
 #include "third_party/blink/renderer/platform/wtf/cross_thread_copier.h"
 #include "third_party/blink/renderer/platform/wtf/cross_thread_functional.h"
+#include "third_party/blink/renderer/platform/wtf/functional.h"
 #include "third_party/webrtc/api/call/call_factory_interface.h"
 #include "third_party/webrtc/api/peer_connection_interface.h"
 #include "third_party/webrtc/api/rtc_event_log/rtc_event_log_factory.h"
@@ -66,6 +71,15 @@
 #include "third_party/webrtc/rtc_base/ssl_adapter.h"
 #include "third_party/webrtc_overrides/task_queue_factory.h"
 
+namespace WTF {
+template <>
+struct CrossThreadCopier<base::RepeatingCallback<void(base::TimeDelta)>>
+    : public CrossThreadCopierPassThrough<
+          base::RepeatingCallback<void(base::TimeDelta)>> {
+  STATIC_ONLY(CrossThreadCopier);
+};
+}  // namespace WTF
+
 namespace blink {
 
 namespace {
@@ -100,7 +114,7 @@ bool IsValidPortRange(uint16_t min_port, uint16_t max_port) {
 // class.
 class ProxyAsyncResolverFactory final : public webrtc::AsyncResolverFactory {
  public:
-  ProxyAsyncResolverFactory(IpcPacketSocketFactory* ipc_psf)
+  explicit ProxyAsyncResolverFactory(IpcPacketSocketFactory* ipc_psf)
       : ipc_psf_(ipc_psf) {
     DCHECK(ipc_psf);
   }
@@ -149,10 +163,14 @@ class PeerConnectionStaticDeps {
     if (!worker_thread_) {
       PostCrossThreadTask(
           *chrome_worker_thread_->task_runner(), FROM_HERE,
-          CrossThreadBindOnce(&PeerConnectionStaticDeps::InitializeOnThread,
-                              CrossThreadUnretained(this),
-                              CrossThreadUnretained(&worker_thread_),
-                              CrossThreadUnretained(&init_worker_event)));
+          CrossThreadBindOnce(
+              &PeerConnectionStaticDeps::InitializeOnThread,
+              CrossThreadUnretained(&worker_thread_),
+              CrossThreadUnretained(&init_worker_event),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskLatencyWorker)),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskDurationWorker))));
     }
     return init_worker_event;
   }
@@ -161,10 +179,14 @@ class PeerConnectionStaticDeps {
     if (!network_thread_) {
       PostCrossThreadTask(
           *chrome_network_thread_.task_runner(), FROM_HERE,
-          CrossThreadBindOnce(&PeerConnectionStaticDeps::InitializeOnThread,
-                              CrossThreadUnretained(this),
-                              CrossThreadUnretained(&network_thread_),
-                              CrossThreadUnretained(&init_network_event)));
+          CrossThreadBindOnce(
+              &PeerConnectionStaticDeps::InitializeOnThread,
+              CrossThreadUnretained(&network_thread_),
+              CrossThreadUnretained(&init_network_event),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskLatencyNetwork)),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskDurationNetwork))));
     }
     return init_network_event;
   }
@@ -173,10 +195,14 @@ class PeerConnectionStaticDeps {
     if (!signaling_thread_) {
       PostCrossThreadTask(
           *chrome_signaling_thread_.task_runner(), FROM_HERE,
-          CrossThreadBindOnce(&PeerConnectionStaticDeps::InitializeOnThread,
-                              CrossThreadUnretained(this),
-                              CrossThreadUnretained(&signaling_thread_),
-                              CrossThreadUnretained(&init_signaling_event)));
+          CrossThreadBindOnce(
+              &PeerConnectionStaticDeps::InitializeOnThread,
+              CrossThreadUnretained(&signaling_thread_),
+              CrossThreadUnretained(&init_signaling_event),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskLatencySignaling)),
+              ConvertToBaseRepeatingCallback(CrossThreadBindRepeating(
+                  PeerConnectionStaticDeps::LogTaskDurationSignaling))));
     }
     return init_signaling_event;
   }
@@ -192,9 +218,47 @@ class PeerConnectionStaticDeps {
   base::Thread& GetChromeNetworkThread() { return chrome_network_thread_; }
 
  private:
-  void InitializeOnThread(rtc::Thread** thread, base::WaitableEvent* event) {
+  static void LogTaskLatencyWorker(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Latency.Worker", sample,
+                               base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+  static void LogTaskDurationWorker(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Duration.Worker", sample,
+                               base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+  static void LogTaskLatencyNetwork(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Latency.Network", sample,
+                               base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+  static void LogTaskDurationNetwork(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Duration.Network", sample,
+                               base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+  static void LogTaskLatencySignaling(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Latency.Signaling",
+                               sample, base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+  static void LogTaskDurationSignaling(base::TimeDelta sample) {
+    UMA_HISTOGRAM_CUSTOM_TIMES("WebRTC.PeerConnection.Duration.Signaling",
+                               sample, base::TimeDelta::FromMicroseconds(1),
+                               base::TimeDelta::FromSeconds(10), 50);
+  }
+
+  static void InitializeOnThread(
+      rtc::Thread** thread,
+      base::WaitableEvent* event,
+      base::RepeatingCallback<void(base::TimeDelta)> latency_callback,
+      base::RepeatingCallback<void(base::TimeDelta)> duration_callback) {
     jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
     jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
+    jingle_glue::JingleThreadWrapper::current()
+        ->SetLatencyAndTaskDurationCallbacks(std::move(latency_callback),
+                                             std::move(duration_callback));
     if (!*thread) {
       *thread = jingle_glue::JingleThreadWrapper::current();
       event->Signal();