Implement Add/Remove FileDescriptorWatch for the PerfettoTaskRunner

This is needed in a future CL to connect via the IPC mechanism used by
the system Perfetto service on Android.

Bug: 966047
Change-Id: I0521d9e91cc6d3332607be747743a2ab590255ed
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1624817
Commit-Queue: Stephen Nusko <nuskos@chromium.org>
Auto-Submit: Stephen Nusko <nuskos@chromium.org>
Reviewed-by: Eric Seckler <eseckler@chromium.org>
Reviewed-by: Sami Kyöstilä <skyostil@chromium.org>
Reviewed-by: François Doray <fdoray@chromium.org>
Cr-Commit-Position: refs/heads/master@{#672063}

services/tracing/perfetto/consumer_host_unittest.cc

@@ -73,6 +73,17 @@ class ThreadedPerfettoService : public mojom::TracingSessionClient {
       wait_for_destruction.Run();
     }
 
+    {
+      base::RunLoop wait_for_destruction;
+      task_runner_->PostTaskAndReply(
+          FROM_HERE,
+          base::BindOnce(
+              [](std::unique_ptr<PerfettoService> service) { service.reset(); },
+              std::move(perfetto_service_)),
+          wait_for_destruction.QuitClosure());
+      wait_for_destruction.Run();
+    }
+
     {
       base::RunLoop wait_for_destruction;
       PerfettoTracedProcess::GetTaskRunner()

services/tracing/perfetto/test_utils.cc

@@ -258,10 +258,19 @@ MockProducer::MockProducer(const std::string& producer_name,
                            base::OnceClosure on_datasource_registered,
                            base::OnceClosure on_tracing_started,
                            size_t num_packets) {
-  data_source_ =
-      std::make_unique<TestDataSource>(data_source_name, num_packets);
+  // Construct MockProducerClient before TestDataSource to avoid a race.
+  //
+  // TestDataSource calls AddDataSource on the global PerfettoTracedProcess,
+  // which PostTasks to the threadpool in the task it will access the
+  // |producer_client_| pointer that the PerfettoTracedProcess owns. However in
+  // the constructor for MockProducerClient we will set the |producer_client_|
+  // from the real client to the mock, however this is done on a different
+  // sequence and thus we have a race. By setting the pointer before we
+  // construct the data source the TestDataSource can not race.
   producer_client_ = std::make_unique<MockProducerClient>(
       /* num_data_sources = */ 1, std::move(on_tracing_started));
+  data_source_ =
+      std::make_unique<TestDataSource>(data_source_name, num_packets);
 
   producer_host_ = std::make_unique<MockProducerHost>(
       producer_name, data_source_name, service, producer_client_.get(),

services/tracing/public/cpp/perfetto/task_runner.cc

@@ -9,6 +9,7 @@
 
 #include "base/bind.h"
 #include "base/no_destructor.h"
+#include "base/stl_util.h"
 #include "base/task/common/checked_lock_impl.h"
 #include "base/task/common/scoped_defer_task_posting.h"
 #include "base/task/post_task.h"
@@ -24,7 +25,12 @@ PerfettoTaskRunner::PerfettoTaskRunner(
     scoped_refptr<base::SequencedTaskRunner> task_runner)
     : task_runner_(std::move(task_runner)) {}
 
-PerfettoTaskRunner::~PerfettoTaskRunner() = default;
+PerfettoTaskRunner::~PerfettoTaskRunner() {
+  DCHECK(GetOrCreateTaskRunner()->RunsTasksInCurrentSequence());
+#if defined(OS_ANDROID)
+  fd_controllers_.clear();
+#endif  // defined(OS_ANDROID)
+}
 
 void PerfettoTaskRunner::PostTask(std::function<void()> task) {
   base::ScopedDeferTaskPosting::PostOrDefer(
@@ -70,12 +76,29 @@ bool PerfettoTaskRunner::RunsTasksOnCurrentThread() const {
   return task_runner_->RunsTasksInCurrentSequence();
 }
 
-void PerfettoTaskRunner::AddFileDescriptorWatch(int fd, std::function<void()>) {
+void PerfettoTaskRunner::AddFileDescriptorWatch(
+    int fd,
+    std::function<void()> callback) {
+#if !defined(OS_ANDROID)
   NOTREACHED();
+#else
+  DCHECK(GetOrCreateTaskRunner()->RunsTasksInCurrentSequence());
+  DCHECK(!base::Contains(fd_controllers_, fd));
+  fd_controllers_[fd] = base::FileDescriptorWatcher::WatchReadable(
+      fd,
+      base::BindRepeating([](std::function<void()> callback) { callback(); },
+                          std::move(callback)));
+#endif  // !defined(OS_ANDROID)
 }
 
 void PerfettoTaskRunner::RemoveFileDescriptorWatch(int fd) {
+#if !defined(OS_ANDROID)
   NOTREACHED();
+#else
+  DCHECK(GetOrCreateTaskRunner()->RunsTasksInCurrentSequence());
+  DCHECK(base::Contains(fd_controllers_, fd));
+  fd_controllers_.erase(fd);
+#endif  // !defined(OS_ANDROID)
 }
 
 void PerfettoTaskRunner::ResetTaskRunnerForTesting(

services/tracing/public/cpp/perfetto/task_runner.h

@@ -12,9 +12,17 @@
 #include "base/sequenced_task_runner.h"
 #include "base/synchronization/lock.h"
 #include "base/timer/timer.h"
+#include "build/build_config.h"
 #include "services/tracing/public/mojom/perfetto_service.mojom.h"
 #include "third_party/perfetto/include/perfetto/base/task_runner.h"
 
+#if defined(OS_ANDROID)
+#include <map>
+// Needed for base::FileDescriptorWatcher::Controller and for implementing
+// AddFileDescriptorWatch & RemoveFileDescriptorWatch on Android.
+#include "base/files/file_descriptor_watcher_posix.h"
+#endif  // defined(OS_ANDROID)
+
 namespace tracing {
 
 // This wraps a base::TaskRunner implementation to be able
@@ -40,7 +48,7 @@ class COMPONENT_EXPORT(TRACING_CPP) PerfettoTaskRunner
   scoped_refptr<base::SequencedTaskRunner> GetOrCreateTaskRunner();
   bool HasTaskRunner() const { return !!task_runner_; }
 
-  // Not used in Chrome.
+  // These are only used on Android when talking to the system Perfetto service.
   void AddFileDescriptorWatch(int fd, std::function<void()>) override;
   void RemoveFileDescriptorWatch(int fd) override;
 
@@ -54,6 +62,10 @@ class COMPONENT_EXPORT(TRACING_CPP) PerfettoTaskRunner
   void OnDeferredTasksDrainTimer();
 
   scoped_refptr<base::SequencedTaskRunner> task_runner_;
+#if defined(OS_ANDROID)
+  std::map<int, std::unique_ptr<base::FileDescriptorWatcher::Controller>>
+      fd_controllers_;
+#endif  // defined(OS_ANDROID)
 
   DISALLOW_COPY_AND_ASSIGN(PerfettoTaskRunner);
 };

services/tracing/public/cpp/perfetto/task_runner_unittest.cc

@@ -91,7 +91,8 @@ class PosterThread : public base::SimpleThread {
 class PerfettoTaskRunnerTest : public testing::Test {
  public:
   void SetUp() override {
-    task_runner_ = std::make_unique<PerfettoTaskRunner>(CreateNewTaskrunner());
+    sequenced_task_runner_ = CreateNewTaskrunner();
+    task_runner_ = std::make_unique<PerfettoTaskRunner>(sequenced_task_runner_);
   }
 
   scoped_refptr<base::SequencedTaskRunner> CreateNewTaskrunner() {
@@ -105,15 +106,18 @@ class PerfettoTaskRunnerTest : public testing::Test {
         number_of_sequences, expected_tasks, std::move(on_complete));
   }
 
-  void TearDown() override {}
+  void TearDown() override {
+    sequenced_task_runner_->DeleteSoon(FROM_HERE, std::move(task_runner_));
+  }
 
   PerfettoTaskRunner* task_runner() { return task_runner_.get(); }
   TaskDestination* destination() { return task_destination_.get(); }
 
  private:
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
+  scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
   std::unique_ptr<PerfettoTaskRunner> task_runner_;
   std::unique_ptr<TaskDestination> task_destination_;
-  base::test::ScopedTaskEnvironment scoped_task_environment_;
 };
 
 TEST_F(PerfettoTaskRunnerTest, SequentialTasks) {