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 <map>
+#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/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 CollectionContext {
+    CollectionContext(PlatformThreadId target,
+                      const SamplingParams& params,
+                      const CompletedCallback& callback,
+                      std::unique_ptr<NativeStackSampler> sampler)
+        : collection_id(next_collection_id_.GetNext()),
+          target(target),
+          params(params),
+          callback(callback),
+          native_sampler(std::move(sampler)) {}
+    ~CollectionContext() {}
+
+    // An identifier for this collection, used to uniquely identify it to
+    // outside interests.
+    const int collection_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 collected stack samples. The active profile is always at the back().
+    CallStackProfiles profiles;
+
+   private:
+    static StaticAtomicSequenceNumber next_collection_id_;
+  };
+
+  // Gets the single instance of this class.
+  static SamplingThread* GetInstance();
+
+  // Starts the thread.
+  void Start();
+
+  // Adds a new CollectionContext 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<CollectionContext> collection);
+
+  // Stops an active collection 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>;
+
+  // Get task runner that is usable from the outside.
+  scoped_refptr<SingleThreadTaskRunner> GetOrCreateTaskRunner();
+  scoped_refptr<SingleThreadTaskRunner> GetTaskRunner();
+
+  // Get tas krunner that is usable from the sampling thread itself.

[Mike Wittman, 2017/01/06 16:02:40]

nit: task runner

[bcwhite, 2017/01/06 16:43:52]

Done.

+  scoped_refptr<SingleThreadTaskRunner> GetTaskRunnerFromSamplingThread();
+
+  // Finishes a collection and reports collected data via callback.
+  void FinishCollection(CollectionContext* collection);
+
+  // Records a single sample of a collection.
+  void RecordSample(CollectionContext* collection);
+
+  // These methods are tasks that get posted to the internal message queue.
+  void StartCollectionTask(std::unique_ptr<CollectionContext> collection_ptr);
+  void StopCollectionTask(int id);
+  void PerformCollectionTask(int id);
+
+  // Updates the |next_sample_time| time based on configured parameters.
+  bool UpdateNextSampleTime(CollectionContext* collection);
+
+  // Thread:
+  void CleanUp() override;
+
+  // 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 collection contexts. Because this class is a singleton
+  // that is never destroyed, context objects will never be destructed except
+  // by explicit action. Thus, it's acceptable to pass unretained pointers
+  // to these objects when posting tasks.
+  std::map<int, std::unique_ptr<CollectionContext>> active_collections_;
+
+  DISALLOW_COPY_AND_ASSIGN(SamplingThread);
+};
+
+StaticAtomicSequenceNumber StackSamplingProfiler::SamplingThread::
+    CollectionContext::next_collection_id_;
+
+StackSamplingProfiler::SamplingThread::SamplingThread()
+    : Thread("Chrome_SamplingProfilerThread") {}
+
+StackSamplingProfiler::SamplingThread::~SamplingThread() {
+  Thread::Stop();
+}
+
+StackSamplingProfiler::SamplingThread*
+StackSamplingProfiler::SamplingThread::GetInstance() {
+  return Singleton<SamplingThread, LeakySingletonTraits<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<CollectionContext> collection) {
+  int id = collection->collection_id;
+  GetOrCreateTaskRunner()->PostTask(
+      FROM_HERE, Bind(&SamplingThread::StartCollectionTask, Unretained(this),
+                      Passed(&collection)));
+  return id;
+}
+
+void StackSamplingProfiler::SamplingThread::Stop(int id) {
+  scoped_refptr<SingleThreadTaskRunner> task_runner = GetTaskRunner();
+  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::StopCollectionTask,
+                                        Unretained(this), id));
+}
+
+scoped_refptr<SingleThreadTaskRunner>
+StackSamplingProfiler::SamplingThread::GetOrCreateTaskRunner() {
+  AutoLock lock(task_runner_lock_);
+  if (!task_runner_) {
+    // The thread is not running. Start it and get associated runner. The task-
+    // runner has to be saved for future use because though it can be used from
+    // any thread, it can be acquired via task_runner() only on the created
+    // thread and the thread that creates it (i.e. this thread).
+    Start();
+    task_runner_ = task_runner();

[Mike Wittman, 2017/01/06 16:02:40]

nit: Thread::task_runner() to be explicitly clear where this function is coming
from

[bcwhite, 2017/01/06 16:43:52]

Done.

+  } else {
+    // This shouldn't be called from the sampling thread as it's inefficient.
+    // Use GetTaskRunnerFromSamplingThread() instead.
+    DCHECK_NE(GetThreadId(), PlatformThread::CurrentId());
+  }
+
+  return task_runner_;
+}
+
+scoped_refptr<SingleThreadTaskRunner>
+StackSamplingProfiler::SamplingThread::GetTaskRunner() {
+  // This shouldn't be called from the sampling thread as it's inefficient. Use
+  // GetTaskRunnerFromSamplingThread() instead.
+  DCHECK_NE(GetThreadId(), PlatformThread::CurrentId());
+
+  AutoLock lock(task_runner_lock_);
+  return task_runner_;
+}
+
+scoped_refptr<SingleThreadTaskRunner>
+StackSamplingProfiler::SamplingThread::GetTaskRunnerFromSamplingThread() {

[Mike Wittman, 2017/01/06 16:02:40]

How about GetTaskRunnerOnSamplingThread? GetTaskRunnerFromSamplingThread is
ambiguous since "SamplingThread" could refer to either the thread of execution
or the SamplingThread class itself.

[bcwhite, 2017/01/06 16:43:52]

Done.

+  // This should be called only from the sampling thread as it has limited
+  // accessibility.
+  DCHECK_EQ(GetThreadId(), PlatformThread::CurrentId());
+
+  return task_runner();

[Mike Wittman, 2017/01/06 16:02:40]

nit: Thread::task_runner() here also

[bcwhite, 2017/01/06 16:43:52]

Done.

+}
+
+void StackSamplingProfiler::SamplingThread::FinishCollection(
+    CollectionContext* collection) {
+  // If there is no duration for the final profile (because it was stopped),
+  // calculated it now.
+  if (!collection->profiles.empty() &&
+      collection->profiles.back().profile_duration == TimeDelta()) {
+    collection->profiles.back().profile_duration =
+        Time::Now() - collection->profile_start_time;
+  }
+
+  // Run the associated callback, passing the collected profiles. It's okay to
+  // move them because this collection is about to be deleted.
+  collection->callback.Run(std::move(collection->profiles));
+
+  // Remove this collection from the map of known ones.  This must be done
+  // last as the |collection| parameter is invalid after this point.
+  size_t count = active_collections_.erase(collection->collection_id);
+  DCHECK_EQ(1U, count);
+}
+
+void StackSamplingProfiler::SamplingThread::RecordSample(
+    CollectionContext* collection) {
+  DCHECK(collection->native_sampler);
+
+  // If this is the first sample of a burst, a new Profile needs to be created
+  // and filled.
+  if (collection->sample == 0) {
+    collection->profiles.push_back(CallStackProfile());
+    CallStackProfile& profile = collection->profiles.back();
+    profile.sampling_period = collection->params.sampling_interval;
+    collection->profile_start_time = Time::Now();
+    collection->native_sampler->ProfileRecordingStarting(&profile.modules);
+  }
+
+  // The currently active profile being acptured.
+  CallStackProfile& profile = collection->profiles.back();
+
+  // Record a single sample.
+  profile.samples.push_back(Sample());
+  collection->native_sampler->RecordStackSample(&profile.samples.back());
+
+  // If this is the last sample of a burst, record the total time.
+  if (collection->sample == collection->params.samples_per_burst - 1) {
+    profile.profile_duration = Time::Now() - collection->profile_start_time;
+    collection->native_sampler->ProfileRecordingStopped();
+  }
+}
+
+void StackSamplingProfiler::SamplingThread::StartCollectionTask(
+    std::unique_ptr<CollectionContext> collection_ptr) {
+  // Ownership of the collection is going to be given to a map but a pointer
+  // to it will be needed later.
+  CollectionContext* collection = collection_ptr.get();
+  active_collections_.insert(
+      std::make_pair(collection->collection_id, std::move(collection_ptr)));
+
+  GetTaskRunnerFromSamplingThread()->PostDelayedTask(
+      FROM_HERE, Bind(&SamplingThread::PerformCollectionTask, Unretained(this),
+                      collection->collection_id),
+      collection->params.initial_delay);
+}
+
+void StackSamplingProfiler::SamplingThread::StopCollectionTask(int id) {
+  auto found = active_collections_.find(id);
+  if (found == active_collections_.end())
     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))
+  FinishCollection(found->second.get());
+}
+
+void StackSamplingProfiler::SamplingThread::PerformCollectionTask(int id) {
+  auto found = active_collections_.find(id);
+
+  // The task won't be found if it has been stopped.
+  if (found == active_collections_.end())
     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();
+  CollectionContext* collection = found->second.get();
+
+  // Handle first-run with no "next time".
+  if (collection->next_sample_time == Time())
+    collection->next_sample_time = Time::Now();
+
+  // Do the collection of a single sample.
+  RecordSample(collection);
+
+  // Update the time of the next sample recording.
+  if (UpdateNextSampleTime(collection)) {
+    bool success = GetTaskRunnerFromSamplingThread()->PostDelayedTask(
+        FROM_HERE,
+        Bind(&SamplingThread::PerformCollectionTask, Unretained(this), id),
+        std::max(collection->next_sample_time - Time::Now(), TimeDelta()));
+    DCHECK(success);
+  } else {
+    // All capturing has completed so finish the collection. Let object expire.
+    // The |collection| variable will be invalid after this call.
+    FinishCollection(collection);
+  }
+}
+
+bool StackSamplingProfiler::SamplingThread::UpdateNextSampleTime(
+    CollectionContext* collection) {
+  if (++collection->sample < collection->params.samples_per_burst) {
+    collection->next_sample_time += collection->params.sampling_interval;
+    return true;
+  }
+
+  // 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.
+  if (++collection->burst < collection->params.bursts) {
+    collection->sample = 0;
+    collection->next_sample_time += collection->params.burst_interval;
+    return true;
+  }
+
+  return false;
+}
+
+void StackSamplingProfiler::SamplingThread::CleanUp() {
+  // There should be no collections remaining when the thread stops.
+  DCHECK(active_collections_.empty());
+
+  // 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();
+  collection_id_ = SamplingThread::GetInstance()->Add(
+      MakeUnique<SamplingThread::CollectionContext>(
+          thread_id_, params_, completed_callback_, std::move(native_sampler)));
 }
 
 void StackSamplingProfiler::Stop() {
-  if (sampling_thread_)
-    sampling_thread_->Stop();
+  SamplingThread::GetInstance()->Stop(collection_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