TaskScheduler: Add SchedulerWorkerPool backed by libdispatch

PlatformNativeWorkerPoolMac is an alternative worker pool backed by
libdispatch. We expect libdispatch to perform better than the current
implementation, as it is optimized for thread pool parallelism.

In order to assess the benefits of using libdispatch, an experiment to
to use PlatformNativeWorkerPoolMac in place of SchedulerWorkerPoolImpl
will be conducted on macOS and iOS.

Bug: 946147
Change-Id: Iebc4386070bad629e01ab05438546be545a4f3b4
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1536442Reviewed-by: Robert Sesek <rsesek@chromium.org>
Reviewed-by: François Doray <fdoray@chromium.org>
Commit-Queue: Aditya Keerthi <adityakeerthi@google.com>
Cr-Commit-Position: refs/heads/master@{#644806}

base/BUILD.gn

@@ -795,6 +795,10 @@ jumbo_component("base") {
     "task/task_scheduler/environment_config.h",
     "task/task_scheduler/initialization_util.cc",
     "task/task_scheduler/initialization_util.h",
+    "task/task_scheduler/platform_native_worker_pool.cc",
+    "task/task_scheduler/platform_native_worker_pool.h",
+    "task/task_scheduler/platform_native_worker_pool_mac.h",
+    "task/task_scheduler/platform_native_worker_pool_mac.mm",
     "task/task_scheduler/platform_native_worker_pool_win.cc",
     "task/task_scheduler/platform_native_worker_pool_win.h",
     "task/task_scheduler/priority_queue.cc",
@@ -1853,6 +1857,8 @@ jumbo_component("base") {
       "strings/sys_string_conversions_mac.mm",
       "synchronization/waitable_event_mac.cc",
       "system/sys_info_ios.mm",
+      "task/task_scheduler/platform_native_worker_pool_mac.h",
+      "task/task_scheduler/platform_native_worker_pool_mac.mm",
       "threading/platform_thread_mac.mm",
       "time/time_conversion_posix.cc",
       "time/time_mac.cc",

base/task/task_scheduler/platform_native_worker_pool.cc

+// Copyright 2019 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/task/task_scheduler/platform_native_worker_pool.h"
+
+#include <algorithm>
+#include <utility>
+
+#include "base/system/sys_info.h"
+#include "base/task/task_scheduler/task_tracker.h"
+
+namespace base {
+namespace internal {
+
+class PlatformNativeWorkerPool::ScopedWorkersExecutor
+    : public SchedulerWorkerPool::BaseScopedWorkersExecutor {
+ public:
+  ScopedWorkersExecutor(PlatformNativeWorkerPool* outer) : outer_(outer) {}
+  ~ScopedWorkersExecutor() {
+    SchedulerLock::AssertNoLockHeldOnCurrentThread();
+
+    for (size_t i = 0; i < num_threadpool_work_to_submit_; ++i)
+      outer_->SubmitWork();
+  }
+
+  // Sets the number of threadpool work to submit upon destruction.
+  void set_num_threadpool_work_to_submit(size_t num) {
+    DCHECK_EQ(num_threadpool_work_to_submit_, 0U);
+    num_threadpool_work_to_submit_ = num;
+  }
+
+ private:
+  PlatformNativeWorkerPool* const outer_;
+  size_t num_threadpool_work_to_submit_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(ScopedWorkersExecutor);
+};
+
+PlatformNativeWorkerPool::PlatformNativeWorkerPool(
+    TrackedRef<TaskTracker> task_tracker,
+    TrackedRef<Delegate> delegate)
+    : SchedulerWorkerPool(std::move(task_tracker), std::move(delegate)) {}
+
+PlatformNativeWorkerPool::~PlatformNativeWorkerPool() {
+#if DCHECK_IS_ON()
+  // Verify join_for_testing has been called to ensure that there is no more
+  // outstanding work. Otherwise, work may try to de-reference an invalid
+  // pointer to this class.
+  DCHECK(join_for_testing_returned_);
+#endif
+}
+
+void PlatformNativeWorkerPool::Start() {
+  StartImpl();
+
+  ScopedWorkersExecutor executor(this);
+  AutoSchedulerLock auto_lock(lock_);
+  DCHECK(!started_);
+  started_ = true;
+  EnsureEnoughWorkersLockRequired(&executor);
+}
+
+void PlatformNativeWorkerPool::JoinForTesting() {
+  JoinImpl();
+#if DCHECK_IS_ON()
+  DCHECK(!join_for_testing_returned_);
+  join_for_testing_returned_ = true;
+#endif
+}
+
+void PlatformNativeWorkerPool::RunNextSequenceImpl() {
+  BindToCurrentThread();
+
+  scoped_refptr<Sequence> sequence = GetWork();
+  DCHECK(sequence);
+
+  sequence = task_tracker_->RunAndPopNextTask(std::move(sequence), this);
+
+  if (sequence) {
+    ScopedWorkersExecutor workers_executor(this);
+    ScopedReenqueueExecutor reenqueue_executor;
+    auto sequence_and_transaction =
+        SequenceAndTransaction::FromSequence(std::move(sequence));
+    AutoSchedulerLock auto_lock(lock_);
+    ReEnqueueSequenceLockRequired(&workers_executor, &reenqueue_executor,
+                                  std::move(sequence_and_transaction));
+  }
+
+  UnbindFromCurrentThread();
+}
+
+scoped_refptr<Sequence> PlatformNativeWorkerPool::GetWork() {
+  AutoSchedulerLock auto_lock(lock_);
+  DCHECK_GT(num_pending_threadpool_work_, 0U);
+  --num_pending_threadpool_work_;
+  // There can be more pending threadpool work than Sequences in the
+  // PriorityQueue after RemoveSequence().
+  if (priority_queue_.IsEmpty())
+    return nullptr;
+  return priority_queue_.PopSequence();
+}
+
+void PlatformNativeWorkerPool::UpdateSortKey(
+    SequenceAndTransaction sequence_and_transaction) {
+  ScopedWorkersExecutor executor(this);
+  UpdateSortKeyImpl(&executor, std::move(sequence_and_transaction));
+}
+
+void PlatformNativeWorkerPool::PushSequenceAndWakeUpWorkers(
+    SequenceAndTransaction sequence_and_transaction) {
+  ScopedWorkersExecutor executor(this);
+  PushSequenceAndWakeUpWorkersImpl(&executor,
+                                   std::move(sequence_and_transaction));
+}
+
+void PlatformNativeWorkerPool::EnsureEnoughWorkersLockRequired(
+    BaseScopedWorkersExecutor* executor) {
+  if (!started_)
+    return;
+  // Ensure that there is at least one pending threadpool work per Sequence in
+  // the PriorityQueue.
+  const size_t desired_num_pending_threadpool_work = priority_queue_.Size();
+  if (desired_num_pending_threadpool_work > num_pending_threadpool_work_) {
+    static_cast<ScopedWorkersExecutor*>(executor)
+        ->set_num_threadpool_work_to_submit(
+            desired_num_pending_threadpool_work - num_pending_threadpool_work_);
+    num_pending_threadpool_work_ = desired_num_pending_threadpool_work;
+  }
+}
+
+size_t PlatformNativeWorkerPool::GetMaxConcurrentNonBlockedTasksDeprecated()
+    const {
+  // Native thread pools give us no control over the number of workers that are
+  // active at one time. Consequently, we cannot report a true value here.
+  // Instead, the values were chosen to match
+  // TaskScheduler::StartWithDefaultParams.
+  const int num_cores = SysInfo::NumberOfProcessors();
+  return std::max(3, num_cores - 1);
+}
+
+void PlatformNativeWorkerPool::ReportHeartbeatMetrics() const {
+  // Native thread pools do not provide the capability to determine the
+  // number of worker threads created.
+}
+
+}  // namespace internal
+}  // namespace base

base/task/task_scheduler/platform_native_worker_pool.h

+// Copyright 2019 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.
+
+#ifndef BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_H_
+#define BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_H_
+
+#include "base/base_export.h"
+#include "base/synchronization/atomic_flag.h"
+#include "base/task/task_scheduler/scheduler_worker_pool.h"
+
+namespace base {
+namespace internal {
+
+class BASE_EXPORT PlatformNativeWorkerPool : public SchedulerWorkerPool {
+ public:
+  // Destroying a PlatformNativeWorkerPool is not allowed in
+  // production; it is always leaked. In tests, it can only be destroyed after
+  // JoinForTesting() has returned.
+  ~PlatformNativeWorkerPool() override;
+
+  // Starts the worker pool and allows tasks to begin running.
+  void Start();
+
+  // SchedulerWorkerPool:
+  void JoinForTesting() override;
+  size_t GetMaxConcurrentNonBlockedTasksDeprecated() const override;
+  void ReportHeartbeatMetrics() const override;
+
+ protected:
+  PlatformNativeWorkerPool(TrackedRef<TaskTracker> task_tracker,
+                           TrackedRef<Delegate> delegate);
+
+  // Runs a task off the next sequence on the |priority_queue_|. Called by
+  // callbacks posted to platform native thread pools.
+  void RunNextSequenceImpl();
+
+  virtual void JoinImpl() = 0;
+  virtual void StartImpl() = 0;
+  virtual void SubmitWork() = 0;
+
+ private:
+  class ScopedWorkersExecutor;
+
+  // SchedulerWorkerPool:
+  void UpdateSortKey(SequenceAndTransaction sequence_and_transaction) override;
+  void PushSequenceAndWakeUpWorkers(
+      SequenceAndTransaction sequence_and_transaction) override;
+  void EnsureEnoughWorkersLockRequired(BaseScopedWorkersExecutor* executor)
+      override EXCLUSIVE_LOCKS_REQUIRED(lock_);
+
+  // Returns the top Sequence off the |priority_queue_|. Returns nullptr
+  // if the |priority_queue_| is empty.
+  scoped_refptr<Sequence> GetWork();
+
+  // Indicates whether the pool has been started yet.
+  bool started_ GUARDED_BY(lock_) = false;
+
+  // Number of threadpool work submitted to the pool which haven't popped a
+  // Sequence from the PriorityQueue yet.
+  size_t num_pending_threadpool_work_ GUARDED_BY(lock_) = 0;
+
+#if DCHECK_IS_ON()
+  // Set once JoinForTesting() has returned.
+  bool join_for_testing_returned_ = false;
+#endif
+
+  DISALLOW_COPY_AND_ASSIGN(PlatformNativeWorkerPool);
+};
+
+}  // namespace internal
+}  // namespace base
+
+#endif  // BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_H_

base/task/task_scheduler/platform_native_worker_pool_mac.h

+// Copyright 2019 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.
+
+#ifndef BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_MAC_H_
+#define BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_MAC_H_
+
+#include <dispatch/dispatch.h>
+
+#include "base/base_export.h"
+#include "base/mac/scoped_dispatch_object.h"
+#include "base/task/task_scheduler/platform_native_worker_pool.h"
+
+namespace base {
+namespace internal {
+
+// A SchedulerWorkerPool implementation backed by libdispatch.
+//
+// libdispatch official documentation:
+// https://developer.apple.com/documentation/dispatch
+//
+// Guides:
+// https://apple.github.io/swift-corelibs-libdispatch/tutorial/
+// https://developer.apple.com/library/archive/documentation/General/Conceptual/ConcurrencyProgrammingGuide/OperationQueues/OperationQueues.html
+class BASE_EXPORT PlatformNativeWorkerPoolMac
+    : public PlatformNativeWorkerPool {
+ public:
+  PlatformNativeWorkerPoolMac(TrackedRef<TaskTracker> task_tracker,
+                              TrackedRef<Delegate> delegate);
+
+  ~PlatformNativeWorkerPoolMac() override;
+
+ private:
+  // PlatformNativeWorkerPool:
+  void JoinImpl() override;
+  void StartImpl() override;
+  void SubmitWork() override;
+
+  // Dispatch queue on which work is scheduled. Backed by a shared thread pool
+  // managed by libdispatch.
+  ScopedDispatchObject<dispatch_queue_t> queue_;
+
+  // Dispatch group to enable synchronization.
+  ScopedDispatchObject<dispatch_group_t> group_;
+
+  DISALLOW_COPY_AND_ASSIGN(PlatformNativeWorkerPoolMac);
+};
+
+}  // namespace internal
+}  // namespace base
+
+#endif  // BASE_TASK_TASK_SCHEDULER_PLATFORM_NATIVE_WORKER_POOL_MAC_H_

base/task/task_scheduler/platform_native_worker_pool_mac.mm

+// Copyright 2019 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/task/task_scheduler/platform_native_worker_pool_mac.h"
+
+#include "base/task/task_scheduler/task_tracker.h"
+
+namespace base {
+namespace internal {
+
+PlatformNativeWorkerPoolMac::PlatformNativeWorkerPoolMac(
+    TrackedRef<TaskTracker> task_tracker,
+    TrackedRef<Delegate> delegate)
+    : PlatformNativeWorkerPool(std::move(task_tracker), std::move(delegate)) {}
+
+PlatformNativeWorkerPoolMac::~PlatformNativeWorkerPoolMac() {}
+
+void PlatformNativeWorkerPoolMac::StartImpl() {
+  queue_.reset(dispatch_queue_create(
+      "org.chromium.base.TaskScheduler.WorkerPool", DISPATCH_QUEUE_CONCURRENT));
+  group_.reset(dispatch_group_create());
+}
+
+void PlatformNativeWorkerPoolMac::JoinImpl() {
+  dispatch_group_wait(group_, DISPATCH_TIME_FOREVER);
+}
+
+void PlatformNativeWorkerPoolMac::SubmitWork() {
+  // TODO(adityakeerthi): Handle priorities by having multiple dispatch queues
+  // with different qualities-of-service.
+  dispatch_group_async(group_, queue_, ^{
+    RunNextSequenceImpl();
+  });
+}
+
+}  // namespace internal
+}  // namespace base

base/task/task_scheduler/platform_native_worker_pool_win.cc

@@ -4,60 +4,23 @@
 
 #include "base/task/task_scheduler/platform_native_worker_pool_win.h"
 
-#include <algorithm>
-#include <utility>
-
-#include "base/system/sys_info.h"
 #include "base/task/task_scheduler/task_tracker.h"
 
 namespace base {
 namespace internal {
 
-class PlatformNativeWorkerPoolWin::ScopedWorkersExecutor
-    : public SchedulerWorkerPool::BaseScopedWorkersExecutor {
- public:
-  ScopedWorkersExecutor(PlatformNativeWorkerPoolWin* outer) : outer_(outer) {}
-  ~ScopedWorkersExecutor() {
-    SchedulerLock::AssertNoLockHeldOnCurrentThread();
-
-    // TODO(fdoray): Handle priorities by having different work objects and
-    // using SetThreadpoolCallbackPriority() and
-    // SetThreadpoolCallbackRunsLong().
-    for (size_t i = 0; i < num_threadpool_work_to_submit_; ++i)
-      ::SubmitThreadpoolWork(outer_->work_);
-  }
-
-  // Sets the number of threadpool work to submit upon destruction.
-  void set_num_threadpool_work_to_submit(size_t num) {
-    DCHECK_EQ(num_threadpool_work_to_submit_, 0U);
-    num_threadpool_work_to_submit_ = num;
-  }
-
- private:
-  PlatformNativeWorkerPoolWin* const outer_;
-  size_t num_threadpool_work_to_submit_ = 0;
-
-  DISALLOW_COPY_AND_ASSIGN(ScopedWorkersExecutor);
-};
-
 PlatformNativeWorkerPoolWin::PlatformNativeWorkerPoolWin(
     TrackedRef<TaskTracker> task_tracker,
     TrackedRef<Delegate> delegate)
-    : SchedulerWorkerPool(std::move(task_tracker), std::move(delegate)) {}
+    : PlatformNativeWorkerPool(std::move(task_tracker), std::move(delegate)) {}
 
 PlatformNativeWorkerPoolWin::~PlatformNativeWorkerPoolWin() {
-#if DCHECK_IS_ON()
-  // Verify join_for_testing has been called to ensure that there is no more
-  // outstanding work. Otherwise, work may try to de-reference an invalid
-  // pointer to this class.
-  DCHECK(join_for_testing_returned_.IsSet());
-#endif
   ::DestroyThreadpoolEnvironment(&environment_);
   ::CloseThreadpoolWork(work_);
   ::CloseThreadpool(pool_);
 }
 
-void PlatformNativeWorkerPoolWin::Start() {
+void PlatformNativeWorkerPoolWin::StartImpl() {
   ::InitializeThreadpoolEnvironment(&environment_);
 
   pool_ = ::CreateThreadpool(nullptr);
@@ -68,20 +31,16 @@ void PlatformNativeWorkerPoolWin::Start() {
   work_ = ::CreateThreadpoolWork(&RunNextSequence, this, &environment_);
   DCHECK(work_) << "LastError: " << GetLastError();
   ::SetThreadpoolCallbackPool(&environment_, pool_);
-
-  ScopedWorkersExecutor executor(this);
-  AutoSchedulerLock auto_lock(lock_);
-  DCHECK(!started_);
-  started_ = true;
-  EnsureEnoughWorkersLockRequired(&executor);
 }
 
-void PlatformNativeWorkerPoolWin::JoinForTesting() {
+void PlatformNativeWorkerPoolWin::JoinImpl() {
   ::WaitForThreadpoolWorkCallbacks(work_, true);
-#if DCHECK_IS_ON()
-  DCHECK(!join_for_testing_returned_.IsSet());
-  join_for_testing_returned_.Set();
-#endif
+}
+
+void PlatformNativeWorkerPoolWin::SubmitWork() {
+  // TODO(fdoray): Handle priorities by having different work objects and using
+  // SetThreadpoolCallbackPriority() and SetThreadpoolCallbackRunsLong().
+  ::SubmitThreadpoolWork(work_);
 }
 
 // static
@@ -91,81 +50,7 @@ void CALLBACK PlatformNativeWorkerPoolWin::RunNextSequence(
     PTP_WORK) {
   auto* worker_pool = static_cast<PlatformNativeWorkerPoolWin*>(
       scheduler_worker_pool_windows_impl);
-
-  worker_pool->BindToCurrentThread();
-
-  scoped_refptr<Sequence> sequence = worker_pool->GetWork();
-  DCHECK(sequence);
-
-  sequence = worker_pool->task_tracker_->RunAndPopNextTask(std::move(sequence),
-                                                           worker_pool);
-
-  if (sequence) {
-    ScopedWorkersExecutor workers_executor(worker_pool);
-    ScopedReenqueueExecutor reenqueue_executor;
-    auto sequence_and_transaction =
-        SequenceAndTransaction::FromSequence(std::move(sequence));
-    AutoSchedulerLock auto_lock(worker_pool->lock_);
-    worker_pool->ReEnqueueSequenceLockRequired(
-        &workers_executor, &reenqueue_executor,
-        std::move(sequence_and_transaction));
-  }
-
-  worker_pool->UnbindFromCurrentThread();
-}
-
-scoped_refptr<Sequence> PlatformNativeWorkerPoolWin::GetWork() {
-  AutoSchedulerLock auto_lock(lock_);
-  DCHECK_GT(num_pending_threadpool_work_, 0U);
-  --num_pending_threadpool_work_;
-  // There can be more pending threadpool work than Sequences in the
-  // PriorityQueue after RemoveSequence().
-  if (priority_queue_.IsEmpty())
-    return nullptr;
-  return priority_queue_.PopSequence();
-}
-
-void PlatformNativeWorkerPoolWin::UpdateSortKey(
-    SequenceAndTransaction sequence_and_transaction) {
-  ScopedWorkersExecutor executor(this);
-  UpdateSortKeyImpl(&executor, std::move(sequence_and_transaction));
-}
-
-void PlatformNativeWorkerPoolWin::PushSequenceAndWakeUpWorkers(
-    SequenceAndTransaction sequence_and_transaction) {
-  ScopedWorkersExecutor executor(this);
-  PushSequenceAndWakeUpWorkersImpl(&executor,
-                                   std::move(sequence_and_transaction));
-}
-
-void PlatformNativeWorkerPoolWin::EnsureEnoughWorkersLockRequired(
-    BaseScopedWorkersExecutor* executor) {
-  if (!started_)
-    return;
-  // Ensure that there is at least one pending threadpool work per Sequence in
-  // the PriorityQueue.
-  const size_t desired_num_pending_threadpool_work = priority_queue_.Size();
-  if (desired_num_pending_threadpool_work > num_pending_threadpool_work_) {
-    static_cast<ScopedWorkersExecutor*>(executor)
-        ->set_num_threadpool_work_to_submit(
-            desired_num_pending_threadpool_work - num_pending_threadpool_work_);
-    num_pending_threadpool_work_ = desired_num_pending_threadpool_work;
-  }
-}
-
-size_t PlatformNativeWorkerPoolWin::GetMaxConcurrentNonBlockedTasksDeprecated()
-    const {
-  // The Windows Thread Pool API gives us no control over the number of workers
-  // that are active at one time. Consequently, we cannot report a true value
-  // here. Instead, the values were chosen to match
-  // TaskScheduler::StartWithDefaultParams.
-  const int num_cores = SysInfo::NumberOfProcessors();
-  return std::max(3, num_cores - 1);
-}
-
-void PlatformNativeWorkerPoolWin::ReportHeartbeatMetrics() const {
-  // Windows Thread Pool API does not provide the capability to determine the
-  // number of worker threads created.
+  worker_pool->RunNextSequenceImpl();
 }
 
 }  // namespace internal

base/task/task_scheduler/platform_native_worker_pool_win.h

@@ -8,9 +8,7 @@
 #include <windows.h>
 
 #include "base/base_export.h"
-#include "base/logging.h"
-#include "base/synchronization/atomic_flag.h"
-#include "base/task/task_scheduler/scheduler_worker_pool.h"
+#include "base/task/task_scheduler/platform_native_worker_pool.h"
 
 namespace base {
 namespace internal {
@@ -26,42 +24,24 @@ namespace internal {
 // https://msdn.microsoft.com/magazine/hh456398.aspx
 // https://msdn.microsoft.com/magazine/hh547107.aspx
 // https://msdn.microsoft.com/magazine/hh580731.aspx
-class BASE_EXPORT PlatformNativeWorkerPoolWin : public SchedulerWorkerPool {
+class BASE_EXPORT PlatformNativeWorkerPoolWin
+    : public PlatformNativeWorkerPool {
  public:
   PlatformNativeWorkerPoolWin(TrackedRef<TaskTracker> task_tracker,
                               TrackedRef<Delegate> delegate);
 
-  // Destroying a PlatformNativeWorkerPoolWin is not allowed in
-  // production; it is always leaked. In tests, it can only be destroyed after
-  // JoinForTesting() has returned.
   ~PlatformNativeWorkerPoolWin() override;
 
-  // Starts the worker pool and allows tasks to begin running.
-  void Start();
-
-  // SchedulerWorkerPool:
-  void JoinForTesting() override;
-  size_t GetMaxConcurrentNonBlockedTasksDeprecated() const override;
-  void ReportHeartbeatMetrics() const override;
-
  private:
-  class ScopedWorkersExecutor;
-
-  // Callback that gets run by |pool_|. It runs a task off the next sequence on
-  // the |priority_queue_|.
+  // Callback that gets run by |pool_|.
   static void CALLBACK RunNextSequence(PTP_CALLBACK_INSTANCE,
                                        void* scheduler_worker_pool_windows_impl,
                                        PTP_WORK);
-  // SchedulerWorkerPool:
-  void UpdateSortKey(SequenceAndTransaction sequence_and_transaction) override;
-  void PushSequenceAndWakeUpWorkers(
-      SequenceAndTransaction sequence_and_transaction) override;
-  void EnsureEnoughWorkersLockRequired(BaseScopedWorkersExecutor* executor)
-      override EXCLUSIVE_LOCKS_REQUIRED(lock_);
 
-  // Returns the top Sequence off the |priority_queue_|. Returns nullptr
-  // if the |priority_queue_| is empty.
-  scoped_refptr<Sequence> GetWork();
+  // PlatformNativeWorkerPool:
+  void JoinImpl() override;
+  void StartImpl() override;
+  void SubmitWork() override;
 
   // Thread pool object that |work_| gets executed on.
   PTP_POOL pool_ = nullptr;
@@ -75,18 +55,6 @@ class BASE_EXPORT PlatformNativeWorkerPoolWin : public SchedulerWorkerPool {
   // it.
   PTP_WORK work_ = nullptr;
 
-  // Indicates whether the pool has been started yet.
-  bool started_ GUARDED_BY(lock_) = false;
-
-  // Number of threadpool work submitted to the pool which haven't popped a
-  // Sequence from the PriorityQueue yet.
-  size_t num_pending_threadpool_work_ GUARDED_BY(lock_) = 0;
-
-#if DCHECK_IS_ON()
-  // Set once JoinForTesting() has returned.
-  AtomicFlag join_for_testing_returned_;
-#endif
-
   DISALLOW_COPY_AND_ASSIGN(PlatformNativeWorkerPoolWin);
 };
 

base/task/task_scheduler/scheduler_worker_pool_unittest.cc

@@ -32,6 +32,8 @@
 
 #if defined(OS_WIN)
 #include "base/task/task_scheduler/platform_native_worker_pool_win.h"
+#elif defined(OS_MACOSX)
+#include "base/task/task_scheduler/platform_native_worker_pool_mac.h"
 #endif
 
 namespace base {
@@ -39,6 +41,15 @@ namespace internal {
 
 namespace {
 
+#if defined(OS_WIN) || defined(OS_MACOSX)
+using PlatformNativeWorkerPoolType =
+#if defined(OS_WIN)
+    PlatformNativeWorkerPoolWin;
+#elif defined(OS_MACOSX)
+    PlatformNativeWorkerPoolMac;
+#endif
+#endif
+
 constexpr size_t kMaxTasks = 4;
 // By default, tests allow half of the pool to be used by best-effort tasks.
 constexpr size_t kMaxBestEffortTasks = kMaxTasks / 2;
@@ -47,8 +58,8 @@ constexpr size_t kNumTasksPostedPerThread = 150;
 
 enum class PoolType {
   GENERIC,
-#if defined(OS_WIN)
-  WINDOWS,
+#if defined(OS_WIN) || defined(OS_MACOSX)
+  NATIVE,
 #endif
 };
 
@@ -123,9 +134,9 @@ class TaskSchedulerWorkerPoolTest
             task_tracker_.GetTrackedRef(),
             tracked_ref_factory_.GetTrackedRef());
         break;
-#if defined(OS_WIN)
-      case PoolType::WINDOWS:
-        worker_pool_ = std::make_unique<PlatformNativeWorkerPoolWin>(
+#if defined(OS_WIN) || defined(OS_MACOSX)
+      case PoolType::NATIVE:
+        worker_pool_ = std::make_unique<PlatformNativeWorkerPoolType>(
             task_tracker_.GetTrackedRef(),
             tracked_ref_factory_.GetTrackedRef());
         break;
@@ -148,11 +159,11 @@ class TaskSchedulerWorkerPoolTest
             SchedulerWorkerPoolImpl::WorkerEnvironment::NONE);
         break;
       }
-#if defined(OS_WIN)
-      case PoolType::WINDOWS: {
-        PlatformNativeWorkerPoolWin* scheduler_worker_pool_windows_impl =
-            static_cast<PlatformNativeWorkerPoolWin*>(worker_pool_.get());
-        scheduler_worker_pool_windows_impl->Start();
+#if defined(OS_WIN) || defined(OS_MACOSX)
+      case PoolType::NATIVE: {
+        PlatformNativeWorkerPoolType* scheduler_worker_pool_native_impl =
+            static_cast<PlatformNativeWorkerPoolType*>(worker_pool_.get());
+        scheduler_worker_pool_native_impl->Start();
         break;
       }
 #endif
@@ -411,16 +422,15 @@ INSTANTIATE_TEST_SUITE_P(GenericSequenced,
                              PoolType::GENERIC,
                              test::ExecutionMode::SEQUENCED}));
 
-#if defined(OS_WIN)
-INSTANTIATE_TEST_SUITE_P(WinParallel,
+#if defined(OS_WIN) || defined(OS_MACOSX)
+INSTANTIATE_TEST_SUITE_P(NativeParallel,
                          TaskSchedulerWorkerPoolTest,
                          ::testing::Values(PoolExecutionType{
-                             PoolType::WINDOWS,
-                             test::ExecutionMode::PARALLEL}));
-INSTANTIATE_TEST_SUITE_P(WinSequenced,
+                             PoolType::NATIVE, test::ExecutionMode::PARALLEL}));
+INSTANTIATE_TEST_SUITE_P(NativeSequenced,
                          TaskSchedulerWorkerPoolTest,
                          ::testing::Values(PoolExecutionType{
-                             PoolType::WINDOWS,
+                             PoolType::NATIVE,
                              test::ExecutionMode::SEQUENCED}));
 #endif