TaskScheduler [6] SchedulerWorkerThread

base/task_scheduler/worker_thread.h

+// Copyright 2016 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_SCHEDULER_WORKER_THREAD_H_
+#define BASE_TASK_SCHEDULER_WORKER_THREAD_H_
+
+#include "base/atomicops.h"
+#include "base/base_export.h"
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/task_scheduler/sequence.h"
+#include "base/threading/platform_thread.h"
+
+namespace base {
+namespace internal {
+
+class TaskTracker;
+
+// A thread that runs Tasks from Sequences returned by a callback.
+//
+// A WorkerThread is woken up when its WakeUp() method is called. After a wake-
+// up, a WorkerThread runs Tasks from Sequences returned by its "get work"
+// callback as long as it doesn't return nullptr. It also periodically checks
+// with its TaskTracker whether shutdown has completed and exits when it has.
+//
+// This class is thread-safe.
+class BASE_EXPORT WorkerThread : public PlatformThread::Delegate {
+ public:
+  // Callback invoked to get a Sequence from which to run a Task on
+  // |worker_thread|.
+  using GetWorkCallback =

[robliao, 2016/03/30 19:38:27]

At some point it might make sense to have a WorkerThreadDelegate interface
instead since these are going to be base::bind(...,
base::Unretained(ThreadPool)).

class WorkerThreadDelegate {
  scoped_refptr<Sequence> GetWork(const WorkerThread* worker_thread) = 0;
  void FinishedWork(const WorkerThread* worker_thread,
                    scoped_refptr<Sequence> sequence) = 0;
};

[fdoray, 2016/03/30 19:56:37]

In fact, the target of GetWorkCallback is a ThreadPool (shared+single-threaded
PriorityQueues will be owned by the WorkerThread's ThreadPool) and the target of
RanTaskFromSequenceCallback is the TaskScheduler (because a Sequence can be
reinserted in a PriorityQueue that belongs to another ThreadPool). I would have
used a delegate if multiple methods were called on the same target, but here I
prefer callbacks. WDYT?

[robliao, 2016/03/30 21:35:26]

sgtm.

+      Callback<scoped_refptr<Sequence>(const WorkerThread* worker_thread)>;
+
+  // Callback invoked after |worker_thread| has tried to run a Task from
+  // |sequence| (a TaskTracker might have prevented the Task from running).
+  using RanTaskFromSequenceCallback =
+      Callback<void(const WorkerThread* worker_thread,
+                    scoped_refptr<Sequence> sequence)>;
+
+  // Creates a WorkerThread with priority |thread_priority| that runs Tasks from
+  // Sequences returned by |get_work_callback|.
+  // |ran_task_from_sequence_callback| is invoked after the WorkerThread has
+  // tried to run a Task from a Sequence returned by |get_work_callback|.
+  // |task_tracker| is used to handle shutdown behavior of Tasks.
+  static scoped_ptr<WorkerThread> CreateWorkerThread(
+      ThreadPriority thread_priority,
+      const GetWorkCallback& get_work_callback,
+      const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+      TaskTracker* task_tracker);
+
+  // Destroying a WorkerThread in production is not allowed; it is always
+  // leaked. In tests, it can only be destroyed after JoinForTesting() has
+  // returned.
+  ~WorkerThread() override;
+
+  // Wakes up this WorkerThread. After this is called, this WorkerThread will
+  // run Tasks from Sequences returned by |get_work_callback_| until it returns
+  // nullptr.
+  void WakeUp();
+
+  // Joins this WorkerThread. If a Task is already running, it will be allowed
+  // to complete its execution. This can only be called once.
+  void JoinForTesting();
+
+ private:
+  WorkerThread(
+      ThreadPriority thread_priority,
+      const GetWorkCallback& get_work_callback,
+      const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+      TaskTracker* task_tracker);
+
+  // PlatformThread::Delegate:
+  void ThreadMain() override;
+
+  bool should_exit_for_testing() const {
+    base::subtle::MemoryBarrier();
+    return should_exit_for_testing_;
+  }
+
+  // Platform thread managed by this WorkerThread.
+  PlatformThreadHandle thread_handle_;
+
+  // Event signaled to wake up this WorkerThread.
+  WaitableEvent wake_up_event_;
+
+  const GetWorkCallback get_work_callback_;
+  const RanTaskFromSequenceCallback ran_task_from_sequence_callback_;
+  TaskTracker* const task_tracker_;
+
+  // True once JoinForTesting() has been called.
+  bool should_exit_for_testing_ = false;
+
+  DISALLOW_COPY_AND_ASSIGN(WorkerThread);
+};
+
+}  // namespace internal
+}  // namespace base
+
+#endif  // BASE_TASK_SCHEDULER_WORKER_THREAD_H_