Support parallel captures from the StackSamplingProfiler.

base/profiler/stack_sampling_profiler.cc

Index: base/profiler/stack_sampling_profiler.cc
diff --git a/base/profiler/stack_sampling_profiler.cc b/base/profiler/stack_sampling_profiler.cc
index d77858427edd6c922a17df7f29e985551640e99d..24db08b62e42d491bb6719531adea552705c4177 100644
--- a/base/profiler/stack_sampling_profiler.cc
+++ b/base/profiler/stack_sampling_profiler.cc
@@ -5,16 +5,21 @@
 #include "base/profiler/stack_sampling_profiler.h"
 
 #include <algorithm>
+#include <queue>
 #include <utility>
 
+#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/profiler/native_stack_sampler.h"
 #include "base/synchronization/lock.h"
+#include "base/threading/simple_thread.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/timer/elapsed_timer.h"
 
@@ -22,9 +27,6 @@ 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
@@ -160,102 +162,288 @@ StackSamplingProfiler::CallStackProfile::CallStackProfile(
 
 // 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) {}
+class StackSamplingProfiler::SamplingThread : public SimpleThread {
+ public:
+  struct ActiveCapture {
+    ActiveCapture(PlatformThreadId target,
+                  const SamplingParams& params,
+                  const CompletedCallback& callback,
+                  std::unique_ptr<NativeStackSampler> sampler)
+        : capture_id(subtle::NoBarrier_AtomicIncrement(&next_capture_id_, 1)),
+          target(target),
+          params(params),
+          callback(callback),
+          native_sampler(std::move(sampler)) {}
+    ~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() {
+      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;
+    }
+
+    // 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;
+
+    // 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 subtle::AtomicWord next_capture_id_;

[Mike Wittman, 2016/12/06 21:04:58]

We probably can avoid need for a thread-safe id by identifying the ActiveCapture
by its address (e.g. as an opaque void*).

[bcwhite, 2016/12/07 15:15:30]

My concern with that is that addresses may be reused.  A capture could start and
then complete, getting freed.  A new capture could start and reuse the same
address, reasonably likely given that the allocation is the exact same number of
bytes as the free'd block.  Then a stop-request for the first one could be made
and cause the new one to stop.

The incrementing integer will also repeat but not for a long, long time.

[Mike Wittman, 2016/12/07 16:25:02]

Yes, care would need to be taken to ensure the StackSamplingProfiler doesn't
retain the address beyond when the object is deleted. This may be feasible,
depending on the synchronization we ultimately have in place with the threads
owning the StackSamplingProfilers. Let's reconsider once we have something
closer to a final implementation.

[Mike Wittman, 2016/12/07 17:20:42]

Also, the current implementation can use base::StaticAtomicSequenceNumber.

[bcwhite, 2016/12/15 18:07:50]

Done.

+  };
+
+  // Gets the single instance of this class.
+  static SamplingThread* GetInstance();
+
+  // 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 Start(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 is started, when it is
+  // finished, and for each individual sample, respectively.
+  void StartCapture(ActiveCapture* capture);
+  void FinishCapture(ActiveCapture* capture);
+  void PerformCapture(ActiveCapture* capture);
+
+  // SimpleThread:
+  void Run() override;
+
+  // Compares two active capture pointers for which is less. This is used to
+  // order the priority queue.
+  static bool ActiveCaptureOrder(std::unique_ptr<ActiveCapture>& lhs,
+                                 std::unique_ptr<ActiveCapture>& rhs) {
+    // Null pointers always have the greater priority so they can be cleared
+    // from the queue.
+    if (!lhs)
+      return false;
+    if (!rhs)
+      return true;
+
+    // Compare the next sample times. The one farthest out is the lesser.
+    return lhs->next_sample_time > rhs->next_sample_time;
+  }
+
+  // A vector of active captures. Entries are managed using push_heap() and
+  // pop_heap() to keep it as a priority queue. Because of the need to iterate
+  // over the entries without popping them, a std::priority_queue is
+  // insufficient.
+  std::vector<std::unique_ptr<ActiveCapture>> active_captures_;
+
+  // This signals that something has changed with the capture configuration.
+  // The capture thread will check for changes as soon as possible.
+  WaitableEvent capture_change_;
+  Lock capture_change_lock_;
+  std::vector<std::unique_ptr<ActiveCapture>> start_captures_;
+  std::vector<int> stop_captures_;
+
+  DISALLOW_COPY_AND_ASSIGN(SamplingThread);
+};
+
+subtle::AtomicWord
+    StackSamplingProfiler::SamplingThread::ActiveCapture::next_capture_id_ = 0;
+
+StackSamplingProfiler::SamplingThread::SamplingThread()
+    : SimpleThread("Chrome_SamplingProfilerThread", SimpleThread::Options()),
+      capture_change_(WaitableEvent::ResetPolicy::AUTOMATIC,
+                      WaitableEvent::InitialState::NOT_SIGNALED) {}
 
 StackSamplingProfiler::SamplingThread::~SamplingThread() {}
 
-void StackSamplingProfiler::SamplingThread::ThreadMain() {
-  PlatformThread::SetName("Chrome_SamplingProfilerThread");
+StackSamplingProfiler::SamplingThread*
+StackSamplingProfiler::SamplingThread::GetInstance() {
+  return Singleton<SamplingThread>::get();
+}
 
-  // For now, just ignore any requests to profile while another profiler is
-  // working.
-  if (!concurrent_profiling_lock.Get().Try())
-    return;
+int StackSamplingProfiler::SamplingThread::Start(
+    std::unique_ptr<ActiveCapture> capture) {
+  int id = capture->capture_id;
 
-  CallStackProfiles profiles;
-  CollectProfiles(&profiles);
-  concurrent_profiling_lock.Get().Release();
-  completed_callback_.Run(std::move(profiles));
+  {
+    AutoLock lock(capture_change_lock_);
+    start_captures_.push_back(std::move(capture));
+  }
+
+  if (!HasBeenStarted())
+    SimpleThread::Start();
+
+  return id;
 }
 
-// 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();
+void StackSamplingProfiler::SamplingThread::Stop(int id) {
+  AutoLock lock(capture_change_lock_);
+  stop_captures_.push_back(id);
+}
+
+void StackSamplingProfiler::SamplingThread::StartCapture(
+    ActiveCapture* capture) {
+  DCHECK(capture->native_sampler);
+}
+
+void StackSamplingProfiler::SamplingThread::FinishCapture(
+    ActiveCapture* capture) {
+  // 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;
   }
 
-  *elapsed_time = profile_timer.Elapsed();
-  profile->profile_duration = *elapsed_time;
-  native_sampler_->ProfileRecordingStopped();
+  // 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));
 }
 
-// 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))
-    return;
+void StackSamplingProfiler::SamplingThread::PerformCapture(
+    ActiveCapture* capture) {
+  // 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);

[Mike Wittman, 2016/12/06 21:04:57]

The matching call to ProfileRecordingStopped has been dropped with these
changes.

[bcwhite, 2016/12/15 18:07:50]

Done.

+  }
 
-  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;
-    }
+  // The currently active profile being acptured.
+  CallStackProfile& profile = capture->profiles.back();
 
-    CallStackProfile profile;
-    bool was_stopped = false;
-    CollectProfile(&profile, &previous_elapsed_profile_time, &was_stopped);
-    if (!profile.samples.empty())
-      profiles->push_back(std::move(profile));
+  // Capture a single sample.
+  profile.samples.push_back(Sample());
+  capture->native_sampler->RecordStackSample(&profile.samples.back());
 
-    if (was_stopped)
-      return;
+  // 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;
   }
 }
 
-void StackSamplingProfiler::SamplingThread::Stop() {
-  stop_event_.Signal();
+void StackSamplingProfiler::SamplingThread::Run() {
+  while (true) {
+    TimeDelta wait;
+    ActiveCapture* capture = nullptr;
+    if (!active_captures_.empty()) {
+      capture = active_captures_.front().get();
+      if (!capture) {
+        // If the top is null, it must have been explicitly removed. Try next.
+        std::pop_heap(active_captures_.begin(), active_captures_.end(),
+                      &ActiveCaptureOrder);
+        active_captures_.pop_back();
+        continue;
+      }
+      wait = active_captures_.front()->next_sample_time - Time::Now();
+    } else {
+      wait = TimeDelta::FromDays(365);  // A long, long time.

[Mike Wittman, 2016/12/06 21:04:58]

There's a general desire to have as few persistent threads as possible in
Chrome, so we probably should have the sampling thread terminate after a period
of inactivity, and restart on demand.

[bcwhite, 2016/12/15 18:07:50]

To be done in a future CL.

+    }
+
+    if (wait > TimeDelta() && capture_change_.TimedWait(wait)) {
+      // Something has changed.
+      AutoLock lock(capture_change_lock_);
+
+      // Add any new captures to the priority queue and start it.
+      while (!start_captures_.empty()) {
+        std::unique_ptr<ActiveCapture> capture_ptr =
+            std::move(start_captures_.back());
+        start_captures_.pop_back();
+        capture->next_sample_time =
+            Time::Now() + capture_ptr->params.initial_delay;
+        StartCapture(capture_ptr.get());
+        active_captures_.push_back(std::move(capture_ptr));
+        std::push_heap(active_captures_.begin(), active_captures_.end(),
+                       &ActiveCaptureOrder);
+      }
+
+      // Remove any captures that are to be stopped.
+      while (!stop_captures_.empty()) {
+        int stop_id = stop_captures_.back();
+        stop_captures_.pop_back();
+
+        // Loop through all active captures and finish the one with a matching
+        // ID. There are certainly more efficient ways to do this for large
+        // collections but given that there will generally be very few active
+        // captures and very few captures being stopped, it would likely be
+        // slower to do something more complicated.
+        for (auto& capture_ptr : active_captures_) {
+          if (capture_ptr->capture_id == stop_id) {
+            FinishCapture(capture_ptr.get());
+            capture_ptr.reset();  // Nullify the pointer inside the queue.
+            break;
+          }
+        }
+      }
+
+      // As the thread woke due to a "capture-change" event, start over waiting
+      // for the next capture time.
+      continue;
+    }
+
+    // If there were no active captures, check again.
+    if (!capture)
+      continue;
+
+    // Pop the capture from the head of the queue.
+    std::pop_heap(active_captures_.begin(), active_captures_.end(),
+                  &ActiveCaptureOrder);
+    std::unique_ptr<ActiveCapture> capture_ptr =
+        std::move(active_captures_.back());
+    active_captures_.pop_back();
+    DCHECK_EQ(capture, capture_ptr.get());
+
+    // Do the collection of a single sample.
+    PerformCapture(capture);
+
+    // Update the time of the next capture.
+    if (capture->UpdateNextSampleTime()) {
+      // Place the updated entry back on the queue.
+      active_captures_.push_back(std::move(capture_ptr));

[Mike Wittman, 2016/12/06 21:04:57]

push_heap?

[bcwhite, 2016/12/15 18:07:50]

Acknowledged.

+    } else {
+      // All capturing has completed so finish the collection.
+      FinishCapture(capture);
+    }
+  }
 }
 
 // StackSamplingProfiler ------------------------------------------------------
@@ -287,8 +475,6 @@ StackSamplingProfiler::StackSamplingProfiler(
 
 StackSamplingProfiler::~StackSamplingProfiler() {
   Stop();
-  if (!sampling_thread_handle_.is_null())
-    PlatformThread::Join(sampling_thread_handle_);
 }
 
 // static
@@ -310,16 +496,13 @@ void StackSamplingProfiler::Start() {
   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()->Start(
+      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