Support parallel captures from the StackSamplingProfiler.

base/profiler/stack_sampling_profiler.cc

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "base/profiler/stack_sampling_profiler.h"
 
 #include <algorithm>
+#include <queue>
 #include <utility>
 
+#include "base/atomic_sequence_num.h"
+#include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/macros.h"
+#include "base/memory/ptr_util.h"
+#include "base/memory/singleton.h"
+#include "base/memory/weak_ptr.h"
 #include "base/profiler/native_stack_sampler.h"
 #include "base/synchronization/lock.h"
+#include "base/threading/thread.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/timer/elapsed_timer.h"
 
 namespace base {
 
 namespace {
 
-// Used to ensure only one profiler is running at a time.
-LazyInstance<Lock>::Leaky concurrent_profiling_lock = LAZY_INSTANCE_INITIALIZER;
-
 // AsyncRunner ----------------------------------------------------------------
 
 // Helper class to allow a profiler to be run completely asynchronously from the
 // initiator, without being concerned with the profiler's lifetime.
 class AsyncRunner {
  public:
   // Sets up a profiler and arranges for it to be deleted on its completed
   // callback.
   static void Run(PlatformThreadId thread_id,
                   const StackSamplingProfiler::SamplingParams& params,
@@ skipping 115 matching lines @@
 StackSamplingProfiler::CallStackProfile
 StackSamplingProfiler::CallStackProfile::CopyForTesting() const {
   return CallStackProfile(*this);
 }
 
 StackSamplingProfiler::CallStackProfile::CallStackProfile(
     const CallStackProfile& other) = default;
 
 // StackSamplingProfiler::SamplingThread --------------------------------------
 
-StackSamplingProfiler::SamplingThread::SamplingThread(
-    std::unique_ptr<NativeStackSampler> native_sampler,
-    const SamplingParams& params,
-    const CompletedCallback& completed_callback)
-    : native_sampler_(std::move(native_sampler)),
-      params_(params),
-      stop_event_(WaitableEvent::ResetPolicy::AUTOMATIC,
-                  WaitableEvent::InitialState::NOT_SIGNALED),
-      completed_callback_(completed_callback) {}
-
-StackSamplingProfiler::SamplingThread::~SamplingThread() {}
-
-void StackSamplingProfiler::SamplingThread::ThreadMain() {
-  PlatformThread::SetName("Chrome_SamplingProfilerThread");
-
-  // For now, just ignore any requests to profile while another profiler is
-  // working.
-  if (!concurrent_profiling_lock.Get().Try())
+class StackSamplingProfiler::SamplingThread : public Thread {
+ public:
+  struct ActiveCapture {
+    ActiveCapture(PlatformThreadId target,
+                  const SamplingParams& params,
+                  const CompletedCallback& callback,
+                  std::unique_ptr<NativeStackSampler> sampler)
+        : capture_id(next_capture_id_.GetNext()),
+          target(target),
+          params(params),
+          callback(callback),
+          native_sampler(std::move(sampler)),
+          weak_ptr_factory_(this) {}
+    ~ActiveCapture() {}
+
+    // Updates the |next_sample_time| time based on configured parameters.
+    // This will keep a consistent average interval between samples but will
+    // result in constant series of acquisitions, thus nearly locking out the
+    // target thread, if the interval is smaller than the time it takes to
+    // actually acquire the sample. Anything sampling that quickly is going
+    // to be a problem anyway so don't worry about it.
+    bool UpdateNextSampleTime() {

[bcwhite, 2016/12/22 16:12:10]

Really?  I've seen it many times and Alexei has in the past even requested
methods being added to structs if they're actions are solely confined to the
data of those structs.

[Mike Wittman, 2016/12/22 17:38:22]

The style guide says structs should not have any functionality beyond
access/setting the data members:
https://engdoc.corp.google.com/eng/doc/devguide/cpp/styleguide.shtml?cl=head#...

I think this is the right thing to do regardless, so that all the logic dealing
with the context state is collocated. The relationship between sample and
params.samples_per_burst, for example, is defined across both this code and
SamplingThread::PerformRecording.

[bcwhite, 2017/01/05 16:35:58]

Done.

+      if (stopped)
+        return false;
+
+      if (++sample < params.samples_per_burst) {
+        next_sample_time += params.sampling_interval;
+        return true;
+      }
+
+      if (++burst < params.bursts) {
+        sample = 0;
+        next_sample_time += params.burst_interval;
+        return true;
+      }
+
+      return false;
+    }
+
+    WeakPtr<ActiveCapture> GetWeakPtr() {
+      return weak_ptr_factory_.GetWeakPtr();
+    }
+
+    // An identifier for this capture, used to uniquely identify it to outside
+    // interests.
+    const int capture_id;
+
+    Time next_sample_time;
+
+    PlatformThreadId target;
+    SamplingParams params;
+    CompletedCallback callback;
+
+    std::unique_ptr<NativeStackSampler> native_sampler;
+
+    // Counters that indicate the current position along the acquisition.
+    int burst = 0;
+    int sample = 0;
+
+    // Indicates if the capture has been stopped (and reported).
+    bool stopped = false;
+
+    // The time that a profile was started, for calculating the total duration.
+    Time profile_start_time;
+
+    // The captured stack samples. The active profile is always at the back().
+    CallStackProfiles profiles;
+
+   private:
+    static StaticAtomicSequenceNumber next_capture_id_;
+
+    WeakPtrFactory<ActiveCapture> weak_ptr_factory_;
+  };
+
+  // Gets the single instance of this class.
+  static SamplingThread* GetInstance();
+
+  // Starts the thread.
+  void Start();
+
+  // Adds a new ActiveCapture to the thread. This can be called externally
+  // from any thread. This returns an ID that can later be used to stop
+  // the sampling.
+  int Add(std::unique_ptr<ActiveCapture> capture);
+
+  // Stops an active capture based on its ID, forcing it to run its callback
+  // if any data has been collected. This can be called externally from any
+  // thread.
+  void Stop(int id);
+
+ private:
+  SamplingThread();
+  ~SamplingThread() override;
+  friend struct DefaultSingletonTraits<SamplingThread>;
+
+  // These methods are called when a new capture begins, when it is
+  // finished, and for each individual sample, respectively.
+  void BeginRecording(ActiveCapture* capture);
+  void FinishRecording(ActiveCapture* capture);
+  void PerformRecording(ActiveCapture* capture);
+
+  // These methods are tasks that get posted to the internal message queue.
+  void StartCaptureTask(std::unique_ptr<ActiveCapture> capture);
+  void StopCaptureTask(int id);
+  void PerformCaptureTask(std::unique_ptr<ActiveCapture> capture);
+
+  // Thread:
+  void CleanUp() override;
+
+  static constexpr int kMinimumThreadRunTimeSeconds = 60;
+
+  // The task-runner for the sampling thread. This is saved so that it can
+  // be freely used by any calling thread.
+  scoped_refptr<SingleThreadTaskRunner> task_runner_;
+  Lock task_runner_lock_;
+
+  // A map of IDs to active captures. This is a weak-pointer because it's
+  // possible that objects are deleted because their collection is complete.
+  std::map<int, WeakPtr<ActiveCapture>> active_captures_;

[Mike Wittman, 2016/12/15 20:37:53]

I'm not seeing the benefit of using WeakPtr<ActiveCapture> here, rather than
unique_ptr<ActiveCapture>, and delete by erasing the id. Can you explain?

Seems like unique_ptr could avoid the whole WeakPtr machinery, make it much
easier to reason about ownership, and reduce the amount of state in the system.

[bcwhite, 2016/12/21 16:39:10]

The ownership was with the posted tasks so other pointers needed to be weak. 
But it didn't work out like I was thinking so went another way.  Since the
single instance of this class never gets destructed, ownership can stay in this
map and raw pointers passed to posted tasks.

[Mike Wittman, 2016/12/21 19:38:41]

Can we pass the id rather than the raw pointer? Paying the small overhead of
looking up the context in the map is IMHO well worth the benefit of not having
to consider whether there are lifetime issues between the context references in
the posted tasks and active_captures_.

[bcwhite, 2016/12/22 16:12:10]

Done.

+
+  DISALLOW_COPY_AND_ASSIGN(SamplingThread);
+};
+
+StaticAtomicSequenceNumber
+    StackSamplingProfiler::SamplingThread::ActiveCapture::next_capture_id_;
+
+StackSamplingProfiler::SamplingThread::SamplingThread()
+    : Thread("Chrome_SamplingProfilerThread") {}
+
+StackSamplingProfiler::SamplingThread::~SamplingThread() {
+  Thread::Stop();
+}
+
+StackSamplingProfiler::SamplingThread*
+StackSamplingProfiler::SamplingThread::GetInstance() {
+  return Singleton<SamplingThread>::get();
+}
+
+void StackSamplingProfiler::SamplingThread::Start() {
+  Thread::Options options;
+  // Use a higher priority for a more accurate sampling interval.
+  options.priority = ThreadPriority::DISPLAY;
+  Thread::StartWithOptions(options);
+}
+
+int StackSamplingProfiler::SamplingThread::Add(
+    std::unique_ptr<ActiveCapture> capture) {
+  int id = capture->capture_id;
+
+  AutoLock lock(task_runner_lock_);
+  if (!task_runner_) {
+    // The thread is not running. Start it and get associated runner.
+    Start();
+    task_runner_ = task_runner();
+    DCHECK(task_runner_);

[Mike Wittman, 2016/12/15 20:37:53]

Remove this DCHECK? Seems like it's just verifying Thread's documented behavior
at this point.

[bcwhite, 2016/12/21 16:39:10]

Done.

+  }
+
+  bool success = task_runner_->PostTask(
+      FROM_HERE, Bind(&SamplingThread::StartCaptureTask, Unretained(this),
+                      Passed(&capture)));
+  DCHECK(success);

[Mike Wittman, 2016/12/15 20:37:53]

Remove this one as well? I don't think this will ever fail, and if it does, it's
an issue internal to the message loop/task runner.

[bcwhite, 2016/12/21 16:39:10]

Done.

+
+  return id;
+}
+
+void StackSamplingProfiler::SamplingThread::Stop(int id) {
+  AutoLock lock(task_runner_lock_);
+  if (!task_runner_)
+    return;  // Everything has already stopped.
+
+  // This can fail if the thread were to exit between acquisition of the task
+  // runner above and the call below. In that case, however, everything has
+  // stopped so there's no need to try to stop it.
+  task_runner_->PostTask(
+      FROM_HERE, Bind(&SamplingThread::StopCaptureTask, Unretained(this), id));
+}
+
+void StackSamplingProfiler::SamplingThread::BeginRecording(
+    ActiveCapture* capture) {
+  DCHECK(capture->native_sampler);
+}
+
+void StackSamplingProfiler::SamplingThread::FinishRecording(
+    ActiveCapture* capture) {
+  DCHECK(!capture->stopped);
+  capture->stopped = true;
+
+  // If there is no duration for the final profile (because it was stopped),
+  // calculated it now.
+  if (!capture->profiles.empty() &&
+      capture->profiles.back().profile_duration == TimeDelta()) {
+    capture->profiles.back().profile_duration =
+        Time::Now() - capture->profile_start_time;
+  }
+
+  // Run the associated callback, passing the captured profiles. It's okay to
+  // move them because this capture is about to be deleted.
+  capture->callback.Run(std::move(capture->profiles));
+}

[Mike Wittman, 2016/12/15 20:37:53]

Why not erase the capture from active_captures_ at the end of this function?

[bcwhite, 2016/12/21 16:39:10]

Done.

+
+void StackSamplingProfiler::SamplingThread::PerformRecording(
+    ActiveCapture* capture) {
+  DCHECK(!capture->stopped);
+
+  // If this is the first sample of a burst, a new Profile needs to be created
+  // and filled.
+  if (capture->sample == 0) {
+    capture->profiles.push_back(CallStackProfile());
+    CallStackProfile& profile = capture->profiles.back();
+    profile.sampling_period = capture->params.sampling_interval;
+    capture->profile_start_time = Time::Now();
+    capture->native_sampler->ProfileRecordingStarting(&profile.modules);
+  }
+
+  // The currently active profile being acptured.
+  CallStackProfile& profile = capture->profiles.back();
+
+  // Capture a single sample.
+  profile.samples.push_back(Sample());
+  capture->native_sampler->RecordStackSample(&profile.samples.back());
+
+  // If this is the last sample of a burst, record the total time.
+  if (capture->sample == capture->params.samples_per_burst - 1) {
+    profile.profile_duration = Time::Now() - capture->profile_start_time;
+    capture->native_sampler->ProfileRecordingStopped();
+  }
+}
+
+void StackSamplingProfiler::SamplingThread::StartCaptureTask(
+    std::unique_ptr<ActiveCapture> capture) {
+  active_captures_.insert(
+      std::make_pair(capture->capture_id, capture->GetWeakPtr()));
+  BeginRecording(capture.get());
+  bool success = task_runner()->PostDelayedTask(

[Mike Wittman, 2016/12/15 20:37:53]

While the task_runner() accesses on the profiler thread don't need to be guarded
by the lock, that won't be at all obvious to the casual reader. Can we
encapsulate this subtlety within functions (e.g.
GetTaskRunner/GetOrCreateTaskRunner), handling the locking (if necessary) there?

[bcwhite, 2016/12/21 16:39:10]

Such a method would have to fetch the current thread-id and compare it to the id
of the sampling thread to know whether it needs to use the (lock-protected)
member variable or call Thread::task_runner().  Unfortunately, getting the
current thread-id can be a system call which means we probably shouldn't do it
unless necessary.

[Mike Wittman, 2016/12/21 19:38:41]

Fetching the thread id on Windows is cheap: it's stored in the Thread
Environment Block, which is accessed via a segment register and doesn't require
a syscall. I took a look at glibc and it also does not need a syscall to get the
thread id.

[bcwhite, 2016/12/22 16:12:10]

Linux does a direct syscall():
https://cs.chromium.org/chromium/src/base/threading/platform_thread_posix.cc?...

[Mike Wittman, 2016/12/22 17:38:22]

Ah, missed that the Linux implementation doesn't go through pthread_self().

Given that Linux syscall overhead is in the 10's to 100's of ns, and the task
runner likely will be accessed at most a handful of times every 100ms, I think
we can afford to pay this minimal overhead to make the code less tricky and more
robust to changes.

[bcwhite, 2017/01/05 16:35:58]

I started down this path but it ends up requiring the lock every access.  I
can't compare the current thread-id to the sampling thread's ID without it
waiting for that ID to be valid, which only happens after it has been started.

But I don't want to start it until needed and the only way to tell if its needed
is to call IsRunning() or check the task_runner_ local variable to see if it's
set, both of which require a lock.

Getting the thread's ID does an event-wait so that'll need to be cached and
locked as well, though it can probably share the same lock as task_runner_.

Acquiring a lock isn't expensive but would be required with every sample and
it's not necessary when we already know we're running on the sampling thread.

I think a comment is the better way to make things obvious to the casual reader.

[Mike Wittman, 2017/01/05 21:08:39]

Yeah, it's probably not worth going to the extent of acquiring the lock on the
profiler thread.
I think it would be better to create and use explicit functions for getting the
task runner on either the sampling thread or on other threads (e.g.
GetTaskRunnerOnOtherThread GetOrCreateTaskRunnerOnOtherThread,
GetTaskRunnerOnOwnThread). That will still encapsulate the locking and will
force reviewers and developers to consider the appropriate method for getting
the task runner when making future changes. We should be able to DCHECK in these
functions to validate correct usage as well.

[bcwhite, 2017/01/05 22:04:22]

Trying this but there are issues.

GetOrCreate isn't enough because Stop() needs to know the value without creating
-- we don't want to start the sampling thread there.  It could access
task_runner_ directly (while locked) while it does now but that means multiple
methods accessing task_runner_ which is what creating the method was supposed to
avoid.

If I leave the "create" part in Add() then it would be one to access
task_runner_ directly.

Given that only two methods access task_runner_ currently, there's no win here.

Similarly, the GetFromSamplingThread() method ends up just a wrapper around
Thread::task_runner() since the desired DCHECK is already in
Thread::task_runner().

+      FROM_HERE, Bind(&SamplingThread::PerformCaptureTask, Unretained(this),
+                      Passed(&capture)),
+      capture->params.initial_delay);
+  DCHECK(success);

[Mike Wittman, 2016/12/15 20:37:53]

I think this can be removed for the same reasons as above.

[bcwhite, 2016/12/21 16:39:10]

Done.

+}
+
+void StackSamplingProfiler::SamplingThread::StopCaptureTask(int id) {
+  auto found = active_captures_.find(id);
+  if (found == active_captures_.end())
+    return;  // Gone and forgotten.
+
+  ActiveCapture* capture = found->second.get();
+  if (!capture)
+    return;  // Gone but not forgotten.
+
+  if (capture->stopped)
     return;
 
-  CallStackProfiles profiles;
-  CollectProfiles(&profiles);
-  concurrent_profiling_lock.Get().Release();
-  completed_callback_.Run(std::move(profiles));
-}
-
-// Depending on how long the sampling takes and the length of the sampling
-// interval, a burst of samples could take arbitrarily longer than
-// samples_per_burst * sampling_interval. In this case, we (somewhat
-// arbitrarily) honor the number of samples requested rather than strictly
-// adhering to the sampling intervals. Once we have established users for the
-// StackSamplingProfiler and the collected data to judge, we may go the other
-// way or make this behavior configurable.
-void StackSamplingProfiler::SamplingThread::CollectProfile(
-    CallStackProfile* profile,
-    TimeDelta* elapsed_time,
-    bool* was_stopped) {
-  ElapsedTimer profile_timer;
-  native_sampler_->ProfileRecordingStarting(&profile->modules);
-  profile->sampling_period = params_.sampling_interval;
-  *was_stopped = false;
-  TimeDelta previous_elapsed_sample_time;
-  for (int i = 0; i < params_.samples_per_burst; ++i) {
-    if (i != 0) {
-      // Always wait, even if for 0 seconds, so we can observe a signal on
-      // stop_event_.
-      if (stop_event_.TimedWait(
-              std::max(params_.sampling_interval - previous_elapsed_sample_time,
-                       TimeDelta()))) {
-        *was_stopped = true;
-        break;
-      }
-    }
-    ElapsedTimer sample_timer;
-    profile->samples.push_back(Sample());
-    native_sampler_->RecordStackSample(&profile->samples.back());
-    previous_elapsed_sample_time = sample_timer.Elapsed();
-  }
-
-  *elapsed_time = profile_timer.Elapsed();
-  profile->profile_duration = *elapsed_time;
-  native_sampler_->ProfileRecordingStopped();
-}
-
-// In an analogous manner to CollectProfile() and samples exceeding the expected
-// total sampling time, bursts may also exceed the burst_interval. We adopt the
-// same wait-and-see approach here.
-void StackSamplingProfiler::SamplingThread::CollectProfiles(
-    CallStackProfiles* profiles) {
-  if (stop_event_.TimedWait(params_.initial_delay))
+  FinishRecording(capture);
+}
+
+void StackSamplingProfiler::SamplingThread::PerformCaptureTask(
+    std::unique_ptr<ActiveCapture> capture) {
+  DCHECK(capture);
+  // Don't do anything if the capture has already stopped.
+  if (capture->stopped)
     return;
 
-  TimeDelta previous_elapsed_profile_time;
-  for (int i = 0; i < params_.bursts; ++i) {
-    if (i != 0) {
-      // Always wait, even if for 0 seconds, so we can observe a signal on
-      // stop_event_.
-      if (stop_event_.TimedWait(
-              std::max(params_.burst_interval - previous_elapsed_profile_time,
-                       TimeDelta())))
-        return;
-    }
-
-    CallStackProfile profile;
-    bool was_stopped = false;
-    CollectProfile(&profile, &previous_elapsed_profile_time, &was_stopped);
-    if (!profile.samples.empty())
-      profiles->push_back(std::move(profile));
-
-    if (was_stopped)
-      return;
-  }
-}
-
-void StackSamplingProfiler::SamplingThread::Stop() {
-  stop_event_.Signal();
+  // Handle first-run with no "next time".
+  if (capture->next_sample_time == Time())
+    capture->next_sample_time = Time::Now();
+
+  // Do the collection of a single sample.
+  PerformRecording(capture.get());
+
+  // Update the time of the next capture.
+  if (capture->UpdateNextSampleTime()) {
+    // Place the updated entry back on the queue.
+    bool success = task_runner()->PostDelayedTask(
+        FROM_HERE, Bind(&SamplingThread::PerformCaptureTask, Unretained(this),
+                        Passed(&capture)),
+        std::max(capture->next_sample_time - Time::Now(), TimeDelta()));
+    DCHECK(success);
+  } else {
+    // All capturing has completed so finish the collection. Let object expire.
+    FinishRecording(capture.get());
+  }
+}
+
+void StackSamplingProfiler::SamplingThread::CleanUp() {
+  // Clear out the list of active captures. The sampling thread is exited
+  // so they must all have finished.
+  active_captures_.clear();
+
+  // Let the parent clean up.
+  Thread::CleanUp();
 }
 
 // StackSamplingProfiler ------------------------------------------------------
 
 subtle::Atomic32 StackSamplingProfiler::process_phases_ = 0;
 
 StackSamplingProfiler::SamplingParams::SamplingParams()
     : initial_delay(TimeDelta::FromMilliseconds(0)),
       bursts(1),
       burst_interval(TimeDelta::FromMilliseconds(10000)),
@@ skipping 11 matching lines @@
     PlatformThreadId thread_id,
     const SamplingParams& params,
     const CompletedCallback& callback,
     NativeStackSamplerTestDelegate* test_delegate)
     : thread_id_(thread_id), params_(params), completed_callback_(callback),
       test_delegate_(test_delegate) {
 }
 
 StackSamplingProfiler::~StackSamplingProfiler() {
   Stop();
-  if (!sampling_thread_handle_.is_null())
-    PlatformThread::Join(sampling_thread_handle_);
 }
 
 // static
 void StackSamplingProfiler::StartAndRunAsync(
     PlatformThreadId thread_id,
     const SamplingParams& params,
     const CompletedCallback& callback) {
   CHECK(ThreadTaskRunnerHandle::Get());
   AsyncRunner::Run(thread_id, params, callback);
 }
 
 void StackSamplingProfiler::Start() {
   if (completed_callback_.is_null())
     return;
 
   std::unique_ptr<NativeStackSampler> native_sampler =
       NativeStackSampler::Create(thread_id_, &RecordAnnotations,
                                  test_delegate_);
   if (!native_sampler)
     return;
 
-  sampling_thread_.reset(new SamplingThread(std::move(native_sampler), params_,
-                                            completed_callback_));
-  if (!PlatformThread::Create(0, sampling_thread_.get(),
-                              &sampling_thread_handle_))
-    sampling_thread_.reset();
+  capture_id_ = SamplingThread::GetInstance()->Add(
+      MakeUnique<SamplingThread::ActiveCapture>(
+          thread_id_, params_, completed_callback_, std::move(native_sampler)));
 }
 
 void StackSamplingProfiler::Stop() {
-  if (sampling_thread_)
-    sampling_thread_->Stop();
+  SamplingThread::GetInstance()->Stop(capture_id_);
 }
 
 // static
 void StackSamplingProfiler::SetProcessPhase(int phase) {
   DCHECK_LE(0, phase);
   DCHECK_GT(static_cast<int>(sizeof(process_phases_) * 8), phase);
   DCHECK_EQ(0, subtle::NoBarrier_Load(&process_phases_) & (1 << phase));
   ChangeAtomicFlags(&process_phases_, 1 << phase, 0);
 }
 
@@ skipping 45 matching lines @@
 }
 
 bool operator<(const StackSamplingProfiler::Frame &a,
                const StackSamplingProfiler::Frame &b) {
   return (a.module_index < b.module_index) ||
       (a.module_index == b.module_index &&
        a.instruction_pointer < b.instruction_pointer);
 }
 
 }  // namespace base