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 <iosfwd>
+
+#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/single_thread_task_runner.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/task_scheduler/priority_queue.h"
+#include "base/task_scheduler/sequence.h"
+#include "base/task_scheduler/task_traits.h"
+#include "base/threading/platform_thread.h"
+
+namespace base {
+namespace internal {
+
+class TaskTracker;
+
+// A thread that runs Tasks from a shared and a single-threaded PriorityQueue.
+//
+// There is no guarantee that a WorkerThread will run work from its shared
+// PriorityQueue unless WakeUp() is called after the work is added. This allows
+// a ThreadPool to choose which WorkerThread should run the new work. On the
+// other hand, a WorkerThread is woken up automatically when Tasks are added to
+// its single-threaded PriorityQueue.
+//
+// This class is thread-safe.
+class BASE_EXPORT WorkerThread : public PlatformThread::Delegate {
+ public:
+  // Possible states for a WorkerThread.
+  enum class State {
+    BUSY,
+    IDLE,
+  };
+
+  // Callback invoked to indicate that a shared |sequence| is not empty after
+  // |worker_thread| has popped a Task from it.
+  using SharedSequenceStillHasTasksCallback =
+      Callback<void(const WorkerThread* worker_thread,
+                    scoped_refptr<Sequence> sequence)>;
+
+  // Callback invoked when the state of |worker_thread| becomes |state|.
+  using StateChangedCallback =
+      Callback<void(WorkerThread* worker_thread, State state)>;
+
+  // Creates a WorkerThread with priority |thread_priority| that runs Tasks from
+  // its single-threaded PriorityQueue and from |shared_priority_queue|.

[robliao, 2016/03/24 23:03:26]

Nit: |single_thread_priority_queue_| and from |shared_priority_queue_|

[fdoray, 2016/03/29 18:33:34]

The intent of this comment is to describe the arguments of CreateWorkerThread(),
not the private members of the class.

+  // |shared_sequence_still_has_tasks_callback| is invoked when a Sequence
+  // extracted from |shared_priority_queue| isn't empty after the WorkerThread
+  // has popped a task from it. |state_changed_callback| is invoked when the
+  // WorkerThread becomes IDLE or BUSY. |task_tracker| is used to handle
+  // shutdown behavior of Tasks.
+  static scoped_ptr<WorkerThread> CreateWorkerThread(
+      ThreadPriority thread_priority,
+      PriorityQueue* shared_priority_queue,
+      const SharedSequenceStillHasTasksCallback&
+          shared_sequence_still_has_tasks_callback,
+      const StateChangedCallback& state_changed_callback,
+      TaskTracker* task_tracker);
+
+  // Destroying a WorkerThread in production is not allowed. In tests, it can

[robliao, 2016/03/24 23:03:26]

How are we planning on reclaiming WorkerThreads in the future?

[fdoray, 2016/03/29 18:33:34]

Once a WorkerThread has been created in production, it cannot be reclaimed. A
ThreadPool creates a given number of WorkerThreads when it is created and can
rely on them forever.

We should compare the cost of (1) keeping threads alive vs. the cost of (2)
exiting and recreating threads on demand. If we determine that (2) is better, I
still prefer to have a list of WorkerThread that never go away in ThreadPool.
However, each WorkerThread could manage the lifetime of its platform thread
(e.g. the platform thread exits after 30 seconds of inactivity and is recreated
when WakeUp is called). I think that this solution will lead to simpler code
than a ramping up/down algorithm in ThreadPool. WDYT?

[robliao, 2016/03/30 00:42:42]

I think this will be hard to determine until we actually do (2). Since we're not
ramping up or down yet, we can stick with this, but it's certainly something to
keep in mind.

+  // only be destroyed after JoinForTesting() has returned.
+  ~WorkerThread() override;
+
+  // Wakes up this WorkerThread. When this is called, this WorkerThread will run
+  // until both its PriorityQueues become empty.
+  void WakeUp();
+
+  // Returns a SingleThreadTaskRunner whose PostTask invocations will result in
+  // scheduling Tasks with |traits| on this WorkerThread.
+  scoped_refptr<SingleThreadTaskRunner> CreateTaskRunnerWithTraits(
+      const TaskTraits& traits);
+
+  // 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,
+               PriorityQueue* shared_priority_queue,
+               const SharedSequenceStillHasTasksCallback&
+                   shared_sequence_still_has_tasks_callback,
+               const StateChangedCallback& state_changed_callback,
+               TaskTracker* task_tracker);
+
+  // Extracts the Sequence with the highest priority from
+  // |shared_priority_queue_| or |single_thread_priority_queue_|.
+  // |is_single_threaded| is set to true if the returned Sequence comes from
+  // |single_thread_priority_queue_|.
+  scoped_refptr<Sequence> GetWork(bool* is_single_threaded);
+
+  // Sets the state of this WorkerThread.
+  void SetState(State state);
+
+  // 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_;
+
+  // Current state of this WorkerThread.
+  State state_ = State::BUSY;
+
+  // Event signaled to wake up this WorkerThread.
+  WaitableEvent wake_up_event_;
+
+  // The single-threaded PriorityQueue from which this WorkerThread gets work.
+  PriorityQueue single_thread_priority_queue_;
+
+  // The shared PriorityQueue from which this WorkerThread gets work.
+  PriorityQueue* const shared_priority_queue_;
+
+  const SharedSequenceStillHasTasksCallback
+      shared_sequence_still_has_tasks_callback_;
+  const StateChangedCallback state_changed_callback_;
+  TaskTracker* const task_tracker_;
+
+  // True once JoinForTesting() has been called.
+  bool should_exit_for_testing_ = false;
+
+  DISALLOW_COPY_AND_ASSIGN(WorkerThread);
+};
+
+BASE_EXPORT std::ostream& operator<<(std::ostream& os,
+                                     WorkerThread::State state);
+
+}  // namespace internal
+}  // namespace base
+
+#endif  // BASE_TASK_SCHEDULER_WORKER_THREAD_H_