[tracing] Implement RunsTasksOnCurrentThread for the worker task runner

This is needed for Perfetto which sometimes chooses to bypass the task
queue if the task queue uses a given thread to run tasks.

Bug: v8:8339
Change-Id: Iecec5e7883d174e4b63495ecdadfb96105e4505c
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/1588471
Commit-Queue: Peter Marshall <petermarshall@chromium.org>
Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
Auto-Submit: Peter Marshall <petermarshall@chromium.org>
Cr-Commit-Position: refs/heads/master@{#61316}

src/libplatform/default-worker-threads-task-runner.cc

@@ -11,7 +11,9 @@ namespace platform {
 
 DefaultWorkerThreadsTaskRunner::DefaultWorkerThreadsTaskRunner(
     uint32_t thread_pool_size, TimeFunction time_function)
-    : queue_(time_function), time_function_(time_function) {
+    : queue_(time_function),
+      time_function_(time_function),
+      thread_pool_size_(thread_pool_size) {
   for (uint32_t i = 0; i < thread_pool_size; ++i) {
     thread_pool_.push_back(base::make_unique<WorkerThread>(this));
   }
@@ -23,12 +25,20 @@ double DefaultWorkerThreadsTaskRunner::MonotonicallyIncreasingTime() {
   return time_function_();
 }
 
+bool DefaultWorkerThreadsTaskRunner::RunsTasksOnCurrentThread() const {
+  USE(thread_pool_size_);
+  DCHECK_EQ(thread_pool_size_, 1);
+  return single_worker_thread_id_.load(std::memory_order_relaxed) ==
+         base::OS::GetCurrentThreadId();
+}
+
 void DefaultWorkerThreadsTaskRunner::Terminate() {
   base::MutexGuard guard(&lock_);
   terminated_ = true;
   queue_.Terminate();
   // Clearing the thread pool lets all worker threads join.
   thread_pool_.clear();
+  single_worker_thread_id_.store(0, std::memory_order_relaxed);
 }
 
 void DefaultWorkerThreadsTaskRunner::PostTask(std::unique_ptr<Task> task) {
@@ -69,6 +79,8 @@ DefaultWorkerThreadsTaskRunner::WorkerThread::WorkerThread(
 DefaultWorkerThreadsTaskRunner::WorkerThread::~WorkerThread() { Join(); }
 
 void DefaultWorkerThreadsTaskRunner::WorkerThread::Run() {
+  runner_->single_worker_thread_id_.store(base::OS::GetCurrentThreadId(),
+                                          std::memory_order_relaxed);
   while (std::unique_ptr<Task> task = runner_->GetNext()) {
     task->Run();
   }

src/libplatform/default-worker-threads-task-runner.h

@@ -30,6 +30,10 @@ class V8_PLATFORM_EXPORT DefaultWorkerThreadsTaskRunner
 
   double MonotonicallyIncreasingTime();
 
+  // It is only valid to call this method on a task runner with a single worker
+  // thread. True if the current thread is the worker thread.
+  bool RunsTasksOnCurrentThread() const;
+
   // v8::TaskRunner implementation.
   void PostTask(std::unique_ptr<Task> task) override;
 
@@ -64,6 +68,8 @@ class V8_PLATFORM_EXPORT DefaultWorkerThreadsTaskRunner
   DelayedTaskQueue queue_;
   std::vector<std::unique_ptr<WorkerThread>> thread_pool_;
   TimeFunction time_function_;
+  std::atomic_int single_worker_thread_id_{0};
+  uint32_t thread_pool_size_;
 };
 
 }  // namespace platform

test/unittests/libplatform/default-worker-threads-task-runner-unittest.cc

@@ -274,5 +274,25 @@ TEST(DefaultWorkerThreadsTaskRunnerUnittest, NoIdleTasks) {
   runner.Terminate();
 }
 
+TEST(DefaultWorkerThreadsTaskRunnerUnittest, RunsTasksOnCurrentThread) {
+  DefaultWorkerThreadsTaskRunner runner(1, RealTime);
+
+  base::Semaphore semaphore(0);
+
+  EXPECT_FALSE(runner.RunsTasksOnCurrentThread());
+
+  std::unique_ptr<TestTask> task1 = base::make_unique<TestTask>([&] {
+    EXPECT_TRUE(runner.RunsTasksOnCurrentThread());
+    semaphore.Signal();
+  });
+  runner.PostTask(std::move(task1));
+
+  semaphore.Wait();
+  EXPECT_FALSE(runner.RunsTasksOnCurrentThread());
+
+  runner.Terminate();
+  EXPECT_FALSE(runner.RunsTasksOnCurrentThread());
+}
+
 }  // namespace platform
 }  // namespace v8