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/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"
 
 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) {}
+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");
-
-  // For now, just ignore any requests to profile while another profiler is
-  // working.
-  if (!concurrent_profiling_lock.Get().Try())
-    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))
-    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();
+StackSamplingProfiler::SamplingThread*
+StackSamplingProfiler::SamplingThread::GetInstance() {
+  return Singleton<SamplingThread>::get();
+}
+
+int StackSamplingProfiler::SamplingThread::Start(
+    std::unique_ptr<ActiveCapture> capture) {
+  int id = capture->capture_id;
+
+  {
+    AutoLock lock(capture_change_lock_);
+    start_captures_.push_back(std::move(capture));
+  }
+
+  if (!HasBeenStarted())
+    SimpleThread::Start();
+
+  return id;
+}
+
+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;
+  }
+
+  // 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));
+}
+
+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.

+  }
+
+  // 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;
+  }
+}
+
+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 ------------------------------------------------------
 
 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,

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

We'll need to refactor to use a single stack copy buffer across all
NativeStackSamplers, as the buffer is fairly large.

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

Future CL.

                                  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()->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
 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